Halaman ini diterjemahkan oleh Cloud Translation API.

Memperbarui pipeline yang ada

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

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

Anda dapat memperbarui tugas dengan dua cara:

Pembaruan tugas yang sedang berlangsung: 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 tugas.
Tugas pengganti: Untuk menjalankan kode pipeline yang diperbarui atau memperbarui opsi tugas yang tidak didukung oleh update tugas yang sedang berlangsung, luncurkan tugas baru yang menggantikan tugas yang ada. Untuk memverifikasi apakah tugas pengganti valid, sebelum meluncurkan tugas baru, validasi grafik tugasnya.

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

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

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

Pembaruan opsi tugas dalam penerbangan

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

min-num-workers: jumlah minimum instance Compute Engine.
max-num-workers: jumlah maksimum instance Compute Engine.

Untuk info lowongan terbaru lainnya, Anda harus mengganti pekerjaan saat ini dengan lowongan yang sudah diperbarui. Untuk informasi selengkapnya, lihat Meluncurkan tugas pengganti.

Lakukan update yang sedang berlangsung

Untuk melakukan pembaruan opsi tugas yang sedang berlangsung, 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 \
  JOB_ID

Ganti kode berikut:

REGION: ID wilayah tugas
JOB_ID: ID tugas yang akan diperbarui

Anda juga dapat mengupdate --min-num-workers dan --max-num-workers satu per satu.

REST

Gunakan projects.locations.jobs.update metode:

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

Ganti kode berikut:

PROJECT_ID: ID project Google Cloud untuk tugas Dataflow
REGION: ID wilayah tugas
JOB_ID: ID tugas yang akan diperbarui
MINIMUM_WORKERS: jumlah minimum instance Compute Engine
MAXIMUM_WORKERS: jumlah maksimum instance Compute Engine

Anda juga dapat memperbarui min_num_workers dan max_num_workers satu per satu. Tentukan parameter yang akan diperbarui dalam parameter kueri updateMask, dan sertakan nilai yang telah diperbarui dalam kolom runtimeUpdatableParams dari 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 yang sedang berlangsung. Error terjadi jika tugas belum dimulai atau sudah dibatalkan. Demikian pula, jika Anda meluncurkan tugas pengganti, tunggu hingga tugas tersebut mulai berjalan sebelum mengirimkan update yang sedang berlangsung ke tugas baru.

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

Memvalidasi tugas pengganti

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 pembaruan. Selain itu, Anda dapat memvalidasi update tanpa perlu menghentikan tugas asal, sehingga tugas tidak mengalami periode nonaktif.

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

Java

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

Python

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

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 perbarui.
Sertakan opsi layanan --dataflow_service_options=graph_validate_only.
Jika nama transformasi dalam pipeline telah berubah, Anda harus menyediakan pemetaan transformasi dan meneruskannya menggunakan opsi --transform_name_mapping.

gcloud

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

Teruskan opsi --update.
Setel 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 perbarui.
Sertakan opsi --additional-experiments=graph_validate_only.
Jika nama transformasi dalam pipeline telah 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 lowongan yang ingin Anda perbarui.

REST

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

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

Opsi layanan graph_validate_only hanya memvalidasi update pipeline. Jangan gunakan opsi ini saat membuat atau meluncurkan {i>pipe<i}. Untuk memperbarui pipeline Anda, luncurkan tugas pengganti tanpa opsi layanan graph_validate_only.

Saat 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 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 muncul di log tugas yang menjelaskan error inkompatibilitas. Konten pesan bergantung pada error.

Meluncurkan tugas pengganti

Anda mungkin mengganti pekerjaan yang ada karena alasan berikut:

Untuk menjalankan kode pipeline yang diperbarui.
Untuk memperbarui opsi tugas yang tidak mendukung update yang sedang berlangsung.

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

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

Java

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

Python

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

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 perbarui.
Jika nama transformasi dalam pipeline telah berubah, Anda harus menyediakan pemetaan transformasi dan meneruskannya menggunakan opsi --transform_name_mapping.

