Memperkaya data streaming

Apache Beam menyederhanakan alur kerja pengayaan data dengan menyediakan transformasi pengayaan siap pakai yang dapat Anda tambahkan ke pipeline. Halaman ini menjelaskan cara menggunakan transformasi pengayaan Apache Beam untuk memperkaya data streaming Anda.

Saat Anda memperkaya data, Anda akan menambah data mentah dari satu sumber dengan menambahkan data terkait dari sumber kedua. Data tambahan dapat berasal dari berbagai sumber, seperti Bigtable atau BigQuery. Transformasi pengayaan Apache Beam menggunakan pencarian nilai kunci untuk menghubungkan data tambahan ke data mentah.

Contoh berikut memberikan beberapa kasus saat pengayaan data berguna:

  • Anda ingin membuat pipeline e-commerce yang menangkap aktivitas pengguna dari situs atau aplikasi dan memberikan rekomendasi yang disesuaikan. Transformasi ini menggabungkan aktivitas ke dalam data pipeline sehingga Anda dapat memberikan rekomendasi yang disesuaikan.
  • Anda memiliki data pengguna yang ingin digabungkan dengan data geografis untuk melakukan analisis berbasis geografi.
  • Anda ingin membuat pipeline yang mengumpulkan data dari perangkat internet of things (IoT) yang mengirimkan peristiwa telemetri.

Manfaat

Transformasi pengayaan memiliki manfaat berikut:

  • Mengubah data Anda tanpa mengharuskan Anda menulis kode yang kompleks atau mengelola library yang mendasarinya.
  • Menyediakan pengendali sumber bawaan.
  • Menggunakan pembatasan kapasitas sisi klien untuk mengelola pembatasan kapasitas permintaan. Permintaan akan di-backoff secara eksponensial dengan strategi percobaan ulang default. Anda dapat mengonfigurasi pembatasan kapasitas agar sesuai dengan kasus penggunaan Anda.

Dukungan dan batasan

Transformasi pengayaan memiliki persyaratan berikut:

  • Tersedia untuk pipeline batch dan streaming.
  • Pengendali BigTableEnrichmentHandler tersedia di Apache Beam Python SDK versi 2.54.0 dan yang lebih baru.
  • Pengendali BigQueryEnrichmentHandler tersedia di Apache Beam Python SDK versi 2.57.0 dan yang lebih baru.
  • Pengendali VertexAIFeatureStoreEnrichmentHandler tersedia di Apache Beam Python SDK versi 2.55.0 dan yang lebih baru.
  • Saat menggunakan Apache Beam Python SDK versi 2.55.0 dan yang lebih baru, Anda juga perlu menginstal klien Python untuk Redis.
  • Tugas Dataflow harus menggunakan Runner Portabel.

Menggunakan transformasi pengayaan

Untuk menggunakan transformasi pengayaan, sertakan kode berikut dalam pipeline Anda:

import apache_beam as beam
from apache_beam.transforms.enrichment import Enrichment
from apache_beam.transforms.enrichment_handlers.bigtable import BigTableEnrichmentHandler

bigtable_handler = BigTableEnrichmentHandler(...)

with beam.Pipeline() as p:
  output = (p
            ...
            | "Create" >> beam.Create(data)
            | "Enrich with Bigtable" >> Enrichment(bigtable_handler)
            ...
            )

Karena transformasi pengayaan melakukan gabungan silang secara default, desain gabungan kustom untuk memperkaya data input. Desain ini memastikan bahwa gabungan hanya menyertakan kolom yang ditentukan.

Pada contoh berikut, left adalah elemen input dari transformasi pengayaan, dan right adalah data yang diambil dari layanan eksternal untuk elemen input tersebut.

def custom_join(left: Dict[str, Any], right: Dict[str, Any]):
  enriched = {}
  enriched['FIELD_NAME'] = left['FIELD_NAME']
  ...
  return beam.Row(**enriched)

Parameter

Untuk menggunakan transformasi pengayaan, parameter EnrichmentHandler diperlukan.

Anda juga dapat menggunakan parameter konfigurasi untuk menentukan fungsi lambda untuk fungsi gabungan, waktu tunggu, pembatas kapasitas, atau pengulang (strategi percobaan ulang). Parameter konfigurasi berikut tersedia:

  • join_fn: Fungsi lambda yang mengambil kamus sebagai input dan menampilkan baris yang diperkaya (Callable[[Dict[str, Any], Dict[str, Any]], beam.Row]). Baris yang diperkaya menentukan cara menggabungkan data yang diambil dari API. Setelan defaultnya adalah gabungan silang.
  • timeout: Jumlah detik untuk menunggu permintaan diselesaikan oleh API sebelum waktu tunggu habis. Setelan defaultnya adalah 30 detik.
  • throttler: Menentukan mekanisme pembatasan kapasitas. Satu-satunya opsi yang didukung adalah pembatasan kapasitas adaptif sisi klien default.
  • repeater: Menentukan strategi percobaan ulang saat terjadi error seperti TooManyRequests dan TimeoutException. Setelan defaultnya adalah ExponentialBackOffRepeater.

Langkah berikutnya