Menulis data dari Kafka ke BigQuery menggunakan Dataflow

Halaman ini menunjukkan cara menggunakan Dataflow untuk membaca data dari Google Cloud Managed Service for Apache Kafka dan menulis rekaman ke tabel BigQuery. Tutorial ini menggunakan template Apache Kafka ke BigQuery untuk membuat tugas Dataflow.

Ringkasan

Apache Kafka adalah platform open source untuk streaming peristiwa. Kafka biasanya digunakan dalam arsitektur terdistribusi untuk memungkinkan komunikasi antar-komponen yang tidak terikat erat. Anda dapat menggunakan Dataflow untuk membaca peristiwa dari Kafka, memprosesnya, dan menulis hasilnya ke tabel BigQuery untuk analisis lebih lanjut.

Managed Service for Apache Kafka adalah layanan yang membantu Anda menjalankan cluster Kafka yang aman dan skalabel. Google Cloud

Izin yang diperlukan

Akun layanan pekerja Dataflow harus memiliki peran Identity and Access Management (IAM) berikut:

• Klien Kafka Terkelola (roles/managedkafka.client)
• BigQuery Data Editor (roles/bigquery.dataEditor)

Untuk mengetahui informasi selengkapnya, lihat Keamanan dan izin Dataflow.

Membuat cluster Kafka

Pada langkah ini, Anda akan membuat cluster Managed Service for Apache Kafka. Untuk mengetahui informasi selengkapnya, lihat Membuat cluster Managed Service for Apache Kafka.

gcloud managed-kafka clusters create CLUSTER \
--location=REGION \
--cpu=3 \
--memory=3GiB \
--subnets=projects/PROJECT_ID/regions/REGION/subnetworks/SUBNET_NAME

Pembuatan cluster biasanya memerlukan waktu 20-30 menit.

Membuat topik Kafka

Setelah cluster Managed Service for Apache Kafka dibuat, buat topik.

gcloud managed-kafka topics create TOPIC_NAME \
--cluster=CLUSTER \
--location=REGION \
--partitions=10 \
--replication-factor=3

Membuat tabel BigQuery

Pada langkah ini, Anda akan membuat tabel BigQuery dengan skema berikut:

Jika Anda belum memiliki set data BigQuery, buat set data terlebih dahulu. Untuk mengetahui informasi selengkapnya, lihat Membuat set data. Kemudian, buat tabel kosong baru:

bq mk --table \
  PROJECT_ID:DATASET_NAME.TABLE_NAME \
  name:STRING,customer_id:INTEGER

Menjalankan tugas Dataflow

Setelah membuat cluster Kafka dan tabel BigQuery, jalankan template Dataflow.

gcloud dataflow flex-template run kafka-to-bq \
--template-file-gcs-location gs://dataflow-templates/latest/flex/Kafka_to_BigQuery \
--region LOCATION \
--parameters \
readBootstrapServerAndTopic=projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID/topics/TOPIC,\
persistKafkaKey=false,\
writeMode=SINGLE_TABLE_NAME,\
kafkaReadOffset=earliest,\
kafkaReadAuthenticationMode=APPLICATION_DEFAULT_CREDENTIALS,\
messageFormat=JSON,\
outputTableSpec=PROJECT_ID:DATASET_NAME.TABLE_NAME\
useBigQueryDLQ=true,\
outputDeadletterTable=PROJECT_ID:DATASET_NAME.ERROR_TABLE_NAME

Mengirim pesan ke Kafka

Setelah tugas Dataflow dimulai, Anda dapat mengirim pesan ke Kafka, dan pipeline akan menuliskannya ke BigQuery.

1. Buat VM di subnet yang sama dengan cluster Kafka dan instal alat command line Kafka. Untuk mendapatkan petunjuk mendetail, lihat Menyiapkan mesin klien di Memublikasikan dan menggunakan pesan dengan CLI.

2. Jalankan perintah berikut untuk menulis pesan ke topik Kafka:

   kafka-console-producer.sh \
    --topic TOPIC \
    --bootstrap-server bootstrap.CLUSTER_ID.LOCATION.managedkafka.PROJECT_ID.cloud.goog:9092 \
    --producer.config client.properties

   Ganti variabel berikut:

   • TOPIC: nama topik Kafka
   • CLUSTER_ID: nama cluster
   • LOCATION: region tempat cluster Anda berada
   • PROJECT_ID: nama Google Cloud project Anda