gcloud

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

Teruskan opsi --update.
Setel 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 perbarui.
Jika nama transformasi dalam pipeline telah berubah, Anda harus menyediakan pemetaan transformasi dan meneruskannya menggunakan opsi --transform-name-mappings.

REST

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

Ambil resource job untuk tugas yang ingin Anda ganti dengan 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 Anda berikutnya tidak melebihi batas ukuran. Jika Anda memerlukan informasi pekerjaan 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: ID project Google Cloud untuk tugas Dataflow
- REGION: region tugas yang ingin Anda perbarui
- JOB_ID: ID pekerjaan dari pekerjaan 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 tugas yang sama dengan ID tugas yang ingin Anda perbarui.
- JOB_NAME: nama pekerjaan yang sama dengan nama tugas yang ingin Anda perbarui.
Jika nama transformasi dalam pipeline telah berubah, Anda harus menyediakan pemetaan transformasi dan meneruskannya menggunakan kolom transformNameMapping.
Opsional: Untuk mengirim permintaan Anda 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.

Tentukan nama tugas pengganti Anda

Java

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

Python

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

Go

Saat Anda 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 tugas yang ingin Anda perbarui. Untuk menemukan nilai nama tugas yang benar, pilih pekerjaan Anda sebelumnya di Dataflow Monitoring Interface. Kemudian, di panel samping Info pekerjaan, cari kolom ID Pekerjaan.

Untuk menemukan nilai nama tugas yang benar, pilih pekerjaan Anda sebelumnya di Dataflow Monitoring Interface. Kemudian, di panel samping Info tugas, cari kolom Nama pekerjaan:

Panel samping Info tugas untuk tugas Dataflow yang sedang berjalan. — Gambar 1: Panel samping Info tugas untuk tugas Dataflow yang sedang berjalan dengan kolom Nama tugas.

Atau, buat kueri daftar tugas yang ada dengan menggunakan Antarmuka Command Line Dataflow. Masukkan perintah gcloud dataflow jobs list ke dalam jendela shell atau terminal untuk mendapatkan daftar tugas Dataflow di project Google Cloud Anda, dan cari 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 Anda sebelumnya ke nama dalam kode pipeline pengganti Anda.

Java

Teruskan pemetaan dengan menggunakan opsi command line --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 penggantian Anda.

Saat menjalankan dengan --transformNameMapping, Anda mungkin perlu meng-escape kutipan sebagaimana mestinya untuk shell Anda. Misalnya, di Bash:

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

Python

Teruskan pemetaan dengan menggunakan opsi command line --transform_name_mapping, menggunakan 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 penggantian Anda.

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

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

Go

Teruskan pemetaan dengan menggunakan opsi command line --transform_name_mapping, menggunakan 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 penggantian Anda.

Saat menjalankan dengan --transform_name_mapping, Anda mungkin perlu meng-escape kutipan sebagaimana mestinya untuk shell Anda. Misalnya, di Bash:

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

gcloud

Teruskan pemetaan menggunakan opsi --transform-name-mappings, dengan menggunakan 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 penggantian Anda.

Saat menjalankan dengan --transform-name-mappings, Anda mungkin harus meng-escape tanda kutip sesuai untuk 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 penggantian Anda.

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 Dataflow Monitoring Interface:

Grafik eksekusi untuk pipeline WordCount. — Gambar 2: Grafik eksekusi untuk pipeline WordCount seperti yang ditunjukkan 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 akan diberi nama sesuai dengan transformasi yang memuatnya. Misalnya, pipeline Anda berisi transformasi komposit 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 diganti namanya secara otomatis. Anda harus menentukan nama yang diubah untuk transformasi internal dalam pemetaan transformasi.

Memfaktorkan ulang hierarki transformasi

Jika pipeline pengganti Anda 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 komposit, atau pipeline Anda bergantung pada transformasi gabungan dari library yang telah berubah.

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