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 menyempurnakan kode pipeline.
• 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 ingin menambal kerentanan keamanan yang terkait dengan Container-Optimized OS untuk semua pekerja Dataflow.
• Anda ingin menskalakan pipeline Apache Beam streaming untuk menggunakan jumlah pekerja yang berbeda.

Anda dapat memperbarui tugas dengan dua cara:

• Pembaruan tugas dalam proses: 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 telah diupdate atau untuk memperbarui opsi tugas yang tidak didukung oleh pembaruan tugas dalam proses, luncurkan tugas baru yang menggantikan tugas yang ada. Untuk memverifikasi apakah tugas penggantian valid, sebelum meluncurkan tugas baru, validasi grafik tugasnya.

Saat Anda memperbarui tugas, layanan Dataflow akan melakukan pemeriksaan kompatibilitas antara tugas yang sedang berjalan dan tugas pengganti potensial. 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 saat memperbarui tugas. Untuk mengetahui informasi selengkapnya, lihat Bekerja dengan log pipeline. Untuk mengidentifikasi masalah pada kode pipeline, gunakan tingkat logging DEBUG.

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

Pembaruan opsi tugas dalam proses

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.
• worker-utilization-hint: the target CPU utilization, in the range [0.1, 0.9]

Untuk pembaruan tugas lainnya, Anda harus mengganti tugas saat ini dengan tugas yang diperbarui. Untuk mengetahui informasi selengkapnya, lihat Meluncurkan tugas penggantian.

Melakukan update saat dalam proses

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

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

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
  }
}

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 saat proses berlangsung. Error terjadi jika tugas belum dimulai atau sudah dibatalkan. Demikian pula, jika Anda meluncurkan tugas penggantian, tunggu hingga tugas tersebut mulai berjalan sebelum mengirimkan pembaruan dalam proses ke tugas baru.

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

Memvalidasi tugas penggantian

Untuk memverifikasi apakah tugas penggantian valid, sebelum meluncurkan tugas baru, validasi grafik tugasnya. Di Dataflow, grafik tugas adalah representasi grafis 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 Anda, ikuti langkah-langkah untuk meluncurkan tugas pengganti. Sertakan graph_validate_only Opsi layanan Dataflow dalam perintah update.

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

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

Opsi layanan graph_validate_only hanya memvalidasi pembaruan pipeline. Jangan gunakan opsi ini saat membuat atau meluncurkan pipeline. Untuk memperbarui pipeline, luncurkan tugas penggantian 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 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 ketidakcocokan. Konten pesan bergantung pada error.

Meluncurkan tugas penggantian

Anda dapat mengganti pekerjaan yang ada karena alasan berikut:

• Untuk menjalankan kode pipeline yang telah diupdate.
• Untuk memperbarui opsi tugas yang tidak mendukung pembaruan saat dalam proses.

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

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

Tentukan nama tugas penggantian Anda

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

Atau, 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 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.

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

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

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

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

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

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

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

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

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

Menentukan nama transformasi

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

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

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

  // 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 Monitoring 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, 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 komposit ke nama yang berbeda, semua transformasi bertingkat juga akan otomatis diganti namanya. Anda harus menentukan nama yang diubah untuk transformasi dalam di pemetaan transformasi.

Memfaktorkan ulang 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 memfaktorkan ulang transformasi komposit, atau pipeline Anda bergantung pada transformasi komposit dari library yang berubah.

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

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

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

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

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

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

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

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

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

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

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

Efek penggantian tugas

Saat Anda mengganti tugas yang ada, tugas baru akan menjalankan kode pipeline yang telah diperbarui. Layanan Dataflow mempertahankan nama tugas, tetapi menjalankan tugas pengganti dengan ID Tugas yang diperbarui. Proses ini dapat menyebabkan periode nonaktif saat tugas yang ada berhenti, pemeriksaan kompatibilitas berjalan, dan tugas baru dimulai.

Tugas penggantian mempertahankan item berikut:

