Tepat-sekali di Dataflow

Dataflow mendukung pemrosesan data tepat satu kali. Halaman ini menjelaskan cara Dataflow menerapkan pemrosesan tepat satu kali sekaligus memastikan latensi rendah.

Ringkasan

Pipeline batch selalu menggunakan pemrosesan tepat satu kali. Secara default, pipeline streaming menggunakan pemrosesan tepat satu kali, tetapi juga dapat menggunakan pemrosesan minimal satu kali.

Pemrosesan tepat satu kali memberikan jaminan tentang hasil pemrosesan kumpulan data, termasuk hasil dari setiap tahap pipeline. Secara khusus, untuk setiap kumpulan data yang masuk ke pipeline dari sumber, atau diterima pada suatu tahap dari tahap sebelumnya, Dataflow memastikan hal berikut:

Catatan diproses dan tidak hilang.
Setiap hasil pemrosesan yang tetap berada dalam pipeline ditampilkan maksimal satu kali.

Dengan kata lain, kumpulan data diproses setidaknya satu kali, dan hasilnya di-commit tepat satu kali.

Pemrosesan tepat satu kali memastikan bahwa hasilnya akurat, tanpa kumpulan data duplikat dalam output. Dataflow dioptimalkan untuk meminimalkan latensi, sekaligus mempertahankan semantik tepat satu kali. Namun, pemrosesan tepat satu kali tetap memerlukan biaya untuk melakukan penghapusan duplikat. Untuk kasus penggunaan yang dapat menoleransi data duplikat, sering kali Anda dapat mengurangi biaya dan meningkatkan latensi dengan mengaktifkan mode minimal satu kali. Untuk informasi selengkapnya tentang memilih antara streaming tepat satu kali dan minimal satu kali, lihat Menyetel mode streaming pipeline.

Data terlambat

Pemrosesan tepat satu kali memastikan akurasi pipeline: Jika pipeline memproses kumpulan data, Dataflow memastikan bahwa kumpulan data tercermin dalam output, dan kumpulan data tersebut tidak diduplikasi.

Namun, dalam pipeline streaming, pemrosesan tepat satu kali tidak dapat menjamin bahwa hasil lengkap, karena kumpulan data mungkin terlambat tiba. Misalnya, anggaplah pipeline Anda melakukan agregasi selama jangka waktu tertentu, seperti Count. Dengan pemrosesan tepat satu kali, hasilnya akurat untuk data yang masuk dalam periode yang tepat waktu, tetapi data yang terlambat mungkin akan dihapus.

Umumnya, tidak ada cara untuk menjamin kelengkapan dalam pipeline streaming, karena secara teori, rekaman dapat tiba terlambat secara acak. Dalam kasus yang membatasi ini, Anda harus menunggu selamanya untuk memberikan hasil. Lebih praktisnya, Apache Beam memungkinkan Anda mengonfigurasi batas untuk menghapus data yang terlambat dan kapan harus memunculkan hasil gabungan. Untuk informasi selengkapnya, lihat Watermark dan data terlambat dalam dokumentasi Apache Beam.

Efek samping

Efek samping tidak dijamin memiliki semantik tepat satu kali. Yang penting, ini termasuk penulisan output ke penyimpanan eksternal, kecuali jika sink juga mengimplementasikan semantik tepat satu kali.

Secara khusus, Dataflow tidak menjamin bahwa setiap data melewati setiap transformasi tepat satu kali. Karena percobaan ulang atau kegagalan pekerja, Dataflow mungkin mengirim data melalui transformasi beberapa kali, atau bahkan secara bersamaan di beberapa pekerja.

Sebagai bagian dari pemrosesan tepat satu kali, Dataflow akan menghapus duplikat output. Namun, jika kode dalam transformasi memiliki efek samping, efek tersebut dapat terjadi beberapa kali. Misalnya, jika transformasi melakukan panggilan layanan jarak jauh, panggilan tersebut dapat dilakukan beberapa kali untuk kumpulan data yang sama. Efek samping bahkan dapat menyebabkan kehilangan data dalam beberapa situasi. Misalnya, anggaplah transformasi membaca file untuk menghasilkan output, lalu segera menghapus file tersebut tanpa menunggu output di-commit. Jika terjadi error saat melakukan commit untuk hasilnya, Dataflow akan mencoba kembali transformasi, tetapi sekarang transformasi tidak dapat membaca file yang dihapus.

Logging

Output log dari pemrosesan menunjukkan bahwa pemrosesan terjadi, tetapi tidak menunjukkan apakah data di-commit atau tidak. Oleh karena itu, file log mungkin menunjukkan bahwa data diproses beberapa kali meskipun hasil data yang diproses di-commit ke penyimpanan persisten hanya sekali. Selain itu, log tidak selalu mencerminkan data yang telah diproses dan di-commit. Log mungkin dihapus karena throttling atau hilang karena masalah layanan logging lainnya.

