Google menggunakan teknologi AI untuk menerjemahkan konten ke dalam bahasa pilihan Anda. Terjemahan AI mungkin mengandung kesalahan.

Memperbarui pipeline yang ada

Dokumen ini menjelaskan cara mengupdate tugas streaming yang sedang berlangsung. Anda mungkin ingin mengupdate tugas Dataflow yang ada karena alasan berikut:

Anda ingin menyempurnakan atau meningkatkan kode pipeline Anda.
Anda ingin memperbaiki bug dalam kode pipeline.
Anda ingin memperbarui pipeline untuk menangani perubahan format data, atau untuk memperhitungkan versi atau perubahan lain di sumber data Anda.
Anda ingin melakukan patch pada kerentanan keamanan yang terkait dengan Container-Optimized OS untuk semua worker Dataflow.
Anda ingin menskalakan pipeline Apache Beam streaming untuk menggunakan jumlah worker yang berbeda.

Anda dapat memperbarui tugas dengan dua cara:

Pembaruan tugas saat beroperasi: Untuk tugas streaming yang menggunakan Streaming Engine, Anda dapat memperbarui opsi tugas min-num-workers dan max-num-workers tanpa menghentikan tugas atau mengubah ID pekerjaan.
Tugas pengganti: Untuk menjalankan kode pipeline yang diperbarui atau untuk memperbarui opsi tugas yang tidak didukung oleh update tugas yang sedang berjalan, luncurkan tugas baru yang menggantikan tugas yang ada. Untuk memverifikasi apakah tugas pengganti valid, sebelum meluncurkan tugas baru, validasi grafik tugasnya.

Saat Anda mengupdate tugas, layanan Dataflow akan melakukan pemeriksaan kompatibilitas antara tugas yang sudah berjalan dan potensi tugas pengganti. Pemeriksaan kompatibilitas memastikan bahwa hal-hal seperti informasi status perantara dan data yang di-buffer dapat ditransfer dari tugas sebelumnya ke tugas pengganti.

Anda juga dapat menggunakan infrastruktur logging bawaan Apache Beam SDK untuk mencatat informasi ke dalam log saat Anda mengupdate tugas. Untuk mengetahui informasi selengkapnya, lihat Menggunakan log pipeline. Untuk mengidentifikasi masalah pada kode pipeline, gunakan level logging DEBUG.

Untuk mengetahui petunjuk cara mengupdate tugas streaming yang menggunakan template klasik, lihat Mengupdate tugas streaming template kustom.
Untuk mengetahui petunjuk cara mengupdate tugas streaming yang menggunakan Template Flex, ikuti petunjuk gcloud CLI di halaman ini, atau lihat Mengupdate tugas Template Flex.

Pembaruan opsi tugas yang sedang berjalan

Untuk tugas streaming yang menggunakan Streaming Engine, Anda dapat memperbarui opsi tugas berikut tanpa menghentikan tugas atau mengubah ID pekerjaan:

min-num-workers: jumlah minimum instance Compute Engine.
max-num-workers: jumlah maksimum instance Compute Engine.
worker-utilization-hint: target pemakaian CPU, dalam rentang [0,1, 0,9]
latency-tier: tingkat latensi, yang menetapkan batas backlog. Nilai: low_latency (default, kurang dari 15 detik), medium_latency (kurang dari 2 menit), atau high_latency (kurang dari 10 menit).

Untuk update tugas lainnya, Anda harus mengganti tugas saat ini dengan tugas yang telah diupdate. Untuk mengetahui informasi selengkapnya, lihat Meluncurkan tugas pengganti.

Melakukan update dalam penerbangan

Untuk melakukan update opsi tugas yang sedang berjalan, lakukan langkah-langkah berikut.

gcloud

Gunakan perintah 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 \
  --latency-tier=TIER \
  JOB_ID

Ganti kode berikut:

REGION: ID region tugas
MINIMUM_WORKERS: jumlah minimum instance Compute Engine
MAXIMUM_WORKERS: jumlah maksimum instance Compute Engine
TARGET_UTILIZATION: nilai dalam rentang [0,1, 0,9]
TIER: tingkat latensi target. Salah satu dari berikut ini: low_latency, medium_latency, atau high_latency
JOB_ID: ID tugas yang akan diupdate

Anda juga dapat memperbarui --min-num-workers, --max-num-workers, --worker-utilization-hint, atau --latency-tier satu per satu.

REST

Gunakan metode 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,
    "latency_tier": TIER
  }
}

Ganti kode berikut:

