기존 파이프라인 업데이트

gcloud

gcloud dataflow jobs update-options 명령어를 사용합니다.

gcloud dataflow jobs update-options \
  --region=REGION \
  --min-num-workers=MINIMUM_WORKERS \
  --max-num-workers=MAXIMUM_WORKERS \
  --worker-utilization-hint=TARGET_UTILIZATION \
  JOB_ID

다음을 바꿉니다.

• REGION: 작업 리전 ID
• MINIMUM_WORKERS: 최소 Compute Engine 인스턴스 수
• MAXIMUM_WORKERS: 최대 Compute Engine 인스턴스 수
• TARGET_UTILIZATION: [0.1, 0.9] 범위 이내 값
• JOB_ID: 업데이트할 작업의 ID

--min-num-workers, --max-num-workers 및 worker-utilization-hint를 개별적으로 업데이트할 수도 있습니다.

REST

projects.locations.jobs.update 메서드를 사용합니다.

PUT https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/REGION/jobs/JOB_ID?updateMask=MASK
{
  "runtime_updatable_params": {
    "min_num_workers": MINIMUM_WORKERS,
    "max_num_workers": MAXIMUM_WORKERS,
    "worker_utilization_hint": TARGET_UTILIZATION
  }
}

다음을 바꿉니다.

• MASK: 다음과 같이 업데이트할 쉼표로 구분된 파라미터 목록입니다.
  • runtime_updatable_params.max_num_workers
  • runtime_updatable_params.min_num_workers
  • runtime_updatable_params.worker_utilization_hint
• PROJECT_ID: Dataflow 작업의 Google Cloud 프로젝트 ID
• REGION: 작업 리전 ID
• JOB_ID: 업데이트할 작업의 ID
• MINIMUM_WORKERS: 최소 Compute Engine 인스턴스 수
• MAXIMUM_WORKERS: 최대 Compute Engine 인스턴스 수
• TARGET_UTILIZATION: [0.1, 0.9] 범위 이내 값

min_num_workers, max_num_workers 및 worker_utilization_hint를 개별적으로 업데이트할 수도 있습니다. updateMask 쿼리 매개변수에서 업데이트할 매개변수를 지정하고 요청 본문의 runtimeUpdatableParams 필드에 업데이트된 값을 포함합니다. 다음 예시는 min_num_workers를 업데이트합니다.

PUT https://dataflow.googleapis.com/v1b3/projects/my_project/locations/us-central1/jobs/job1?updateMask=runtime_updatable_params.min_num_workers
{
  "runtime_updatable_params": {
    "min_num_workers": 5
  }
}

자바

• --update 옵션을 전달합니다.
• PipelineOptions의 --jobName 옵션을 업데이트할 작업과 동일한 이름으로 설정합니다.
• --region 옵션을 업데이트할 작업 리전과 동일한 리전으로 설정하세요.
• --dataflowServiceOptions=graph_validate_only 서비스 옵션을 포함합니다.
• 파이프라인의 변환 이름이 변경되면 변환 매핑을 제공하고 --transformNameMapping 옵션을 사용하여 전달해야 합니다.
• 이후 버전의 Apache Beam SDK를 사용하는 대체 작업을 제출하는 경우 --updateCompatibilityVersion을 원래 작업에서 사용된 Apache Beam SDK 버전으로 설정합니다.

Python

• --update 옵션을 전달합니다.
• PipelineOptions의 --job_name 옵션을 업데이트할 작업과 동일한 이름으로 설정합니다.
• --region 옵션을 업데이트할 작업 리전과 동일한 리전으로 설정하세요.
• --dataflow_service_options=graph_validate_only 서비스 옵션을 포함합니다.
• 파이프라인의 변환 이름이 변경되면 변환 매핑을 제공하고 --transform_name_mapping 옵션을 사용하여 전달해야 합니다.
• 이후 버전의 Apache Beam SDK를 사용하는 대체 작업을 제출하는 경우 --updateCompatibilityVersion을 원래 작업에서 사용된 Apache Beam SDK 버전으로 설정합니다.

Go

• --update 옵션을 전달합니다.
• --job_name 옵션을 업데이트할 작업과 동일한 이름으로 설정합니다.
• --region 옵션을 업데이트할 작업 리전과 동일한 리전으로 설정하세요.
• --dataflow_service_options=graph_validate_only 서비스 옵션을 포함합니다.
• 파이프라인의 변환 이름이 변경되면 변환 매핑을 제공하고 --transform_name_mapping 옵션을 사용하여 전달해야 합니다.

