Membuat konektor Sink BigQuery

Konektor BigQuery Sink memungkinkan Anda melakukan streaming data dari Managed Service untuk Apache Kafka ke tabel BigQuery secara real time. Anda dapat menggunakan konektor ini untuk menyerap data streaming ke BigQuery untuk data warehousing, pelaporan analisis, dan mendukung dasbor real-time.

Sebelum memulai

Sebelum membuat konektor BigQuery Sink, pastikan Anda memiliki hal berikut:

Peran dan izin yang diperlukan

Untuk mendapatkan izin yang diperlukan guna membuat konektor, minta administrator untuk memberi Anda peran IAM Managed Kafka Connector Editor (roles/managedkafka.connectorEditor) di project Anda. Untuk mengetahui informasi selengkapnya tentang cara memberikan peran, lihat Mengelola akses ke project, folder, dan organisasi.

Peran bawaan ini berisi izin yang diperlukan untuk membuat konektor. Untuk melihat izin yang benar-benar diperlukan, perluas bagian Izin yang diperlukan:

Izin yang diperlukan

Izin berikut diperlukan untuk membuat konektor:

  • Buat konektor: managedkafka.connectors.create

Anda mungkin juga bisa mendapatkan izin ini dengan peran khusus atau peran bawaan lainnya.

Memberikan izin untuk menulis ke tabel BigQuery

Agen Layanan Kafka Terkelola harus memiliki izin untuk menulis pesan ke tabel BigQuery. Berikan peran BigQuery Data Editor (roles/bigquery.dataEditor) kepada agen layanan pada project yang berisi tabel.

Jika cluster Connect Anda berada di project yang berbeda dengan cluster Managed Service untuk Apache Kafka, lihat Membuat cluster Connect di project yang berbeda.

Parameter konfigurasi utama

Berikut adalah rekomendasi konfigurasi untuk konektor sink BigQuery.

Ukuran antrean

Setelan queueSize mengontrol jumlah maksimum permintaan tulis ke BigQuery. Secara default, nilai ini tidak dikonfigurasi, yang dapat mengakibatkan error kehabisan memori di cluster Connect. Sebaiknya konfigurasikan nilai ini untuk mengurangi kemungkinan error tersebut. Item dalam antrean adalah permintaan tulis, dan tidak sesuai secara langsung dengan jumlah pesan atau byte. Oleh karena itu, nilai yang direkomendasikan sulit diketahui sebelumnya.

Saat pesan di-polling dari Kafka, konektor akan membagi pekerjaan berdasarkan tabel tujuannya. Jika semua pesan dalam polling akan ditulis ke satu tabel, hanya satu item kerja yang akan ditambahkan. Namun, jika pesan dalam polling ditulis ke tiga tabel, misalnya, tiga item pekerjaan akan ditambahkan. Ukuran item tugas dapat diperkirakan secara kasar berdasarkan konfigurasi polling pesan, ukuran pesan Kafka, dan jumlah tabel yang ditulis.

Skema untuk konektor BigQuery Sink

Konektor BigQuery Sink menggunakan pengonversi nilai yang dikonfigurasi (value.converter) untuk mengurai nilai kumpulan data Kafka ke dalam kolom. Kemudian, pipeline akan menulis kolom ke kolom dengan nama yang sama di tabel BigQuery.

Konektor memerlukan skema untuk beroperasi. Skema dapat diberikan dengan cara berikut:

Bagian berikutnya menjelaskan opsi ini.

Skema berbasis pesan

Dalam mode ini, setiap data Kafka menyertakan skema JSON. Konektor menggunakan skema untuk menulis data kumpulan data sebagai baris tabel BigQuery.

Untuk menggunakan skema berbasis pesan, tetapkan properti berikut pada konektor:

  • value.converter=org.apache.kafka.connect.json.JsonConverter
  • value.converter.schemas.enable=true

Contoh nilai data Kafka:

{
  "schema": {
    "type": "struct",
    "fields": [
      {
        "field": "user",
        "type": "string",
        "optional": false
      },
      {
        "field": "age",
        "type": "int64",
        "optional": false
      }
    ]
  },
  "payload": {
    "user": "userId",
    "age": 30
  }
}

Jika tabel tujuan sudah ada, skema tabel BigQuery harus kompatibel dengan skema pesan tersemat. Jika autoCreateTables=true, konektor akan otomatis membuat tabel tujuan jika diperlukan. Untuk mengetahui informasi selengkapnya, lihat Pembuatan tabel.

