Template Apache Kafka ke BigQuery

Template Apache Kafka ke BigQuery adalah pipeline streaming yang menyerap data teks dari cluster Google Cloud Managed Service for Apache Kafka, lalu menghasilkan data yang dihasilkan ke tabel BigQuery. Setiap error yang terjadi saat menyisipkan data ke dalam tabel output akan disisipkan ke dalam tabel error terpisah di BigQuery.

Anda juga dapat menggunakan template Apache Kafka ke BigQuery dengan Kafka eksternal atau yang dikelola sendiri.

Persyaratan pipeline

  • Server broker Apache Kafka harus berjalan dan dapat dijangkau dari mesin pekerja Dataflow.
  • Topik Apache Kafka harus ada.
  • Anda harus mengaktifkan Dataflow, BigQuery, dan Cloud Storage API. Jika autentikasi diperlukan, Anda juga harus mengaktifkan Secret Manager API.
  • Buat set data dan tabel BigQuery dengan skema yang sesuai untuk topik input Kafka Anda. Jika menggunakan beberapa skema dalam topik yang sama dan ingin menulis ke beberapa tabel, Anda tidak perlu membuat tabel sebelum mengonfigurasi pipeline.
  • Saat antrean pesan yang tidak terproses (dead letter) untuk template diaktifkan, buat tabel kosong yang tidak memiliki skema untuk antrean pesan yang tidak terproses.

Format pesan Kafka

Template Apache Kafka ke BigQuery mendukung pembacaan pesan dari Kafka dalam format berikut: CONFLUENT_AVRO_WIRE_FORMAT, AVRO_BINARY_FORMAT, dan JSON.

Autentikasi

Template Apache Kafka ke BigQuery mendukung autentikasi SASL/PLAIN ke broker Kafka.

Parameter template

Parameter yang diperlukan

  • readBootstrapServerAndTopic: Topik Kafka untuk membaca input.
  • writeMode: Menulis kumpulan data ke satu tabel atau beberapa tabel (berdasarkan skema). Mode DYNAMIC_TABLE_NAMES hanya didukung untuk AVRO_CONFLUENT_WIRE_FORMAT Source Message Format dan SCHEMA_REGISTRY Schema Source. Nama tabel target dibuat secara otomatis berdasarkan nama skema Avro dari setiap pesan, yang dapat berupa satu skema (membuat satu tabel) atau beberapa skema (membuat beberapa tabel). Mode SINGLE_TABLE_NAME menulis ke satu tabel (satu skema) yang ditentukan oleh pengguna. Nilai default-nya adalah SINGLE_TABLE_NAME.
  • kafkaReadAuthenticationMode: Mode autentikasi yang akan digunakan dengan cluster Kafka. Gunakan KafkaAuthenticationMethod.NONE untuk tidak ada autentikasi, KafkaAuthenticationMethod.SASL_PLAIN untuk nama pengguna dan sandi SASL/PLAIN, dan KafkaAuthenticationMethod.TLS untuk autentikasi berbasis sertifikat. KafkaAuthenticationMethod.APPLICATION_DEFAULT_CREDENTIALS hanya boleh digunakan untuk cluster Google Cloud Apache Kafka for BigQuery, yang memungkinkan autentikasi menggunakan kredensial default aplikasi.
  • messageFormat: Format pesan Kafka yang akan dibaca. Nilai yang didukung adalah AVRO_CONFLUENT_WIRE_FORMAT (Avro yang dienkode Confluent Schema Registry), AVRO_BINARY_ENCODING (Avro biner biasa), dan JSON. Default-nya adalah: AVRO_CONFLUENT_WIRE_FORMAT.
  • useBigQueryDLQ: Jika true (benar), pesan yang gagal akan ditulis ke BigQuery dengan informasi error tambahan. Defaultnya adalah: false.