• Data status perantara dari tugas sebelumnya. Cache dalam memori tidak disimpan.
• Rekaman data yang di-buffer atau metadata yang saat ini "sedang diproses" dari tugas sebelumnya. Misalnya, beberapa kumpulan data dalam pipeline Anda mungkin di-buffer saat menunggu periode untuk diselesaikan.
• Pembaruan opsi tugas dalam proses yang Anda terapkan ke tugas sebelumnya.

Data status menengah

Data status perantara dari tugas sebelumnya dipertahankan. Data status tidak mencakup cache dalam memori. Jika Anda ingin mempertahankan data cache dalam memori saat memperbarui pipeline, sebagai solusi sementara, refaktorkan pipeline Anda untuk mengonversi cache menjadi data status atau menjadi input samping. Untuk mengetahui informasi selengkapnya tentang penggunaan input samping, lihat Pola input samping dalam dokumentasi Apache Beam.

Pipeline streaming memiliki batas ukuran untuk ValueState dan untuk input samping. Akibatnya, jika Anda memiliki cache besar yang ingin dipertahankan, Anda mungkin perlu menggunakan penyimpanan eksternal, seperti Memorystore atau Bigtable.

Data dalam penerbangan

Data "dalam proses" masih diproses oleh transformasi di pipeline baru Anda. Namun, transformasi tambahan yang Anda tambahkan dalam kode pipeline penggantian mungkin atau mungkin tidak berlaku, bergantung pada tempat data di-buffer. Dalam contoh ini, pipeline yang ada memiliki transformasi berikut:

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

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

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

Anda dapat mengganti tugas dengan kode pipeline baru, sebagai berikut:

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

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

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

Meskipun Anda menambahkan transformasi untuk memfilter string yang diawali dengan huruf "A", transformasi berikutnya (FormatStrings) mungkin masih melihat string dalam buffer atau dalam proses yang diawali dengan "A" yang ditransfer dari tugas sebelumnya.

Mengubah windowing

Anda dapat mengubah strategi windowing dan pemicuan untuk elemen PCollection di pipeline penggantian, tetapi berhati-hatilah. Mengubah strategi penentuan jendela atau pemicu tidak akan memengaruhi data yang sudah di-buffer atau sedang dalam proses.

Sebaiknya Anda hanya mencoba perubahan kecil pada windowing pipeline, seperti mengubah durasi jendela waktu tetap atau geser. Melakukan perubahan besar pada jendela atau pemicu, seperti mengubah algoritma jendela, dapat memberikan hasil yang tidak dapat diprediksi pada output pipeline Anda.

Pemeriksaan kompatibilitas pekerjaan

Saat Anda meluncurkan tugas pengganti, layanan Dataflow akan melakukan pemeriksaan kompatibilitas antara tugas pengganti dan tugas sebelumnya. Jika pemeriksaan kompatibilitas berhasil, tugas sebelumnya akan dihentikan. Tugas pengganti Anda kemudian diluncurkan di layanan Dataflow sambil mempertahankan nama tugas yang sama. Jika pemeriksaan kompatibilitas gagal, tugas sebelumnya akan terus berjalan di layanan Dataflow dan tugas pengganti akan menampilkan error.

Pemeriksaan kompatibilitas menggunakan pemetaan transformasi yang diberikan untuk memastikan Dataflow dapat mentransfer data status perantara dari langkah-langkah dalam tugas sebelumnya ke tugas pengganti. Pemeriksaan kompatibilitas juga memastikan bahwa PCollection di pipeline Anda menggunakan Coder yang sama. Mengubah Coder dapat menyebabkan pemeriksaan kompatibilitas gagal karena data dalam proses atau rekaman yang di-buffer mungkin tidak diserialisasi dengan benar di pipeline penggantian.

Mencegah gangguan kompatibilitas

Perbedaan tertentu antara pipeline sebelumnya dan pipeline pengganti dapat menyebabkan pemeriksaan kompatibilitas gagal. Perbedaan ini mencakup:

• Mengubah grafik pipeline tanpa memberikan pemetaan. Saat Anda memperbarui tugas, Dataflow akan mencoba mencocokkan transformasi dalam tugas sebelumnya dengan transformasi dalam tugas pengganti. Proses pencocokan ini membantu Dataflow mentransfer data status perantara untuk setiap langkah. Jika Anda mengganti nama atau menghapus langkah apa pun, Anda harus memberikan pemetaan transformasi agar Dataflow dapat mencocokkan data status dengan tepat.
• Mengubah input samping untuk langkah. Menambahkan input samping ke atau menghapusnya dari transformasi dalam pipeline penggantian menyebabkan pemeriksaan kompatibilitas gagal.
• Mengubah Coder untuk langkah. Saat Anda memperbarui tugas, Dataflow akan mempertahankan semua rekaman data yang saat ini di-buffer dan menanganinya dalam tugas pengganti. Misalnya, data yang di-buffer dapat terjadi saat windowing sedang diselesaikan. Jika tugas penggantian menggunakan encoding data yang berbeda atau tidak kompatibel, Dataflow tidak dapat melakukan serialisasi atau deserialisasi rekaman ini.

• Menghapus operasi "stateful" dari pipeline Anda. Jika Anda menghapus operasi stateful dari pipeline, tugas pengganti Anda mungkin gagal dalam pemeriksaan kompatibilitas. Dataflow dapat menggabungkan beberapa langkah untuk efisiensi. Jika Anda menghapus operasi yang bergantung pada status dari dalam langkah gabungan, pemeriksaan akan gagal. Operasi stateful meliputi:

  • Transformasi yang menghasilkan atau menggunakan input samping.
  • Pembacaan I/O.
  • Transformasi yang menggunakan status yang dikunci.
  • Transformasi yang memiliki penggabungan jendela.

• Mengubah variabel DoFn stateful. Untuk tugas streaming yang sedang berlangsung, jika pipeline Anda menyertakan DoFns stateful, mengubah variabel DoFn stateful dapat menyebabkan pipeline gagal.

• Mencoba menjalankan penggantian tugas di zona geografis yang berbeda. Jalankan tugas penggantian di zona yang sama dengan tempat Anda menjalankan tugas sebelumnya.

Memperbarui skema

Apache Beam memungkinkan PCollection memiliki skema dengan kolom bernama, sehingga Coder eksplisit tidak diperlukan. Jika nama dan jenis kolom untuk skema tertentu tidak berubah (termasuk kolom bertingkat), skema tersebut tidak menyebabkan pemeriksaan update gagal. Namun, update mungkin masih diblokir jika segmen lain dari pipeline baru tidak kompatibel.

Mengembangkan skema

Sering kali, skema PCollection perlu dikembangkan karena persyaratan bisnis yang terus berubah. Layanan Dataflow memungkinkan melakukan perubahan berikut pada skema saat memperbarui pipeline:

• Menambahkan satu atau beberapa kolom baru ke skema, termasuk kolom bertingkat.
• Membuat jenis kolom wajib diisi (non-nullable) menjadi opsional (nullable).

Penghapusan kolom, perubahan nama kolom, atau perubahan jenis kolom tidak diizinkan selama pembaruan.

Meneruskan data tambahan ke operasi ParDo yang ada

Anda dapat meneruskan data tambahan (di luar band) ke operasi ParDo yang ada dengan menggunakan salah satu metode berikut, bergantung pada kasus penggunaan Anda:

• Serialkan informasi sebagai kolom di subclass DoFn Anda.
• Variabel apa pun yang dirujuk oleh metode dalam DoFn anonim akan diserialisasi secara otomatis.
• Menghitung data di dalam DoFn.startBundle().
• Teruskan data menggunakan ParDo.withSideInputs.

Untuk informasi lebih lanjut, lihat halaman berikut:

• Panduan pemrograman Apache Beam: ParDo, khususnya bagian tentang cara membuat DoFn dan input samping.
• Referensi Apache Beam SDK untuk Java: ParDo