Jika Anda ingin konektor memperbarui skema tabel BigQuery saat skema pesan berubah, tetapkan allowNewBigQueryFields, allowSchemaUnionization, atau allowBigQueryRequiredFieldRelaxation ke true.

Skema berbasis tabel

Dalam mode ini, kumpulan data Kafka berisi data JSON biasa tanpa skema eksplisit. Konektor menyimpulkan skema dari tabel tujuan.

Persyaratan:

  • Tabel BigQuery harus sudah ada.
  • Data kumpulan data Kafka harus kompatibel dengan skema tabel.
  • Mode ini tidak mendukung pembaruan skema dinamis berdasarkan pesan masuk.

Untuk menggunakan skema berbasis tabel, tetapkan properti berikut pada konektor:

  • value.converter=org.apache.kafka.connect.json.JsonConverter
  • value.converter.schemas.enable=false
  • bigQueryPartitionDecorator=false

Jika tabel BigQuery menggunakan partisi berbasis waktu dengan partisi harian, bigQueryPartitionDecorator dapat berupa true. Jika tidak, tetapkan properti ini ke false.

Contoh nilai data Kafka:

{
  "user": "userId",
  "age": 30
}

Registry skema

Dalam mode ini, setiap data Kafka berisi data Apache Avro, dan skema pesan disimpan dalam registry skema.

Untuk menggunakan konektor BigQuery Sink dengan registry skema, tetapkan properti berikut pada konektor:

  • value.converter=io.confluent.connect.avro.AvroConverter
  • value.converter.schema.registry.url=SCHEMA_REGISTRY_URL

Ganti SCHEMA_REGISTRY_URL dengan URL registry skema.

Untuk menggunakan konektor dengan registry skema Managed Service untuk Apache Kafka, tetapkan properti berikut:

  • value.converter.bearer.auth.credentials.source=GCP

Untuk mengetahui informasi selengkapnya, lihat Menggunakan Kafka Connect dengan registry skema.

Tabel terkelola Apache Iceberg

Konektor BigQuery Sink mendukung tabel terkelola Apache Iceberg (selanjutnya disebut tabel terkelola Iceberg) sebagai target sink.

Tabel terkelola Iceberg menyediakan fondasi untuk membangun lakehouse format terbuka di Google Cloud. Tabel terkelola Iceberg menawarkan pengalaman terkelola sepenuhnya yang sama seperti tabel BigQuery, tetapi menyimpan data di bucket penyimpanan milik pelanggan menggunakan Parquet agar dapat dioperasikan dengan format tabel terbuka Apache Iceberg.

Untuk mengetahui informasi tentang cara membuat tabel Apache Iceberg, lihat Membuat tabel Apache Iceberg.

Membuat konektor BigQuery Sink

Konsol

  1. Di konsol Google Cloud , buka halaman Connect Clusters.

    Buka Connect Clusters

  2. Klik cluster Connect tempat Anda ingin membuat konektor.

  3. Klik Create connector.

  4. Untuk nama konektor, masukkan string.

    Untuk panduan tentang cara memberi nama konektor, lihat Panduan untuk memberi nama resource Managed Service untuk Apache Kafka.

  5. Untuk Connector plugin, pilih BigQuery Sink.

  6. Di bagian Topics, tentukan topik Kafka yang akan dibaca. Anda dapat menentukan daftar topik atau ekspresi reguler agar sesuai dengan nama topik.

    • Opsi 1: Pilih Select a list of Kafka topics. Dalam daftar Kafka topics, pilih satu atau beberapa topik. Klik OK.

    • Opsi 2: Pilih Use a topic regex. Di kolom Topic regex, masukkan ekspresi reguler.

  7. Klik Dataset dan tentukan set data BigQuery. Anda dapat memilih set data yang ada atau membuat set data baru.

  8. Opsional: Di kotak Configurations, tambahkan properti konfigurasi atau edit properti default. Untuk mengetahui informasi selengkapnya, lihat Mengonfigurasi konektor.

  9. Pilih Task restart policy. Untuk mengetahui informasi selengkapnya, lihat Kebijakan memulai ulang tugas.

  10. Klik Create.

gcloud

Jalankan perintah gcloud managed-kafka connectors create:

gcloud managed-kafka connectors create CONNECTOR_ID \
    --connect-cluster=CONNECT_CLUSTER_ID \
    --location=LOCATION \
    --config-file=CONFIG_FILE

