I/O terkelola Dataflow

I/O terkelola memungkinkan Dataflow mengelola konektor I/O tertentu yang digunakan dalam pipeline Apache Beam. I/O Terkelola menyederhanakan pengelolaan pipeline yang terintegrasi dengan sumber dan tujuan yang didukung.

I/O Terkelola terdiri dari dua komponen yang bekerja bersama:

  • Transformasi Apache Beam yang menyediakan API umum untuk membuat konektor I/O (sumber dan sink).

  • Layanan Dataflow yang mengelola konektor I/O ini atas nama Anda, termasuk kemampuan untuk mengupgradenya secara independen dari versi Apache Beam.

Keuntungan I/O terkelola meliputi:

  • Upgrade otomatis. Dataflow otomatis mengupgrade konektor I/O terkelola di pipeline Anda. Artinya, pipeline Anda menerima perbaikan keamanan, peningkatan performa, dan perbaikan bug untuk konektor ini, tanpa memerlukan perubahan kode apa pun. Untuk mengetahui informasi selengkapnya, lihat Upgrade otomatis.

  • API yang konsisten. Secara tradisional, konektor I/O di Apache Beam memiliki API yang berbeda, dan setiap konektor dikonfigurasi dengan cara yang berbeda. I/O Terkelola menyediakan satu API konfigurasi yang menggunakan properti key-value, sehingga menghasilkan kode pipeline yang lebih sederhana dan konsisten. Untuk mengetahui informasi selengkapnya, lihat Configuration API.

Persyaratan

  • SDK berikut mendukung I/O terkelola:

    • Apache Beam SDK untuk Java versi 2.58.0 atau yang lebih baru.
    • Apache Beam SDK untuk Python versi 2.61.0 atau yang lebih baru.

  • Layanan backend memerlukan Dataflow Runner v2. Jika Runner v2 tidak diaktifkan, pipeline Anda tetap berjalan, tetapi tidak mendapatkan manfaat dari layanan I/O terkelola.

Upgrade otomatis

Pipeline Dataflow dengan konektor I/O terkelola otomatis menggunakan konektor versi terbaru yang andal. Upgrade otomatis terjadi pada titik berikut dalam siklus proses tugas:

  • Pengiriman tugas. Saat Anda mengirimkan tugas batch atau streaming, Dataflow menggunakan konektor I/O Terkelola versi terbaru yang telah diuji dan berfungsi dengan baik.

  • Upgrade berkelanjutan. Untuk tugas streaming, Dataflow mengupgrade konektor I/O Terkelola Anda di pipeline yang sedang berjalan saat versi baru tersedia. Anda tidak perlu khawatir memperbarui konektor atau versi Apache Beam pipeline secara manual.

    Secara default, upgrade bertahap terjadi dalam periode 30 hari — yaitu, upgrade dilakukan kira-kira setiap 30 hari. Anda dapat menyesuaikan periode atau menonaktifkan upgrade berkelanjutan berdasarkan per tugas. Untuk mengetahui informasi selengkapnya, lihat Menetapkan jendela upgrade bertahap.

    Seminggu sebelum upgrade, Dataflow menulis pesan notifikasi ke log pesan tugas.

  • Tugas penggantian. Untuk tugas streaming, Dataflow memeriksa update setiap kali Anda meluncurkan tugas pengganti, dan otomatis menggunakan versi terbaru yang diketahui berfungsi dengan baik. Dataflow melakukan pemeriksaan ini meskipun Anda tidak mengubah kode apa pun dalam tugas penggantian.

Diagram berikut menunjukkan proses upgrade. Pengguna membuat pipeline Apache Beam menggunakan SDK versi X. Dataflow mengupgrade versi I/O Terkelola ke versi terbaru yang didukung. Upgrade terjadi saat pengguna mengirimkan tugas, setelah periode upgrade bertahap, atau saat pengguna mengirimkan tugas penggantian.

Diagram yang menunjukkan proses upgrade I/O Terkelola.

Proses upgrade menambahkan waktu mulai sekitar dua menit untuk tugas pertama (per project) yang menggunakan I/O terkelola, dan dapat sekitar setengah menit untuk tugas berikutnya. Untuk upgrade berkelanjutan, layanan Dataflow meluncurkan tugas pengganti. Hal ini dapat mengakibatkan periode nonaktif sementara untuk pipeline Anda karena kumpulan pekerja yang ada dimatikan dan kumpulan pekerja baru dimulai. Untuk memeriksa status operasi I/O terkelola, cari entri log yang menyertakan string "Managed Transform(s)".

Menetapkan periode upgrade berkelanjutan

Untuk menentukan periode upgrade tugas Dataflow streaming, tetapkan managed_transforms_rolling_upgrade_window opsi layanan sama dengan jumlah hari. Nilai harus antara 10 dan 90 hari, inklusif.