MASK: daftar parameter yang dipisahkan koma untuk diupdate, dari berikut ini:
- runtime_updatable_params.max_num_workers
- runtime_updatable_params.min_num_workers
- runtime_updatable_params.worker_utilization_hint
PROJECT_ID: Project ID Google Cloud dari tugas Dataflow
REGION: ID region tugas
JOB_ID: ID tugas yang akan diupdate
MINIMUM_WORKERS: jumlah minimum instance Compute Engine
MAXIMUM_WORKERS: jumlah maksimum instance Compute Engine
TARGET_UTILIZATION: nilai dalam rentang [0,1, 0,9]
TIER: tingkat latensi target. Salah satu dari berikut ini: low_latency, medium_latency, atau high_latency

Anda juga dapat memperbarui min_num_workers, max_num_workers, worker_utilization_hint, atau latency_tier satu per satu. Tentukan parameter yang akan diperbarui dalam parameter kueri updateMask, dan sertakan nilai yang diperbarui di kolom runtimeUpdatableParams pada isi permintaan. Contoh berikut memperbarui 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
  }
}

Tugas harus dalam status berjalan agar memenuhi syarat untuk update dalam penerbangan. Error akan terjadi jika tugas belum dimulai atau sudah dibatalkan. Demikian pula, jika Anda meluncurkan tugas pengganti, tunggu hingga tugas tersebut mulai berjalan sebelum mengirimkan update dalam proses ke tugas baru.

Setelah Anda mengirimkan permintaan update, sebaiknya tunggu hingga permintaan tersebut selesai sebelum mengirimkan update lain. Lihat log tugas untuk melihat kapan permintaan selesai.

Memvalidasi tugas penggantian

Untuk memverifikasi apakah tugas pengganti valid, sebelum Anda meluncurkan tugas baru, validasi grafik tugasnya. Di Dataflow, grafik tugas adalah representasi grafis dari pipeline. Dengan memvalidasi grafik tugas, Anda mengurangi risiko pipeline mengalami error atau kegagalan pipeline setelah update. Selain itu, Anda dapat memvalidasi update tanpa perlu menghentikan tugas asli, sehingga tugas tersebut tidak mengalami periode nonaktif.

Untuk memvalidasi grafik tugas, ikuti langkah-langkah untuk meluncurkan tugas pengganti. Sertakan opsi layanan Dataflow graph_validate_only dalam perintah update.

Java

Teruskan opsi --update.
Tetapkan opsi --jobName di PipelineOptions ke nama yang sama dengan tugas yang ingin Anda perbarui.
Tetapkan opsi --region ke region yang sama dengan region tugas yang ingin Anda update.
Sertakan opsi layanan --dataflowServiceOptions=graph_validate_only.
Jika ada nama transformasi di pipeline Anda yang berubah, Anda harus menyediakan pemetaan transformasi dan meneruskannya menggunakan opsi --transformNameMapping.
Jika Anda mengirimkan tugas pengganti yang menggunakan Apache Beam SDK versi yang lebih baru, tetapkan --updateCompatibilityVersion ke versi Apache Beam SDK yang digunakan dalam tugas asli.

Python

Teruskan opsi --update.
Tetapkan opsi --job_name di PipelineOptions ke nama yang sama dengan tugas yang ingin Anda perbarui.
Tetapkan opsi --region ke region yang sama dengan region tugas yang ingin Anda update.
Sertakan opsi layanan --dataflow_service_options=graph_validate_only.
Jika ada nama transformasi di pipeline Anda yang berubah, Anda harus menyediakan pemetaan transformasi dan meneruskannya menggunakan opsi --transform_name_mapping.
Jika Anda mengirimkan tugas pengganti yang menggunakan Apache Beam SDK versi yang lebih baru, tetapkan --updateCompatibilityVersion ke versi Apache Beam SDK yang digunakan dalam tugas asli.

Go

Teruskan opsi --update.
Tetapkan opsi --job_name ke nama yang sama dengan tugas yang ingin Anda perbarui.
Tetapkan opsi --region ke region yang sama dengan region tugas yang ingin Anda update.
Sertakan opsi layanan --dataflow_service_options=graph_validate_only.
Jika ada nama transformasi di pipeline Anda yang berubah, Anda harus menyediakan pemetaan transformasi dan meneruskannya menggunakan opsi --transform_name_mapping.

gcloud

Untuk memvalidasi grafik tugas untuk tugas Template Flex, gunakan perintah gcloud dataflow flex-template run dengan opsi additional-experiments:

Teruskan opsi --update.
Tetapkan JOB_NAME ke nama yang sama dengan tugas yang ingin Anda perbarui.
Tetapkan opsi --region ke region yang sama dengan region tugas yang ingin Anda update.
Sertakan opsi --additional-experiments=graph_validate_only.
Jika ada nama transformasi di pipeline Anda yang berubah, Anda harus menyediakan pemetaan transformasi dan meneruskannya menggunakan opsi --transform-name-mappings.

Contoh:

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

Ganti JOB_NAME dengan nama tugas yang ingin Anda perbarui.

REST

Gunakan kolom additionalExperiments di objek FlexTemplateRuntimeEnvironment (Template Flex) atau RuntimeEnvironment.

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

Opsi layanan graph_validate_only hanya memvalidasi update pipeline. Jangan gunakan opsi ini saat membuat atau meluncurkan pipeline. Untuk mengupdate pipeline, luncurkan tugas pengganti tanpa opsi layanan graph_validate_only.

Jika validasi grafik tugas berhasil, status tugas dan log tugas akan menampilkan status berikut:

Status tugas adalah JOB_STATE_DONE.
Di konsol Google Cloud , Status tugas adalah Succeeded.

Pesan berikut akan muncul di log tugas:

Workflow job: JOB_ID succeeded validation. Marking graph_validate_only job as Done.

Jika validasi grafik tugas gagal, status tugas dan log tugas akan menampilkan status berikut:

Status tugas adalah JOB_STATE_FAILED.
Di konsol Google Cloud , Status tugas adalah Failed.
Pesan akan muncul di log tugas yang menjelaskan error ketidaksesuaian. Konten pesan bergantung pada error.

Meluncurkan tugas penggantian

Anda dapat mengganti tugas yang ada karena alasan berikut:

Untuk menjalankan kode pipeline yang diperbarui.
Untuk memperbarui opsi tugas yang tidak mendukung update dalam proses.

Untuk memverifikasi apakah tugas penggantian valid, sebelum meluncurkan tugas baru, validasi grafik tugasnya.

Saat Anda meluncurkan tugas pengganti, tetapkan opsi pipeline berikut untuk melakukan proses update selain opsi reguler tugas:

Java

Teruskan opsi --update.
Tetapkan opsi --jobName di PipelineOptions ke nama yang sama dengan tugas yang ingin Anda perbarui.
Tetapkan opsi --region ke region yang sama dengan region tugas yang ingin Anda update.
Jika ada nama transformasi di pipeline Anda yang berubah, Anda harus menyediakan pemetaan transformasi dan meneruskannya menggunakan opsi --transformNameMapping.
Jika Anda mengirimkan tugas pengganti yang menggunakan Apache Beam SDK versi yang lebih baru, tetapkan --updateCompatibilityVersion ke versi Apache Beam SDK yang digunakan dalam tugas asli.

Python

Teruskan opsi --update.
Tetapkan opsi --job_name di PipelineOptions ke nama yang sama dengan tugas yang ingin Anda perbarui.
Tetapkan opsi --region ke region yang sama dengan region tugas yang ingin Anda update.
Jika ada nama transformasi di pipeline Anda yang berubah, Anda harus menyediakan pemetaan transformasi dan meneruskannya menggunakan opsi --transform_name_mapping.
Jika Anda mengirimkan tugas pengganti yang menggunakan Apache Beam SDK versi yang lebih baru, tetapkan --updateCompatibilityVersion ke versi Apache Beam SDK yang digunakan dalam tugas asli.

Go

Teruskan opsi --update.
Tetapkan opsi --job_name ke nama yang sama dengan tugas yang ingin Anda perbarui.
Tetapkan opsi --region ke region yang sama dengan region tugas yang ingin Anda update.
Jika ada nama transformasi di pipeline Anda yang berubah, Anda harus menyediakan pemetaan transformasi dan meneruskannya menggunakan opsi --transform_name_mapping.

gcloud

Untuk mengupdate tugas Template Fleksibel menggunakan gcloud CLI, gunakan perintah gcloud dataflow flex-template run. Mengupdate tugas lain menggunakan gcloud CLI tidak didukung.

Teruskan opsi --update.
Tetapkan JOB_NAME ke nama yang sama dengan tugas yang ingin Anda perbarui.
Tetapkan opsi --region ke region yang sama dengan region tugas yang ingin Anda update.
Jika ada nama transformasi di pipeline Anda yang berubah, Anda harus menyediakan pemetaan transformasi dan meneruskannya menggunakan opsi --transform-name-mappings.