gcloud

Flex 템플릿 작업의 작업 그래프를 검증하려면 additional-experiments 옵션과 함께 gcloud dataflow flex-template run 명령어를 사용합니다.

• --update 옵션을 전달합니다.
• JOB_NAME을(를) 업데이트할 작업과 동일한 이름으로 설정합니다.
• --region 옵션을 업데이트할 작업 리전과 동일한 리전으로 설정하세요.
• --additional-experiments=graph_validate_only 옵션을 포함합니다.
• 파이프라인의 변환 이름이 변경되면 변환 매핑을 제공하고 --transform-name-mappings 옵션을 사용하여 전달해야 합니다.

예를 들면 다음과 같습니다.

gcloud dataflow flex-template run JOB_NAME --additional-experiments=graph_validate_only

JOB_NAME을 업데이트할 작업의 이름으로 바꿉니다.

REST

FlexTemplateRuntimeEnvironment(Flex 템플릿) 또는 RuntimeEnvironment 객체에서 additionalExperiments 필드를 사용합니다.

{
  additionalExperiments : ["graph_validate_only"]
  ...
}

자바

• --update 옵션을 전달합니다.
• PipelineOptions의 --jobName 옵션을 업데이트할 작업과 동일한 이름으로 설정합니다.
• --region 옵션을 업데이트할 작업 리전과 동일한 리전으로 설정하세요.
• 파이프라인의 변환 이름이 변경되면 변환 매핑을 제공하고 --transformNameMapping 옵션을 사용하여 전달해야 합니다.
• 이후 버전의 Apache Beam SDK를 사용하는 대체 작업을 제출하는 경우 --updateCompatibilityVersion을 원래 작업에서 사용된 Apache Beam SDK 버전으로 설정합니다.

Python

• --update 옵션을 전달합니다.
• PipelineOptions의 --job_name 옵션을 업데이트할 작업과 동일한 이름으로 설정합니다.
• --region 옵션을 업데이트할 작업 리전과 동일한 리전으로 설정하세요.
• 파이프라인의 변환 이름이 변경되면 변환 매핑을 제공하고 --transform_name_mapping 옵션을 사용하여 전달해야 합니다.
• 이후 버전의 Apache Beam SDK를 사용하는 대체 작업을 제출하는 경우 --updateCompatibilityVersion을 원래 작업에서 사용된 Apache Beam SDK 버전으로 설정합니다.

Go

• --update 옵션을 전달합니다.
• --job_name 옵션을 업데이트할 작업과 동일한 이름으로 설정합니다.
• --region 옵션을 업데이트할 작업 리전과 동일한 리전으로 설정하세요.
• 파이프라인의 변환 이름이 변경되면 변환 매핑을 제공하고 --transform_name_mapping 옵션을 사용하여 전달해야 합니다.

gcloud

gcloud CLI를 사용하여 Flex 템플릿 작업을 업데이트하려면 gcloud dataflow flex-template run 명령어를 사용합니다. gcloud CLI를 사용한 다른 작업 업데이트는 지원되지 않습니다.

• --update 옵션을 전달합니다.
• JOB_NAME을(를) 업데이트할 작업과 동일한 이름으로 설정합니다.
• --region 옵션을 업데이트할 작업 리전과 동일한 리전으로 설정하세요.
• 파이프라인의 변환 이름이 변경되면 변환 매핑을 제공하고 --transform-name-mappings 옵션을 사용하여 전달해야 합니다.

REST

다음 안내에서는 REST API를 사용하여 템플릿이 아닌 작업을 업데이트하는 방법을 보여줍니다. REST API를 사용하여 기본 템플릿 작업을 업데이트하려면 커스텀 템플릿 스트리밍 작업 업데이트를 참조하세요. REST API를 사용하여 Flex 템플릿 작업을 업데이트하려면 Flex 템플릿 작업 업데이트를 참조하세요.