Streaming tepat satu kali

Bagian ini menjelaskan cara Dataflow mengimplementasikan pemrosesan tepat satu kali untuk tugas streaming, termasuk cara Dataflow mengelola kompleksitas seperti pemrosesan non-deterministik, data yang terlambat, dan kode kustom.

Acak streaming Dataflow

Tugas streaming Dataflow berjalan di banyak pekerja berbeda secara paralel dengan menetapkan rentang tugas ke setiap pekerja. Meskipun tugas dapat berubah dari waktu ke waktu sebagai respons terhadap kegagalan pekerja, penskalaan otomatis, atau peristiwa lainnya, setelah setiap transformasi GroupByKey, semua data dengan kunci yang sama akan diproses pada worker yang sama. Transformasi GroupByKey sering digunakan oleh transformasi komposit, seperti Count, FileIO, dan seterusnya. Untuk memastikan bahwa data untuk kunci tertentu berakhir pada pekerja yang sama, pekerja Dataflow mengacak data di antara mereka menggunakan metode remoteproc call (RPC).

Untuk menjamin bahwa data tidak hilang selama shuffle, Dataflow menggunakan pencadangan upstream. Dengan pencadangan upstream, pekerja yang mengirim kumpulan data akan mencoba ulang RPC hingga menerima konfirmasi positif bahwa data telah diterima. Efek samping pemrosesan kumpulan data di-commit ke downstream penyimpanan persisten. Jika pekerja yang mengirim data menjadi tidak tersedia, Dataflow akan terus mencoba ulang RPC, yang memastikan bahwa setiap data dikirim setidaknya satu kali.

Karena percobaan ulang ini dapat membuat duplikat, setiap pesan akan diberi tag dengan ID unik. Setiap penerima menyimpan katalog semua ID yang telah dilihat dan diproses. Saat data diterima, Dataflow mencari ID-nya di katalog. Jika ID ditemukan, berarti data tersebut telah diterima dan di-commit, lalu dihapus sebagai duplikat. Untuk memastikan bahwa ID kumpulan data stabil, setiap output dari langkah ke langkah akan diperiksa ke penyimpanan. Akibatnya, jika pesan yang sama dikirim beberapa kali karena panggilan RPC berulang, pesan hanya akan di-commit ke penyimpanan sekali.

Memastikan latensi rendah

Agar pemrosesan tepat satu kali dapat dilakukan, I/O harus dikurangi, khususnya dengan mencegah I/O di setiap kumpulan data. Untuk mencapai tujuan ini, Dataflow menggunakan filter Bloom dan pembersihan sampah memori.

Filter mekar

Filter bloom adalah struktur data ringkas yang memungkinkan pemeriksaan cepat untuk keanggotaan set. Dalam Dataflow, setiap pekerja menyimpan filter Bloom untuk setiap ID yang dilihatnya. Saat ID kumpulan data baru tiba, pekerja akan mencari ID tersebut di filter. Jika filter menampilkan nilai salah (false), kumpulan data ini bukan duplikat, dan worker tidak mencari ID dalam penyimpanan yang stabil.

Dataflow menyimpan sekumpulan filter Bloom bergulir yang dikelompokkan menurut waktu. Saat data masuk, Dataflow memilih filter yang sesuai untuk diperiksa berdasarkan stempel waktu sistem. Langkah ini mencegah filter Bloom agar tidak jenuh saat filter dikumpulkan sampah, dan juga membatasi jumlah data yang perlu dipindai saat memulai.

Pembersihan sampah memori

Agar penyimpanan tidak terisi dengan ID kumpulan data, Dataflow menggunakan pembersihan sampah memori untuk menghapus kumpulan data lama. Dataflow menggunakan stempel waktu sistem untuk menghitung watermark pembersihan sampah memori.

Watermark ini didasarkan pada jumlah waktu fisik yang dihabiskan untuk menunggu pada suatu tahap. Oleh karena itu, LLM juga memberikan informasi tentang bagian pipeline mana yang lambat. Metadata ini adalah dasar untuk metrik jeda sistem yang ditampilkan di Antarmuka pemantauan Dataflow.

Jika kumpulan data memiliki stempel waktu yang lebih lama dari watermark, dan jika ID untuk periode tersebut sudah dibersihkan sampah memorinya, kumpulan data tersebut akan diabaikan. Karena watermark rendah yang memicu pembersihan sampah memori tidak dilanjutkan hingga pengiriman rekaman dikonfirmasi, catatan yang terlambat diterima ini adalah duplikat.

Sumber non-deterministik