Ganti kode berikut:

  • CONNECTOR_ID: ID atau nama konektor.
  • CONNECT_CLUSTER_ID: ID cluster Connect tempat konektor dibuat.
  • LOCATION: Lokasi cluster Connect.
  • CONFIG_FILE: Jalur ke file konfigurasi YAML atau JSON lokal.

Berikut adalah contoh file konfigurasi untuk konektor BigQuery Sink:

name: "BQ_SINK_CONNECTOR_ID"
project: "GCP_PROJECT_ID"
topics: "GMK_TOPIC_ID"
tasks.max: 3
connector.class: "com.wepay.kafka.connect.bigquery.BigQuerySinkConnector"
key.converter: "org.apache.kafka.connect.storage.StringConverter"
value.converter: "org.apache.kafka.connect.json.JsonConverter"
value.converter.schemas.enable: "false"
defaultDataset: "BQ_DATASET_ID"

Ganti kode berikut:

  • BQ_SINK_CONNECTOR_ID: ID atau nama konektor BigQuery Sink. Untuk panduan tentang cara memberi nama konektor, lihat Panduan untuk memberi nama resource Managed Service untuk Apache Kafka. Nama konektor tidak dapat diubah.

  • GCP_PROJECT_ID: ID project Google Cloud tempat set data BigQuery Anda berada.

  • GMK_TOPIC_ID: ID topik Managed Service untuk Apache Kafka tempat data mengalir ke konektor BigQuery Sink.

  • BQ_DATASET_ID: ID set data BigQuery yang bertindak sebagai sink untuk pipeline.

Terraform

Anda dapat menggunakan resource Terraform untuk membuat konektor.

resource "google_managed_kafka_connector" "example-bigquery-sink-connector" {
  project         = data.google_project.default.project_id
  connector_id    = "my-bigquery-sink-connector"
  connect_cluster = google_managed_kafka_connect_cluster.default.connect_cluster_id
  location        = "us-central1"

  configs = {
    "name"                           = "my-bigquery-sink-connector"
    "project"                        = data.google_project.default.project_id
    "topics"                         = "GMK_TOPIC_ID"
    "tasks.max"                      = "3"
    "connector.class"                = "com.wepay.kafka.connect.bigquery.BigQuerySinkConnector"
    "key.converter"                  = "org.apache.kafka.connect.storage.StringConverter"
    "value.converter"                = "org.apache.kafka.connect.json.JsonConverter"
    "value.converter.schemas.enable" = "false"
    "defaultDataset"                 = "BQ_DATASET_ID"
  }

  provider = google-beta
}

Untuk mempelajari cara menerapkan atau menghapus konfigurasi Terraform, lihat Perintah dasar Terraform.

Go

Sebelum mencoba contoh ini, ikuti petunjuk penyiapan Go di Menginstal library klien. Untuk mengetahui informasi selengkapnya, lihat dokumentasi referensi Managed Service for Apache Kafka Go API.

Untuk melakukan autentikasi ke Managed Service untuk Apache Kafka, siapkan Kredensial Default Aplikasi(ADC). Untuk mengetahui informasi selengkapnya, lihat Menyiapkan ADC untuk lingkungan pengembangan lokal.

import (
	"context"
	"fmt"
	"io"

	managedkafka "cloud.google.com/go/managedkafka/apiv1"
	"cloud.google.com/go/managedkafka/apiv1/managedkafkapb"
	"google.golang.org/api/option"
)