1. projects.locations.jobs.get 메서드를 사용하여 교체하려는 작업의 job 리소스를 가져옵니다. view 쿼리 매개변수에 JOB_VIEW_DESCRIPTION 값을 포함합니다. JOB_VIEW_DESCRIPTION을 포함하면 후속 요청이 크기 한도를 초과하지 않도록 응답에 포함할 데이터 양이 제한됩니다. 자세한 작업 정보가 필요하면 JOB_VIEW_ALL 값을 사용합니다.

   GET https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/REGION/jobs/JOB_ID?view=JOB_VIEW_DESCRIPTION
   

   다음 값을 바꿉니다.

   • PROJECT_ID: Dataflow 작업의 Google Cloud 프로젝트 ID
   • REGION: 업데이트할 작업의 리전입니다.
   • JOB_ID: 업데이트할 작업의 작업 ID입니다.

2. 작업을 업데이트하려면 projects.locations.jobs.create 메서드를 사용합니다. 요청 본문에서 가져온 job 리소스를 사용합니다.

   POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/REGION/jobs
   {
     "id": JOB_ID,
     "replaceJobId": JOB_ID,
     "name": JOB_NAME,
     "type": "JOB_TYPE_STREAMING",
     "transformNameMapping": {
       string: string,
       ...
     },
   }
   

   다음을 바꿉니다.

   • JOB_ID: 업데이트할 작업의 ID와 동일한 작업 ID입니다.
   • JOB_NAME: 업데이트할 작업 이름과 동일한 작업 이름입니다.

   파이프라인의 변환 이름이 변경되면 변환 매핑을 제공하고 transformNameMapping 필드를 사용하여 전달해야 합니다.

3. 선택사항: curl(Linux, macOS 또는 Cloud Shell)을 사용하여 요청을 보내려면 요청을 JSON 파일에 저장한 후 다음 명령어를 실행합니다.

   curl -X POST -d "@FILE_PATH" -H "Content-Type: application/json" -H "Authorization: Bearer $(gcloud auth print-access-token)"  https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/REGION/jobs
   

   FILE_PATH를 요청 본문이 포함된 JSON 파일의 경로로 바꿉니다.

자바

교체 작업 실행 시 --jobName 옵션과 관련하여 전달하는 값은 교체할 작업 이름과 정확히 일치해야 합니다.

Python

교체 작업 실행 시 --job_name 옵션과 관련하여 전달하는 값은 교체할 작업 이름과 정확히 일치해야 합니다.

Go

교체 작업 실행 시 --job_name 옵션과 관련하여 전달하는 값은 교체할 작업 이름과 정확히 일치해야 합니다.

gcloud

교체 작업 실행 시 JOB_NAME은 교체할 작업 이름과 정확히 일치해야 합니다.

REST

replaceJobId 필드 값을 업데이트할 작업과 동일한 작업 ID로 설정합니다. 올바른 작업 이름 값을 찾으려면 Dataflow 모니터링 인터페이스에서 이전 작업을 선택합니다. 그런 다음 작업 정보 측면 패널에서 작업 ID 필드를 찾습니다.

자바

--transformNameMapping 명령줄 옵션을 사용하여 매핑을 전달할 때 다음의 일반적인 형식을 사용합니다.

--transformNameMapping= . 
{"oldTransform1":"newTransform1","oldTransform2":"newTransform2",...}

이전 파이프라인과 교체 파이프라인 사이에 변경된 변환 이름에 대해서만 --transformNameMapping에 매핑 항목을 제공해야 합니다.

--transformNameMapping을 사용하여 실행할 때 따옴표를 셸에 맞게 이스케이프 처리해야 할 수 있습니다. 예를 들어 Bash에서는 다음과 같습니다.

--transformNameMapping='{"oldTransform1":"newTransform1",...}'

Python

--transform_name_mapping 명령줄 옵션을 사용하여 매핑을 전달할 때 다음의 일반적인 형식을 사용합니다.

--transform_name_mapping= .
{"oldTransform1":"newTransform1","oldTransform2":"newTransform2",...}

이전 파이프라인과 교체 파이프라인 사이에 변경된 변환 이름에 대해서만 --transform_name_mapping에 매핑 항목을 제공해야 합니다.

--transform_name_mapping을 사용하여 실행할 때 따옴표를 셸에 맞게 이스케이프 처리해야 할 수 있습니다. 예를 들어 Bash에서는 다음과 같습니다.

--transform_name_mapping='{"oldTransform1":"newTransform1",...}'

Go

--transform_name_mapping 명령줄 옵션을 사용하여 매핑을 전달할 때 다음의 일반적인 형식을 사용합니다.

--transform_name_mapping= .
{"oldTransform1":"newTransform1","oldTransform2":"newTransform2",...}