3. Di perintah, masukkan baris teks berikut untuk mengirim pesan ke Kafka:

   {"name": "Alice", "customer_id": 1}
   {"name": "Bob", "customer_id": 2}
   {"name": "Charles", "customer_id": 3}
   

Menggunakan antrean pesan yang dihentikan pengirimannya

Saat tugas berjalan, pipeline mungkin gagal menulis setiap pesan ke BigQuery. Kemungkinan error meliputi:

• Error serialisasi, termasuk JSON yang diformat dengan buruk.
• Error konversi jenis, yang disebabkan oleh ketidakcocokan dalam skema tabel dan data JSON.
• Kolom tambahan dalam data JSON yang tidak ada dalam skema tabel.

Error ini tidak menyebabkan tugas gagal, dan tidak muncul sebagai error di log tugas Dataflow. Sebagai gantinya, pipeline menggunakan antrean surat yang tidak terkirim untuk menangani jenis error ini.

Untuk mengaktifkan antrean pesan yang tidak terkirim saat Anda menjalankan template, tetapkan parameter template berikut:

• useBigQueryDLQ: true
• outputDeadletterTable: nama tabel BigQuery yang sepenuhnya memenuhi syarat; misalnya, my-project:dataset1.errors

Pipeline akan otomatis membuat tabel. Jika terjadi error saat memproses pesan Kafka, pipeline akan menulis entri error ke tabel.

Contoh pesan error:

Bekerja dengan jenis data BigQuery

Secara internal, konektor I/O Kafka mengonversi payload pesan JSON menjadi objek TableRow Apache Beam, dan menerjemahkan nilai kolom TableRow menjadi jenis BigQuery.

Tabel berikut menunjukkan representasi JSON dari jenis data BigQuery.

Data terstruktur

Jika pesan JSON Anda mengikuti skema yang konsisten, Anda dapat merepresentasikan objek JSON menggunakan jenis data STRUCT di BigQuery.

Dalam contoh berikut, kolom answers adalah objek JSON dengan dua subkolom, a dan b:

{"name":"Emily","answers":{"a":"yes","b":"no"}}

Pernyataan SQL berikut membuat tabel BigQuery dengan skema yang kompatibel:

CREATE TABLE my_dataset.kafka_events (name STRING, answers STRUCT<a STRING, b STRING>);

Tabel yang dihasilkan akan terlihat seperti berikut:

+-------+----------------------+
| name  |       answers        |
+-------+----------------------+
| Emily | {"a":"yes","b":"no"} |
+-------+----------------------+

Data semi-terstruktur

Jika pesan JSON Anda tidak mengikuti skema yang ketat, pertimbangkan untuk menyimpannya di BigQuery sebagai jenis data JSON. Dengan menyimpan data JSON sebagai jenis JSON, Anda tidak perlu menentukan skema di awal. Setelah penyerapan data, Anda dapat membuat kueri data menggunakan operator akses kolom (notasi titik) dan akses array di GoogleSQL. Untuk mengetahui informasi selengkapnya, lihat Bekerja dengan data JSON di GoogleSQL.

Menggunakan UDF untuk mengubah data

Tutorial ini mengasumsikan bahwa pesan Kafka diformat sebagai JSON, dan skema tabel BigQuery cocok dengan data JSON, tanpa transformasi yang diterapkan pada data.

Secara opsional, Anda dapat memberikan fungsi yang ditentukan pengguna (UDF) JavaScript yang mentransformasi data sebelum ditulis ke BigQuery. UDF juga dapat melakukan pemrosesan tambahan, seperti memfilter, menghapus informasi identitas pribadi (PII), atau memperkaya data dengan kolom tambahan.

Untuk mengetahui informasi selengkapnya, lihat Membuat fungsi yang ditentukan pengguna untuk template Dataflow.

Langkah berikutnya

• Menggunakan Dataflow dengan Managed Service for Apache Kafka
• Pelajari BigQuery