REST

Petunjuk ini menunjukkan cara mengupdate tugas non-template menggunakan REST API. Untuk menggunakan REST API guna mengupdate tugas template klasik, lihat Mengupdate tugas streaming template kustom. Untuk menggunakan REST API guna mengupdate tugas Template Flex, lihat Mengupdate tugas Template Flex.

Ambil resource job untuk tugas yang ingin Anda ganti menggunakan metode projects.locations.jobs.get. Sertakan parameter kueri view dengan nilai JOB_VIEW_DESCRIPTION. Menyertakan JOB_VIEW_DESCRIPTION akan membatasi jumlah data dalam respons sehingga permintaan berikutnya tidak melebihi batas ukuran. Jika Anda memerlukan informasi tugas yang lebih mendetail, gunakan nilai JOB_VIEW_ALL.
```
GET https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/REGION/jobs/JOB_ID?view=JOB_VIEW_DESCRIPTION
```
Ganti nilai berikut:
- PROJECT_ID: Project ID Google Cloud dari tugas Dataflow
- REGION: region tugas yang ingin Anda update
- JOB_ID: ID tugas yang ingin Anda perbarui
Untuk memperbarui tugas, gunakan metode projects.locations.jobs.create. Dalam isi permintaan, gunakan resource job yang Anda ambil.
```
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,
    ...
  },
}
```
Ganti kode berikut:
- JOB_ID: ID pekerjaan yang sama dengan ID pekerjaan yang ingin Anda update.
- JOB_NAME: nama tugas yang sama dengan nama tugas yang ingin Anda perbarui.
Jika ada nama transformasi di pipeline Anda yang berubah, Anda harus menyediakan pemetaan transformasi dan meneruskannya menggunakan kolom transformNameMapping.

Opsional: Untuk mengirim permintaan menggunakan curl (Linux, macOS, atau Cloud Shell), simpan permintaan ke file JSON, lalu jalankan perintah berikut:

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

Ganti FILE_PATH dengan jalur ke file JSON yang berisi isi permintaan.

Menentukan nama tugas pengganti

Java

Saat meluncurkan tugas pengganti, nilai yang Anda teruskan untuk opsi --jobName harus sama persis dengan nama tugas yang ingin Anda ganti.

Python

Saat meluncurkan tugas pengganti, nilai yang Anda teruskan untuk opsi --job_name harus sama persis dengan nama tugas yang ingin Anda ganti.

Go

Saat meluncurkan tugas pengganti, nilai yang Anda teruskan untuk opsi --job_name harus sama persis dengan nama tugas yang ingin Anda ganti.

gcloud

Saat Anda meluncurkan tugas pengganti, JOB_NAME harus sama persis dengan nama tugas yang ingin Anda ganti.

REST

Tetapkan nilai kolom replaceJobId ke ID pekerjaan yang sama dengan pekerjaan yang ingin Anda perbarui. Untuk menemukan nilai nama tugas yang benar, pilih tugas Anda sebelumnya di Dataflow Monitoring Interface. Kemudian, di panel samping Job info, temukan kolom Job ID.

Untuk menemukan nilai nama tugas yang benar, pilih tugas Anda sebelumnya di Dataflow Monitoring Interface. Kemudian, di panel samping Job info, temukan kolom Job name:

Panel samping Job info untuk tugas Dataflow yang sedang berjalan. — Gambar 1: Panel samping Job info untuk tugas Dataflow yang sedang berjalan dengan kolom Job name.

Atau, buat kueri daftar tugas yang ada menggunakan Antarmuka Command-line Dataflow. Masukkan perintah gcloud dataflow jobs list ke jendela shell atau terminal Anda untuk mendapatkan daftar tugas Dataflow di project Google CloudAnda, dan temukan kolom NAME untuk tugas yang ingin Anda ganti:

JOB_ID                                    NAME                        TYPE       CREATION_TIME        STATE    REGION
2020-12-28_12_01_09-yourdataflowjobid     ps-topic                    Streaming  2020-12-28 20:01:10  Running  us-central1

Membuat pemetaan transformasi

Jika pipeline pengganti Anda mengubah nama transformasi dari nama di pipeline sebelumnya, layanan Dataflow memerlukan pemetaan transformasi. Pemetaan transformasi memetakan transformasi bernama dalam kode pipeline sebelumnya ke nama dalam kode pipeline pengganti.

Java

Teruskan pemetaan menggunakan opsi command line --transformNameMapping, dengan format umum berikut:

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