이전 파이프라인과 교체 파이프라인 사이에 변경된 변환 이름에 대해서만 --transform_name_mapping에 매핑 항목을 제공해야 합니다.

--transform_name_mapping을 사용하여 실행할 때 따옴표를 셸에 맞게 이스케이프 처리해야 할 수 있습니다. 예를 들어 Bash에서는 다음과 같습니다.

--transform_name_mapping='{"oldTransform1":"newTransform1",...}'

gcloud

--transform-name-mappings 옵션을 사용하여 매핑을 전달할 때 다음의 일반적인 형식을 사용합니다.

--transform-name-mappings= .
{"oldTransform1":"newTransform1","oldTransform2":"newTransform2",...}

이전 파이프라인과 교체 파이프라인 사이에 변경된 변환 이름에 대해서만 --transform-name-mappings에 매핑 항목을 제공해야 합니다.

--transform-name-mappings을 사용하여 실행할 때 따옴표를 셸에 맞게 이스케이프 처리해야 할 수 있습니다. 예를 들어 Bash에서는 다음과 같습니다.

--transform-name-mappings='{"oldTransform1":"newTransform1",...}'

REST

transformNameMapping 필드를 사용하여 매핑을 전달할 때 다음의 일반적인 형식을 사용합니다.

"transformNameMapping": {
  oldTransform1: newTransform1,
  oldTransform2: newTransform2,
  ...
}

이전 파이프라인과 교체 파이프라인 사이에 변경된 변환 이름에 대해서만 transformNameMapping에 매핑 항목을 제공해야 합니다.

자바

  .apply("FormatResults", ParDo
    .of(new DoFn<KV<String, Long>>, String>() {
      ...
     }
  }))

Python

  | 'FormatResults' >> beam.ParDo(MyDoFn())

Go

  // In Go, this is always the package-qualified name of the DoFn itself.
  // For example, if the FormatResults DoFn is in the main package, its name
  // is "main.FormatResults".
  beam.ParDo(s, FormatResults, results)

자바

--transformNameMapping={"CountWidgets/Parse":"CountWidgets/Scan/Parse"}

교체 파이프라인에서 변환을 완전히 삭제하는 경우, null 매핑을 제공해야 합니다. 교체 파이프라인이 CountWidgets/Parse 변환을 완전히 제거한다고 가정합니다.

--transformNameMapping={"CountWidgets/Parse":""}

Python

--transform_name_mapping={"CountWidgets/Parse":"CountWidgets/Scan/Parse"}

교체 파이프라인에서 변환을 완전히 삭제하는 경우, null 매핑을 제공해야 합니다. 교체 파이프라인이 CountWidgets/Parse 변환을 완전히 제거한다고 가정합니다.

--transform_name_mapping={"CountWidgets/Parse":""}

Go

--transform_name_mapping={"CountWidgets/main.Parse":"CountWidgets/Scan/main.Parse"}

교체 파이프라인에서 변환을 완전히 삭제하는 경우, null 매핑을 제공해야 합니다. 교체 파이프라인이 CountWidgets/Parse 변환을 완전히 제거한다고 가정합니다.

--transform_name_mapping={"CountWidgets/main.Parse":""}

gcloud

--transform-name-mappings={"CountWidgets/Parse":"CountWidgets/Scan/Parse"}

교체 파이프라인에서 변환을 완전히 삭제하는 경우, null 매핑을 제공해야 합니다. 교체 파이프라인이 CountWidgets/Parse 변환을 완전히 제거한다고 가정합니다.

--transform-name-mappings={"CountWidgets/main.Parse":""}

REST

"transformNameMapping": {
  CountWidgets/Parse: CountWidgets/Scan/Parse
}

교체 파이프라인에서 변환을 완전히 삭제하는 경우, null 매핑을 제공해야 합니다. 교체 파이프라인이 CountWidgets/Parse 변환을 완전히 제거한다고 가정합니다.

"transformNameMapping": {
  CountWidgets/main.Parse: null
}

Java

  p.apply("Read", ReadStrings())
   .apply("Format", FormatStrings());

Python

  p | 'Read' >> beam.io.ReadFromPubSub(subscription=known_args.input_subscription)
    | 'Format' >> FormatStrings()

Go

   beam.ParDo(s, ReadStrings)
   beam.ParDo(s, FormatStrings)