Parameter opsional

  • outputTableSpec: Lokasi tabel BigQuery tempat output akan ditulis. Nama harus dalam format <project>:<dataset>.<table_name>. Skema tabel harus cocok dengan objek input.
  • persistKafkaKey: Jika true, pipeline akan mempertahankan kunci pesan Kafka di tabel BigQuery, di kolom _key dari jenis BYTES. Default-nya adalah false (Kunci diabaikan).
  • outputProject: Project output BigQuery tempat set data berada. Tabel akan dibuat secara dinamis dalam set data. Default-nya adalah kosong.
  • outputDataset: Set data output BigQuery tempat output akan ditulis. Tabel akan dibuat secara dinamis dalam set data. Jika tabel dibuat sebelumnya, nama tabel harus mengikuti konvensi penamaan yang ditentukan. Namanya harus bqTableNamePrefix + Avro Schema FullName , setiap kata akan dipisahkan dengan tanda hubung -. Default-nya adalah kosong.
  • bqTableNamePrefix: Awalan penamaan yang akan digunakan saat membuat tabel output BigQuery. Hanya berlaku saat menggunakan registry skema. Default-nya adalah kosong.
  • createDisposition: CreateDisposition BigQuery. Misalnya: CREATE_IF_NEEDED, CREATE_NEVER. Defaultnya adalah: CREATE_IF_NEEDED.
  • writeDisposition: BigQuery WriteDisposition. Misalnya: WRITE_APPEND, WRITE_EMPTY, atau WRITE_TRUNCATE. Default-nya adalah: WRITE_APPEND.
  • useAutoSharding: Jika benar, pipeline akan menggunakan sharding otomatis saat menulis ke BigQuery. Nilai defaultnya adalah true.
  • numStorageWriteApiStreams: Menentukan jumlah aliran tulis, parameter ini harus ditetapkan. Default-nya adalah 0.
  • storageWriteApiTriggeringFrequencySec: Menentukan frekuensi pemicuan dalam detik, parameter ini harus ditetapkan. Defaultnya adalah 5 detik.
  • useStorageWriteApiAtLeastOnce: Parameter ini hanya berlaku jika "Gunakan BigQuery Storage Write API" diaktifkan. Jika diaktifkan, semantik minimal satu kali akan digunakan untuk Storage Write API, jika tidak, semantik tepat satu kali akan digunakan. Defaultnya adalah: false.
  • enableCommitOffsets: Menyimpan offset pesan yang diproses ke Kafka. Jika diaktifkan, hal ini akan meminimalkan kesenjangan atau pemrosesan duplikat pesan saat memulai ulang pipeline. Memerlukan penentuan ID Grup Konsumen. Defaultnya adalah: false.
  • consumerGroupId: ID unik untuk grup konsumen tempat pipeline ini berada. Wajib jika Commit Offsets to Kafka diaktifkan. Default-nya adalah kosong.
  • kafkaReadOffset: Titik awal untuk membaca pesan saat tidak ada offset yang di-commit. Pesan terlama dimulai dari awal, pesan terbaru dimulai dari pesan terbaru. Default-nya adalah: latest.
  • kafkaReadUsernameSecretId: ID secret Google Cloud Secret Manager yang berisi nama pengguna Kafka yang akan digunakan dengan autentikasi SASL_PLAIN. Misalnya, projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>. Default-nya adalah kosong.
  • kafkaReadPasswordSecretId: ID secret Google Cloud Secret Manager yang berisi sandi Kafka yang akan digunakan dengan autentikasi SASL_PLAIN. Misalnya, projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>. Default-nya adalah kosong.
  • kafkaReadKeystoreLocation: Jalur Google Cloud Storage ke file Java KeyStore (JKS) yang berisi sertifikat TLS dan kunci pribadi yang akan digunakan saat mengautentikasi dengan cluster Kafka. Contoh, gs://your-bucket/keystore.jks.
  • kafkaReadTruststoreLocation: Jalur Google Cloud Storage ke file Java TrustStore (JKS) yang berisi sertifikat tepercaya yang akan digunakan untuk memverifikasi identitas broker Kafka.
  • kafkaReadTruststorePasswordSecretId: ID secret Google Cloud Secret Manager yang berisi sandi yang akan digunakan untuk mengakses file Java TrustStore (JKS) untuk autentikasi TLS Kafka. Misalnya, projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>.
  • kafkaReadKeystorePasswordSecretId: ID secret Google Cloud Secret Manager yang berisi sandi yang akan digunakan untuk mengakses file Java KeyStore (JKS) untuk autentikasi TLS Kafka. Contoh, projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>.
  • kafkaReadKeyPasswordSecretId: ID secret Google Cloud Secret Manager yang berisi sandi yang akan digunakan untuk mengakses kunci pribadi dalam file Java KeyStore (JKS) untuk autentikasi TLS Kafka. Contoh, projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>.
  • schemaFormat: Format skema Kafka. Dapat diberikan sebagai SINGLE_SCHEMA_FILE atau SCHEMA_REGISTRY. Jika SINGLE_SCHEMA_FILE ditentukan, gunakan skema yang disebutkan dalam file skema avro untuk semua pesan. Jika SCHEMA_REGISTRY ditentukan, pesan dapat memiliki satu skema atau beberapa skema. Setelan defaultnya adalah: SINGLE_SCHEMA_FILE.
  • confluentAvroSchemaPath: Jalur Google Cloud Storage ke satu file skema Avro yang digunakan untuk mendekode semua pesan dalam topik. Default-nya adalah kosong.
  • schemaRegistryConnectionUrl: URL untuk instance Confluent Schema Registry yang digunakan untuk mengelola skema Avro untuk decoding pesan. Default-nya adalah kosong.
  • binaryAvroSchemaPath: Jalur Google Cloud Storage ke file skema Avro yang digunakan untuk mendekode pesan Avro yang dienkode biner. Default-nya adalah kosong.
  • schemaRegistryAuthenticationMode: Mode autentikasi Schema Registry. Dapat berupa NONE, TLS, atau OAUTH. Defaultnya adalah: NONE.
  • schemaRegistryTruststoreLocation: Lokasi sertifikat SSL tempat trust store untuk autentikasi ke Schema Registry disimpan. Contoh, /your-bucket/truststore.jks.
  • schemaRegistryTruststorePasswordSecretId: SecretId di secret manager tempat sandi untuk mengakses secret di truststore disimpan. Contoh, projects/your-project-number/secrets/your-secret-name/versions/your-secret-version.
  • schemaRegistryKeystoreLocation: Lokasi keystore yang berisi sertifikat SSL dan kunci pribadi. Contoh, /your-bucket/keystore.jks.
  • schemaRegistryKeystorePasswordSecretId: SecretId di secret manager tempat sandi untuk mengakses file keystore. Misalnya, projects/your-project-number/secrets/your-secret-name/versions/your-secret-version.
  • schemaRegistryKeyPasswordSecretId: SecretId sandi yang diperlukan untuk mengakses kunci pribadi klien yang disimpan dalam keystore. Misalnya, projects/your-project-number/secrets/your-secret-name/versions/your-secret-version.
  • schemaRegistryOauthClientId: Client ID yang digunakan untuk mengautentikasi klien Schema Registry dalam mode OAUTH. Diperlukan untuk format pesan AVRO_CONFLUENT_WIRE_FORMAT.
  • schemaRegistryOauthClientSecretId: ID rahasia Google Cloud Secret Manager yang berisi Client Secret yang akan digunakan untuk mengautentikasi klien Schema Registry dalam mode OAUTH. Diperlukan untuk format pesan AVRO_CONFLUENT_WIRE_FORMAT. Contoh, projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>.
  • schemaRegistryOauthScope: Cakupan token akses yang digunakan untuk mengautentikasi klien Schema Registry dalam mode OAUTH. Kolom ini bersifat opsional, karena permintaan dapat dilakukan tanpa parameter cakupan yang diteruskan. Contoh, openid.
  • schemaRegistryOauthTokenEndpointUrl: URL berbasis HTTP(S) untuk penyedia identitas OAuth/OIDC yang digunakan untuk mengautentikasi klien Schema Registry dalam mode OAUTH. Diperlukan untuk format pesan AVRO_CONFLUENT_WIRE_FORMAT.
  • outputDeadletterTable: Nama tabel BigQuery yang sepenuhnya memenuhi syarat untuk pesan yang gagal. Pesan yang gagal mencapai tabel output karena berbagai alasan (misalnya, skema yang tidak cocok, JSON yang salah format) akan ditulis ke tabel ini. Tabel akan dibuat oleh template. Contoh, your-project-id:your-dataset.your-table-name.
  • javascriptTextTransformGcsPath: URI Cloud Storage file .js yang menentukan fungsi yang ditentukan pengguna (UDF) JavaScript yang akan digunakan. Contoh, gs://my-bucket/my-udfs/my_file.js.
  • javascriptTextTransformFunctionName: Nama fungsi yang ditentukan pengguna (UDF) JavaScript yang akan digunakan. Misalnya, jika kode fungsi JavaScript Anda adalah myTransform(inJson) { /*...do stuff...*/ }, nama fungsinya adalah myTransform. Untuk contoh UDF JavaScript, lihat Contoh UDF (https://github.com/GoogleCloudPlatform/DataflowTemplates#udf-examples).
  • javascriptTextTransformReloadIntervalMinutes: Menentukan seberapa sering UDF dimuat ulang, dalam hitungan menit. Jika nilainya lebih besar dari 0, Dataflow akan memeriksa file UDF di Cloud Storage secara berkala, dan memuat ulang UDF jika file diubah. Parameter ini memungkinkan Anda mengupdate UDF saat pipeline berjalan, tanpa perlu memulai ulang tugas. Jika nilainya adalah 0, pemuatan ulang UDF akan dinonaktifkan. Nilai defaultnya adalah 0.

Menjalankan template

Konsol

  1. Buka halaman Create job from template Dataflow.
    2. Buka Buat tugas dari template
  2. Di kolom Nama tugas, masukkan nama tugas yang unik.
  3. Opsional: Untuk Endpoint regional, pilih nilai dari menu drop-down. Region defaultnya adalah us-central1.

    Untuk mengetahui daftar region tempat Anda dapat menjalankan tugas Dataflow, lihat Lokasi Dataflow.

  4. Dari menu drop-down Dataflow template, pilih the Kafka to BigQuery template.
  5. Di kolom parameter yang disediakan, masukkan nilai parameter Anda.
  6. Opsional: Untuk beralih dari pemrosesan tepat satu kali ke mode streaming setidaknya sekali, pilih Setidaknya Sekali.
  7. Klik Run job.

gcloud

Di shell atau terminal, jalankan template:

gcloud dataflow flex-template run JOB_NAME \
    --project=PROJECT_ID \
    --region=REGION_NAME \
    --template-file-gcs-location=gs://dataflow-templates-REGION_NAME/VERSION/flex/Kafka_to_BigQuery_Flex \
    --parameters \
outputTableSpec=BIGQUERY_TABLE,\
inputTopics=KAFKA_TOPICS,\
javascriptTextTransformGcsPath=PATH_TO_JAVASCRIPT_UDF_FILE,\
javascriptTextTransformFunctionName=JAVASCRIPT_FUNCTION,\
bootstrapServers=KAFKA_SERVER_ADDRESSES

Ganti kode berikut:

  • PROJECT_ID: ID project Google Cloud tempat Anda ingin menjalankan tugas Dataflow
  • JOB_NAME: nama tugas unik pilihan Anda
  • REGION_NAME: region tempat Anda ingin men-deploy tugas Dataflow—misalnya, us-central1
  • VERSION: versi template yang ingin Anda gunakan

    Anda dapat menggunakan nilai berikut:

  • BIGQUERY_TABLE: nama tabel BigQuery Anda
  • KAFKA_TOPICS: daftar topik Apache Kakfa. Jika ada beberapa topik yang diberikan, Anda harus meng-escape koma. Lihat gcloud topic escaping.
  • PATH_TO_JAVASCRIPT_UDF_FILE: URI Cloud Storage dari file .js yang menentukan fungsi yang ditentukan pengguna (UDF) JavaScript yang ingin Anda gunakan—misalnya, gs://my-bucket/my-udfs/my_file.js
  • JAVASCRIPT_FUNCTION: nama fungsi yang ditentukan pengguna (UDF) JavaScript yang ingin Anda gunakan

    Misalnya, jika kode fungsi JavaScript Anda adalah myTransform(inJson) { /*...do stuff...*/ }, nama fungsinya adalah myTransform. Untuk contoh UDF JavaScript, lihat Contoh UDF.

  • KAFKA_SERVER_ADDRESSES: daftar alamat IP server broker Apache Kafka. Setiap alamat IP memerlukan nomor port tempat server dapat diakses. Contoh: 35.70.252.199:9092. Jika beberapa alamat diberikan, Anda harus meng-escape koma. Lihat gcloud topic escaping.

API

Untuk menjalankan template menggunakan REST API, kirim permintaan POST HTTP. Untuk mengetahui informasi selengkapnya tentang API dan cakupan otorisasinya, lihat projects.templates.launch.

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/flexTemplates:launch
{
   "launch_parameter": {
      "jobName": "JOB_NAME",
      "parameters": {
          "outputTableSpec": "BIGQUERY_TABLE",
          "inputTopics": "KAFKA_TOPICS",
          "javascriptTextTransformGcsPath": "PATH_TO_JAVASCRIPT_UDF_FILE",
          "javascriptTextTransformFunctionName": "JAVASCRIPT_FUNCTION",
          "bootstrapServers": "KAFKA_SERVER_ADDRESSES"
      },
      "containerSpecGcsPath": "gs://dataflow-templates-LOCATION/VERSION/flex/Kafka_to_BigQuery_Flex",
   }
}

Ganti kode berikut:

  • PROJECT_ID: ID project Google Cloud tempat Anda ingin menjalankan tugas Dataflow
  • JOB_NAME: nama tugas unik pilihan Anda
  • LOCATION: region tempat Anda ingin men-deploy tugas Dataflow—misalnya, us-central1
  • VERSION: versi template yang ingin Anda gunakan

    Anda dapat menggunakan nilai berikut:

  • BIGQUERY_TABLE: nama tabel BigQuery Anda
  • KAFKA_TOPICS: daftar topik Apache Kakfa. Jika ada beberapa topik yang diberikan, Anda harus meng-escape koma. Lihat gcloud topic escaping.
  • PATH_TO_JAVASCRIPT_UDF_FILE: URI Cloud Storage dari file .js yang menentukan fungsi yang ditentukan pengguna (UDF) JavaScript yang ingin Anda gunakan—misalnya, gs://my-bucket/my-udfs/my_file.js
  • JAVASCRIPT_FUNCTION: nama fungsi yang ditentukan pengguna (UDF) JavaScript yang ingin Anda gunakan

    Misalnya, jika kode fungsi JavaScript Anda adalah myTransform(inJson) { /*...do stuff...*/ }, nama fungsinya adalah myTransform. Untuk contoh UDF JavaScript, lihat Contoh UDF.

  • KAFKA_SERVER_ADDRESSES: daftar alamat IP server broker Apache Kafka. Setiap alamat IP memerlukan nomor port tempat server dapat diakses. Contoh: 35.70.252.199:9092. Jika beberapa alamat diberikan, Anda harus meng-escape koma. Lihat gcloud topic escaping.

Untuk mengetahui informasi selengkapnya, lihat Menulis data dari Kafka ke BigQuery dengan Dataflow.

Langkah berikutnya