Anda hanya perlu memberikan entri pemetaan di --transformNameMapping untuk nama transformasi yang telah berubah antara pipeline sebelumnya dan pipeline pengganti.

Saat menjalankan dengan --transformNameMapping, Anda mungkin perlu meng-escape tanda kutip sesuai dengan shell Anda. Misalnya, di Bash:

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

Python

Teruskan pemetaan menggunakan opsi command line --transform_name_mapping, dengan format umum berikut:

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

Anda hanya perlu memberikan entri pemetaan di --transform_name_mapping untuk nama transformasi yang telah berubah antara pipeline sebelumnya dan pipeline pengganti.

Saat menjalankan dengan --transform_name_mapping, Anda mungkin perlu meng-escape tanda kutip sesuai dengan shell Anda. Misalnya, di Bash:

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

Go

Teruskan pemetaan menggunakan opsi command line --transform_name_mapping, dengan format umum berikut:

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

Anda hanya perlu memberikan entri pemetaan di --transform_name_mapping untuk nama transformasi yang telah berubah antara pipeline sebelumnya dan pipeline pengganti.

Saat menjalankan dengan --transform_name_mapping, Anda mungkin perlu meng-escape tanda kutip sesuai dengan shell Anda. Misalnya, di Bash:

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

gcloud

Teruskan pemetaan menggunakan opsi --transform-name-mappings, dengan format umum berikut:

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

Anda hanya perlu memberikan entri pemetaan di --transform-name-mappings untuk nama transformasi yang telah berubah antara pipeline sebelumnya dan pipeline pengganti.

Saat menjalankan dengan --transform-name-mappings, Anda mungkin perlu meng-escape tanda kutip sesuai dengan shell Anda. Misalnya, di Bash:

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

REST

Teruskan pemetaan menggunakan kolom transformNameMapping, menggunakan format umum berikut:

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

Anda hanya perlu memberikan entri pemetaan di transformNameMapping untuk nama transformasi yang telah berubah antara pipeline sebelumnya dan pipeline pengganti.

Menentukan nama transformasi

Nama transformasi di setiap instance dalam peta adalah nama yang Anda berikan saat menerapkan transformasi di pipeline. Contoh:

Java

  .apply("FormatResults", ParDo
    .of(new DoFn&lt;KV&lt;String, Long&gt;&gt;, 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)

Anda juga bisa mendapatkan nama transformasi untuk tugas sebelumnya dengan memeriksa grafik eksekusi tugas di Antarmuka Pemantauan Dataflow:

Grafik eksekusi untuk pipeline WordCount. — Gambar 2: Grafik eksekusi untuk pipeline WordCount seperti yang ditampilkan di Antarmuka Pemantauan Dataflow.

Penamaan transformasi gabungan

Nama transformasi bersifat hierarkis, berdasarkan hierarki transformasi di pipeline Anda. Jika pipeline Anda memiliki transformasi gabungan, transformasi bertingkat diberi nama berdasarkan transformasi yang memuatnya. Misalnya, misalkan pipeline Anda berisi transformasi gabungan bernama CountWidgets, yang berisi transformasi dalam bernama Parse. Nama lengkap transformasi Anda adalah CountWidgets/Parse, dan Anda harus menentukan nama lengkap tersebut dalam pemetaan transformasi.

Jika pipeline baru Anda memetakan transformasi gabungan ke nama yang berbeda, semua transformasi bertingkat juga akan otomatis diganti namanya. Anda harus menentukan nama yang diubah untuk transformasi dalam di pemetaan transformasi.

Merefaktorisasi hierarki transformasi

Jika pipeline pengganti menggunakan hierarki transformasi yang berbeda dengan pipeline sebelumnya, Anda harus mendeklarasikan pemetaan secara eksplisit. Anda mungkin memiliki hierarki transformasi yang berbeda karena Anda telah memfaktorkan ulang transformasi gabungan, atau pipeline Anda bergantung pada transformasi gabungan dari library yang berubah.

Misalnya, pipeline Anda sebelumnya menerapkan transformasi gabungan, CountWidgets, yang berisi transformasi dalam bernama Parse. Pipeline pengganti merefaktorisasi CountWidgets, dan menyusun bertingkat Parse di dalam transformasi lain bernama Scan. Agar update berhasil, Anda harus secara eksplisit memetakan nama transformasi lengkap di pipeline sebelumnya (CountWidgets/Parse) ke nama transformasi di pipeline baru (CountWidgets/Scan/Parse):