Java

  p.apply("Read", ReadStrings())
   .apply("Remove", RemoveStringsStartingWithA())
   .apply("Format", FormatStrings());

Python

  p | 'Read' >> beam.io.ReadFromPubSub(subscription=known_args.input_subscription)
    | 'Remove' >> RemoveStringsStartingWithA()
    | 'Format' >> FormatStrings()

Go

  beam.ParDo(s, ReadStrings)
  beam.ParDo(s, RemoveStringsStartingWithA)
  beam.ParDo(s, FormatStrings)

자바

제한사항으로 인해 콘솔 또는 터미널에서 실패한 업데이트 시도 오류를 보려면 실행 차단을 사용해야 합니다. 현재 해결 방법은 다음 단계로 구성됩니다.

1. 파이프라인 코드에 pipeline.run().waitUntilFinish()를 사용합니다.
2. --update 옵션을 사용하여 교체 파이프라인 프로그램을 실행합니다.
3. 교체 작업이 호환성 검사를 통과할 때까지 기다립니다.
4. Ctrl+C를 입력하여 차단 실행자 프로세스를 종료합니다.

또는 Dataflow 모니터링 인터페이스에서 교체 작업 상태를 모니터링할 수 있습니다. 작업이 성공적으로 시작되었으면 호환성 검사를 통과한 것입니다.

Python

제한사항으로 인해 콘솔 또는 터미널에서 실패한 업데이트 시도 오류를 보려면 실행 차단을 사용해야 합니다. 현재 해결 방법은 다음 단계로 구성됩니다.

1. 파이프라인 코드에 pipeline.run().wait_until_finish()를 사용합니다.
2. --update 옵션을 사용하여 교체 파이프라인 프로그램을 실행합니다.
3. 교체 작업이 호환성 검사를 통과할 때까지 기다립니다.
4. Ctrl+C를 입력하여 차단 실행자 프로세스를 종료합니다.

또는 Dataflow 모니터링 인터페이스에서 교체 작업 상태를 모니터링할 수 있습니다. 작업이 성공적으로 시작되었으면 호환성 검사를 통과한 것입니다.

Go

제한사항으로 인해 콘솔 또는 터미널에서 실패한 업데이트 시도 오류를 보려면 실행 차단을 사용해야 합니다. 특히 --execute_async 또는 --async 플래그를 사용하여 비차단 실행을 지정해야 합니다. 현재 해결 방법은 다음 단계로 구성됩니다.

1. --update 옵션을 사용하고 --execute_async 또는 --async 플래그 없이 교체 파이프라인 프로그램을 실행합니다.
2. 교체 작업이 호환성 검사를 통과할 때까지 기다립니다.
3. Ctrl+C를 입력하여 차단 실행자 프로세스를 종료합니다.

gcloud

제한사항으로 인해 콘솔 또는 터미널에서 실패한 업데이트 시도 오류를 보려면 실행 차단을 사용해야 합니다. 현재 해결 방법은 다음 단계로 구성됩니다.

1. Java 파이프라인의 경우 파이프라인 코드에 pipeline.run().waitUntilFinish()를 사용합니다. Python 파이프라인의 경우 파이프라인 코드에 pipeline.run().wait_until_finish()를 사용합니다. Go 파이프라인의 경우 Go 탭의 단계를 따르세요.
2. --update 옵션을 사용하여 교체 파이프라인 프로그램을 실행합니다.
3. 교체 작업이 호환성 검사를 통과할 때까지 기다립니다.
4. Ctrl+C를 입력하여 차단 실행자 프로세스를 종료합니다.

REST

제한사항으로 인해 콘솔 또는 터미널에서 실패한 업데이트 시도 오류를 보려면 실행 차단을 사용해야 합니다. 현재 해결 방법은 다음 단계로 구성됩니다.

• Java 파이프라인의 경우 파이프라인 코드에 pipeline.run().waitUntilFinish()를 사용합니다. Python 파이프라인의 경우 파이프라인 코드에 pipeline.run().wait_until_finish()를 사용합니다. Go 파이프라인의 경우 Go 탭의 단계를 따르세요.
• replaceJobId 필드를 사용하여 교체 파이프라인 프로그램을 실행합니다.
• 교체 작업이 호환성 검사를 통과할 때까지 기다립니다.
• Ctrl+C를 입력하여 차단 실행자 프로세스를 종료합니다.