// createBigQuerySinkConnector creates a BigQuery Sink connector.
func createBigQuerySinkConnector(w io.Writer, projectID, region, connectClusterID, connectorID, topics, tasksMax, keyConverter, valueConverter, valueConverterSchemasEnable, defaultDataset string, opts ...option.ClientOption) error {
	// TODO(developer): Update with your config values. Here is a sample configuration:
	// projectID := "my-project-id"
	// region := "us-central1"
	// connectClusterID := "my-connect-cluster"
	// connectorID := "BQ_SINK_CONNECTOR_ID"
	// topics := "GMK_TOPIC_ID"
	// tasksMax := "3"
	// keyConverter := "org.apache.kafka.connect.storage.StringConverter"
	// valueConverter := "org.apache.kafka.connect.json.JsonConverter"
	// valueConverterSchemasEnable := "false"
	// defaultDataset := "BQ_DATASET_ID"
	ctx := context.Background()
	client, err := managedkafka.NewManagedKafkaConnectClient(ctx, opts...)
	if err != nil {
		return fmt.Errorf("managedkafka.NewManagedKafkaConnectClient got err: %w", err)
	}
	defer client.Close()

	parent := fmt.Sprintf("projects/%s/locations/%s/connectClusters/%s", projectID, region, connectClusterID)

	// BigQuery Sink sample connector configuration
	config := map[string]string{
		"name":                           connectorID,
		"project":                        projectID,
		"topics":                         topics,
		"tasks.max":                      tasksMax,
		"connector.class":                "com.wepay.kafka.connect.bigquery.BigQuerySinkConnector",
		"key.converter":                  keyConverter,
		"value.converter":                valueConverter,
		"value.converter.schemas.enable": valueConverterSchemasEnable,
		"defaultDataset":                 defaultDataset,
	}

	connector := &managedkafkapb.Connector{
		Name:    fmt.Sprintf("%s/connectors/%s", parent, connectorID),
		Configs: config,
	}

	req := &managedkafkapb.CreateConnectorRequest{
		Parent:      parent,
		ConnectorId: connectorID,
		Connector:   connector,
	}

	resp, err := client.CreateConnector(ctx, req)
	if err != nil {
		return fmt.Errorf("client.CreateConnector got err: %w", err)
	}
	fmt.Fprintf(w, "Created BigQuery sink connector: %s\n", resp.Name)
	return nil
}

Java

Sebelum mencoba contoh ini, ikuti petunjuk penyiapan Java di Menginstal library klien. Untuk mengetahui informasi selengkapnya, lihat dokumentasi referensi Managed Service for Apache Kafka Java API.

Untuk melakukan autentikasi ke Managed Service untuk Apache Kafka, siapkan Kredensial Default Aplikasi. Untuk mengetahui informasi selengkapnya, lihat Menyiapkan ADC untuk lingkungan pengembangan lokal.


import com.google.api.gax.rpc.ApiException;
import com.google.cloud.managedkafka.v1.ConnectClusterName;
import com.google.cloud.managedkafka.v1.Connector;
import com.google.cloud.managedkafka.v1.ConnectorName;
import com.google.cloud.managedkafka.v1.CreateConnectorRequest;
import com.google.cloud.managedkafka.v1.ManagedKafkaConnectClient;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

public class CreateBigQuerySinkConnector {

  public static void main(String[] args) throws Exception {
    // TODO(developer): Replace these variables before running the example.
    String projectId = "my-project-id";
    String region = "my-region"; // e.g. us-east1
    String connectClusterId = "my-connect-cluster";
    String connectorId = "my-bigquery-sink-connector";
    String bigqueryProjectId = "my-bigquery-project-id";
    String datasetName = "my-dataset";
    String kafkaTopicName = "kafka-topic";
    String maxTasks = "3";
    String connectorClass = "com.wepay.kafka.connect.bigquery.BigQuerySinkConnector";
    String keyConverter = "org.apache.kafka.connect.storage.StringConverter";
    String valueConverter = "org.apache.kafka.connect.json.JsonConverter";
    String valueSchemasEnable = "false";
    createBigQuerySinkConnector(
        projectId,
        region,
        connectClusterId,
        connectorId,
        bigqueryProjectId,
        datasetName,
        kafkaTopicName,
        maxTasks,
        connectorClass,
        keyConverter,
        valueConverter,
        valueSchemasEnable);
  }

  public static void createBigQuerySinkConnector(
      String projectId,
      String region,
      String connectClusterId,
      String connectorId,
      String bigqueryProjectId,
      String datasetName,
      String kafkaTopicName,
      String maxTasks,
      String connectorClass,
      String keyConverter,
      String valueConverter,
      String valueSchemasEnable)
      throws Exception {

    // Build the connector configuration
    Map<String, String> configMap = new HashMap<>();
    configMap.put("name", connectorId);
    configMap.put("project", bigqueryProjectId);
    configMap.put("topics", kafkaTopicName);
    configMap.put("tasks.max", maxTasks);
    configMap.put("connector.class", connectorClass);
    configMap.put("key.converter", keyConverter);
    configMap.put("value.converter", valueConverter);
    configMap.put("value.converter.schemas.enable", valueSchemasEnable);
    configMap.put("defaultDataset", datasetName);

    Connector connector =
        Connector.newBuilder()
            .setName(ConnectorName.of(projectId, region, connectClusterId, connectorId).toString())
            .putAllConfigs(configMap)
            .build();

    try (ManagedKafkaConnectClient managedKafkaConnectClient = ManagedKafkaConnectClient.create()) {
      CreateConnectorRequest request =
          CreateConnectorRequest.newBuilder()
              .setParent(ConnectClusterName.of(projectId, region, connectClusterId).toString())
              .setConnectorId(connectorId)
              .setConnector(connector)
              .build();

      // This operation is being handled synchronously.
      Connector response = managedKafkaConnectClient.createConnector(request);
      System.out.printf("Created BigQuery Sink connector: %s\n", response.getName());
    } catch (IOException | ApiException e) {
      System.err.printf("managedKafkaConnectClient.createConnector got err: %s\n", e.getMessage());
    }
  }
}