Java

--dataflowServiceOptions=managed_transforms_rolling_upgrade_window=DAYS

Python

--dataflow_service_options=managed_transforms_rolling_upgrade_window=DAYS

gcloud

Gunakan perintah gcloud dataflow jobs run dengan opsi additional-experiments. Jika Anda menggunakan Template Flex yang menggunakan I/O Terkelola, gunakan perintah gcloud dataflow flex-template run.

--additional-experiments=managed_transforms_rolling_upgrade_window=DAYS

Untuk menonaktifkan upgrade berkelanjutan, setel opsi layanan managed_transforms_rolling_upgrade_window ke never. Anda tetap dapat memicu update dengan meluncurkan tugas penggantian.

Java

--dataflowServiceOptions=managed_transforms_rolling_upgrade_window=never

Python

--dataflow_service_options=managed_transforms_rolling_upgrade_window=never

Go

--dataflow_service_options=managed_transforms_rolling_upgrade_window=never

gcloud

Gunakan perintah gcloud dataflow jobs run dengan opsi additional-experiments. Jika Anda menggunakan Template Flex, gunakan perintah gcloud dataflow flex-template run.

--additional-experiments=managed_transforms_rolling_upgrade_window=never

Configuration API

I/O Terkelola adalah transformasi Apache Beam siap pakai yang menyediakan API yang konsisten untuk mengonfigurasi sumber dan sink.

Java

Untuk membuat sumber atau tujuan yang didukung oleh I/O Terkelola, Anda menggunakan class Managed. Tentukan sumber atau tujuan yang akan di-instantiate, dan teruskan sekumpulan parameter konfigurasi, mirip dengan berikut ini:

Map config = ImmutableMap.<String, Object>builder()
    .put("config1", "abc")
    .put("config2", 1);

pipeline.apply(Managed.read(/*Which source to read*/).withConfig(config))
    .getSinglePCollection();

Anda juga dapat meneruskan parameter konfigurasi sebagai file YAML. Untuk contoh kode lengkap, lihat Membaca dari Apache Iceberg.

Python

Impor modul apache_beam.transforms.managed dan panggil metode managed.Read atau managed.Write. Tentukan sumber atau tujuan yang akan di-instantiate, dan teruskan serangkaian parameter konfigurasi, mirip dengan berikut ini:

pipeline
| beam.managed.Read(
    beam.managed.SOURCE, # Example: beam.managed.KAFKA
    config={
      "config1": "abc",
      "config2": 1
    }
)

Anda juga dapat meneruskan parameter konfigurasi sebagai file YAML. Untuk contoh kode lengkap, lihat Membaca dari Apache Kafka.

Tujuan dinamis

Untuk beberapa sink, konektor I/O terkelola dapat memilih tujuan secara dinamis berdasarkan nilai kolom dalam data masuk.

Untuk menggunakan tujuan dinamis, berikan string template untuk tujuan. String template dapat menyertakan nama kolom dalam tanda kurung kurawal, seperti "tables.{field1}". Saat runtime, konektor mengganti nilai kolom untuk setiap rekaman masuk, guna menentukan tujuan untuk rekaman tersebut.

Misalnya, data Anda memiliki kolom bernama airport. Anda dapat menetapkan tujuan ke "flights.{airport}". Jika airport=SFO, data akan ditulis ke flights.SFO. Untuk kolom bertingkat, gunakan notasi titik. Misalnya: {top.middle.nested}.

Untuk kode contoh yang menunjukkan cara menggunakan tujuan dinamis, lihat Menulis dengan tujuan dinamis.

Pemfilteran

Anda dapat mengecualikan kolom tertentu sebelum ditulis ke tabel tujuan. Untuk sink yang mendukung tujuan dinamis, Anda dapat menggunakan parameter drop, keep, atau only untuk tujuan ini. Parameter ini memungkinkan Anda menyertakan metadata tujuan dalam rekaman input, tanpa menulis metadata ke tujuan.

Anda dapat menetapkan paling banyak satu parameter ini untuk tujuan tertentu.

Parameter konfigurasi Jenis data Deskripsi
drop daftar string Daftar nama kolom yang akan dihapus sebelum menulis ke tujuan.
keep daftar string Daftar nama kolom yang akan dipertahankan saat menulis ke tujuan. Kolom lainnya akan dihapus.
only string Nama tepat satu kolom yang akan digunakan sebagai rekaman tingkat teratas untuk ditulis saat menulis ke tujuan. Semua kolom lainnya akan dihapus. Kolom ini harus berjenis baris.

Sumber dan sink yang didukung

I/O Terkelola mendukung sumber dan sink berikut.

Untuk mengetahui informasi selengkapnya, lihat Managed I/O Connectors dalam dokumentasi Apache Beam.