Dataflow menggunakan Apache Beam SDK untuk membaca data ke dalam pipeline. Jika pemrosesan gagal, Dataflow mungkin akan mencoba membaca lagi dari sumber. Dalam situasi tersebut, Dataflow perlu memastikan bahwa setiap kumpulan data unik yang dihasilkan oleh sumber dicatat tepat satu kali. Untuk sumber deterministik, seperti Pub/Sub Lite atau Kafka, data dibaca berdasarkan offset yang direkam, sehingga mengurangi kebutuhan untuk langkah ini.

Karena Dataflow tidak dapat menetapkan ID data secara otomatis, sumber non-deterministik harus memberi tahu Dataflow mengenai ID data untuk menghindari duplikasi. Jika sumber memberikan ID unik untuk setiap data, konektor akan menggunakan mode acak di pipeline untuk menghapus duplikat. Kumpulan data dengan ID yang sama akan difilter. Untuk mengetahui contoh cara Dataflow mengimplementasikan pemrosesan tepat satu kali saat menggunakan Pub/Sub sebagai sumber, lihat bagian Penghapusan duplikat yang efisien di halaman Streaming dengan Pub/Sub.

Saat Anda menjalankan DoFn kustom sebagai bagian dari pipeline, Dataflow tidak menjamin kode ini hanya dijalankan sekali per data. Untuk menjamin setidaknya satu pemrosesan jika terjadi kegagalan pekerja, Dataflow dapat menjalankan data tertentu melalui transformasi beberapa kali, atau mungkin menjalankan data yang sama secara bersamaan pada beberapa pekerja. Jika Anda menyertakan kode dalam pipeline yang melakukan hal-hal seperti menghubungi layanan luar, tindakan tersebut dapat dijalankan lebih dari sekali untuk kumpulan data tertentu.

Untuk membuat pemrosesan non-deterministik menjadi deterministik secara efektif, gunakan checkpointing. Saat Anda menggunakan checkpoint, setiap output dari transformasi di-checkpoint ke penyimpanan yang stabil dengan ID uniknya sebelum dikirim ke tahap berikutnya. Mencoba kembali pengiriman acak Dataflow akan menyampaikan output yang telah diberi checkpoint. Meskipun kode Anda mungkin dijalankan beberapa kali, Dataflow memastikan bahwa output hanya dari salah satu operasi tersebut yang akan disimpan. Dataflow menggunakan penyimpanan yang konsisten, sehingga mencegah duplikat ditulis ke penyimpanan yang stabil.

Penayangan output tepat satu kali

Apache Beam SDK menyertakan sink bawaan yang dirancang untuk memastikan bahwa sink tidak menghasilkan duplikat. Jika memungkinkan, gunakan salah satu sink bawaan ini.

Jika Anda perlu menulis sink Anda sendiri, pendekatan terbaik adalah membuat objek fungsi Anda idempoten sehingga dapat dicoba lagi sesering mungkin tanpa menyebabkan efek samping yang tidak diinginkan. Meskipun demikian, sering kali beberapa komponen transformasi yang mengimplementasikan fungsi sink bersifat non-deterministik dan dapat berubah jika dicoba ulang.

Misalnya, dalam agregasi berjendela, kumpulan data di jendela mungkin bersifat non-deterministik. Secara khusus, jendela mungkin mencoba diaktifkan dengan elemen e0, e1, e2. Pekerja mungkin mengalami error sebelum melakukan pemrosesan jendela, tetapi tidak sebelum elemen tersebut dikirim sebagai efek samping. Saat pekerja dimulai ulang, jendela akan diaktifkan lagi, dan elemen akhir e3 muncul. Karena elemen ini muncul sebelum jendela di-commit, elemen tersebut tidak dihitung sebagai data yang terlambat, sehingga DoFn dipanggil lagi dengan elemen e0, e1, e2, e3. Elemen-elemen ini kemudian dikirim ke operasi efek samping. Idempotensi tidak membantu dalam skenario ini, karena kumpulan catatan logis yang berbeda dikirim setiap waktu.

Untuk mengatasi non-determinasi di Dataflow, gunakan transformasi Reshuffle bawaan. Saat Dataflow mengacak data, Dataflow menulis data secara permanen sehingga elemen yang tidak dihasilkan akan stabil jika operasi dicoba ulang setelah pengacakan terjadi. Penggunaan transformasi Reshuffle menjamin bahwa hanya satu versi output DoFn yang dapat melewati batas acak. Pola berikut memastikan bahwa operasi efek samping selalu menerima catatan deterministik ke output:

c.apply(Window.<..>into(FixedWindows.of(Duration.standardMinutes(1))))
 .apply(GroupByKey.<..>.create())
 .apply(new PrepareOutputData())
 .apply(Reshuffle.<..>of())
 .apply(WriteToSideEffect());

Untuk memastikan runner Dataflow mengetahui bahwa elemen harus stabil sebelum menjalankan DoFn, tambahkan anotasi RequiresStableInput ke DoFn.