Python

Sebelum mencoba contoh ini, ikuti petunjuk penyiapan Python di Menginstal library klien. Untuk mengetahui informasi selengkapnya, lihat dokumentasi referensi Python API Managed Service untuk Apache Kafka.

Untuk melakukan autentikasi ke Managed Service untuk Apache Kafka, siapkan Kredensial Default Aplikasi. Untuk mengetahui informasi selengkapnya, lihat Menyiapkan ADC untuk lingkungan pengembangan lokal.

from google.api_core.exceptions import GoogleAPICallError
from google.cloud.managedkafka_v1.services.managed_kafka_connect import (
    ManagedKafkaConnectClient,
)
from google.cloud.managedkafka_v1.types import Connector, CreateConnectorRequest

connect_client = ManagedKafkaConnectClient()
parent = connect_client.connect_cluster_path(project_id, region, connect_cluster_id)

configs = {
    "name": connector_id,
    "project": project_id,
    "topics": topics,
    "tasks.max": tasks_max,
    "connector.class": "com.wepay.kafka.connect.bigquery.BigQuerySinkConnector",
    "key.converter": key_converter,
    "value.converter": value_converter,
    "value.converter.schemas.enable": value_converter_schemas_enable,
    "defaultDataset": default_dataset,
}

connector = Connector()
connector.name = connector_id
connector.configs = configs

request = CreateConnectorRequest(
    parent=parent,
    connector_id=connector_id,
    connector=connector,
)

try:
    operation = connect_client.create_connector(request=request)
    print(f"Waiting for operation {operation.operation.name} to complete...")
    response = operation.result()
    print("Created Connector:", response)
except GoogleAPICallError as e:
    print(f"The operation failed with error: {e}")

Setelah membuat konektor, Anda dapat mengedit, menghapus, menjeda, menghentikan, atau memulai ulang konektor.

Mengonfigurasi konektor

Bagian ini menjelaskan beberapa properti konfigurasi yang dapat Anda tetapkan pada konektor. Untuk mengetahui daftar lengkap properti yang khusus untuk konektor ini, lihat Konfigurasi konektor BigQuery Sink.

Nama tabel

Secara default, konektor menggunakan nama topik sebagai nama tabel BigQuery. Untuk menggunakan nama tabel yang berbeda, tetapkan properti topic2TableMap dengan format berikut:

topic2TableMap=TOPIC_1:TABLE_1,TOPIC_2:TABLE_2,...

Pembuatan tabel

Konektor BigQuery Sink dapat membuat tabel tujuan jika tabel tersebut belum ada.

  • Jika autoCreateTables=true, konektor akan mencoba membuat tabel BigQuery yang belum ada. Setelan ini adalah perilaku default.

  • Jika autoCreateTables=false, konektor tidak akan membuat tabel apa pun. Jika tabel tujuan tidak ada, akan terjadi error.

Jika autoCreateTables adalah true, Anda dapat menggunakan properti konfigurasi berikut untuk kontrol yang lebih mendetail atas cara konektor membuat dan mengonfigurasi tabel baru:

  • allBQFieldsNullable
  • clusteringPartitionFieldNames
  • convertDoubleSpecialValues
  • partitionExpirationMs
  • sanitizeFieldNames
  • sanitizeTopics
  • timestampPartitionFieldName

Untuk mengetahui informasi tentang properti ini, lihat Konfigurasi konektor BigQuery Sink.

Metadata Kafka

Anda dapat memetakan data tambahan dari Kafka seperti informasi metadata dan informasi kunci ke dalam tabel BigQuery dengan mengonfigurasi kolom kafkaDataFieldName dan kafkaKeyFieldName. Contoh informasi metadata mencakup topik, partisi, offset, dan waktu penyisipan Kafka.

Apa langkah selanjutnya?

Apache Kafka® adalah merek dagang terdaftar dari Apache Software Foundation atau afiliasinya di Amerika Serikat dan/atau negara lainnya.