Google 會運用 AI 技術將內容翻譯成你偏好的語言，但可能會出錯。

建立 BigQuery Sink 連接器

BigQuery 接收器連接器可讓您將資料從 Managed Service for Apache Kafka 即時串流至 BigQuery 資料表。您可以使用這個連接器，將串流資料擷取至 BigQuery，用於資料倉儲、分析報表，以及支援即時資訊主頁。

事前準備

建立 BigQuery 接收器連接器前，請確認您具備下列項目：

為 Connect 叢集建立 Managed Service for Apache Kafka 叢集。這個叢集是與 Connect 叢集相關聯的主要 Kafka 叢集。這個叢集也是來源叢集，構成 BigQuery 接收器連接器管道的一端。
建立 Connect 叢集，用於託管 BigQuery 接收器連接器。
建立 BigQuery 資料集，儲存從 Kafka 串流的資料。
在來源叢集中建立及設定 Kafka 主題。資料會從這個 Kafka 主題移至目的地 BigQuery 資料集。

必要角色和權限

如要取得建立連接器所需的權限，請要求管理員授予您專案的代管 Kafka 連接器編輯者 (roles/managedkafka.connectorEditor) IAM 角色。如要進一步瞭解如何授予角色，請參閱「管理專案、資料夾和組織的存取權」。

這個預先定義的角色具備建立連接器所需的權限。如要查看確切的必要權限，請展開「Required permissions」(必要權限) 部分：

所需權限

您必須具備下列權限，才能建立連接器：

建立連接器： managedkafka.connectors.create

您或許還可透過自訂角色或其他預先定義的角色取得這些權限。

授予 BigQuery 資料表的寫入權限

代管 Kafka 服務代理必須具備將訊息寫入 BigQuery 資料表的權限。在資料表所在的專案中，將 BigQuery 資料編輯者 (roles/bigquery.dataEditor) 角色授予服務代理。

如果 Connect 叢集與 Managed Service for Apache Kafka 叢集位於不同專案，請參閱「在其他專案中建立 Connect 叢集」。

重要設定參數

以下是 BigQuery 接收器連接器的設定建議。

佇列大小

queueSize 設定可控管寫入 BigQuery 的要求數量上限。根據預設，系統不會設定這個值，這可能會導致 Connect 叢集發生記憶體不足錯誤。建議您設定這個值，降低發生這類錯誤的機率。佇列中的項目是寫入要求，不會直接對應到訊息或位元組數量。因此，建議值很難事先得知。

從 Kafka 輪詢訊息時，連接器會根據目的地資料表分割工作。如果輪詢中的所有訊息都要寫入單一資料表，則只會新增一個工作項目。不過，如果輪詢中的訊息寫入三個資料表，則會新增三個工作項目。工作項目的大小可根據訊息輪詢設定、Kafka 訊息大小，以及寫入的資料表數量大致估算。

BigQuery 接收器連接器的結構定義

BigQuery 接收器連接器會使用設定的值轉換器 (value.converter)，將 Kafka 記錄值剖析為欄位。然後將欄位寫入 BigQuery 資料表中名稱相同的資料欄。

連接器需要結構定義才能運作。提供結構定義的方式如下：

訊息式結構定義：結構定義會包含在每則訊息中。
資料表結構定義：連接器會根據 BigQuery 資料表結構定義推論訊息結構定義。
結構定義儲存庫：連接器會從結構定義儲存庫讀取結構定義，例如 Managed Service for Apache Kafka 結構定義儲存庫 (預先發布版)。

以下各節將說明這些選項。

訊息式結構定義

在這個模式中，每筆 Kafka 記錄都包含 JSON 結構定義。連接器會使用結構定義，將記錄資料寫入 BigQuery 資料表資料列。

如要使用訊息式結構定義，請在連接器上設定下列屬性：

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

Kafka 記錄值範例：

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

如果目的地資料表已存在，BigQuery 資料表結構定義必須與內嵌訊息結構定義相容。如果為 autoCreateTables=true，連接器會在必要時自動建立目的地資料表。詳情請參閱「建立資料表」。

如要讓連接器在訊息結構定義變更時更新 BigQuery 資料表結構定義，請將 allowNewBigQueryFields、allowSchemaUnionization 或 allowBigQueryRequiredFieldRelaxation 設為 true。

資料表結構定義

在這種模式下，Kafka 記錄包含純 JSON 資料，沒有明確的結構定義。連接器會從目的地資料表推論結構定義。

需求條件：

BigQuery 資料表必須已存在。
Kafka 記錄資料必須與資料表結構定義相容。
這個模式不支援根據收到的訊息動態更新結構定義。

如要使用資料表架構，請在連接器上設定下列屬性：

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

如果 BigQuery 資料表使用時間分區，並以每日分區，則 bigQueryPartitionDecorator 可以是 true。否則，請將這項屬性設為 false。

Kafka 記錄值範例：

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

結構定義儲存庫

在這種模式下，每筆 Kafka 記錄都包含 Apache Avro 資料，且訊息結構定義會儲存在結構定義儲存庫中。

如要搭配使用 BigQuery 接收器連接器和結構定義登錄檔，請在連接器上設定下列屬性：

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

將 SCHEMA_REGISTRY_URL 替換為結構定義登錄檔的網址。

如要搭配使用連接器和 Managed Service for Apache Kafka 結構定義儲存庫，請設定下列屬性：

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

詳情請參閱「搭配使用 Kafka Connect 和結構定義登錄檔」。

Apache Iceberg 代管資料表

BigQuery Sink 連接器支援 Apache Iceberg 代管資料表 (以下簡稱 Iceberg 代管資料表) 做為接收器目標。

Iceberg 代管資料表是 Google Cloud上建構開放格式湖倉的基礎。Iceberg 代管資料表提供與 BigQuery 資料表相同的全代管體驗，但會使用 Parquet 將資料儲存在客戶自有的儲存空間 bucket，以便與 Apache Iceberg 開放式資料表格式互通。

如要瞭解如何建立 Apache Iceberg 資料表，請參閱「建立 Apache Iceberg 資料表」。

建立 BigQuery 接收器連接器

控制台

前往 Google Cloud 控制台的「Connect Clusters」(連結叢集) 頁面。

前往「Connect Clusters」(連結叢集)
按一下要建立連接器的 Connect 叢集。
按一下「Create connector」(建立連接器)。
輸入連接器名稱字串。

如要查看連接器命名準則，請參閱「Managed Service for Apache Kafka 資源命名指南」。
在「Connector plugin」(連接器外掛程式) 部分，選取「BigQuery Sink」(BigQuery 接收器)。
在「Topics」(主題) 專區中，指定要讀取的 Kafka 主題。您可以指定主題清單，或指定要與主題名稱比對的規則運算式。
- 選項 1：選擇「選取 Kafka 主題清單」。在「Kafka topics」(Kafka 主題) 清單中，選取一或多個主題。然後點選「OK」。
- 選項 2：選擇「使用主題規則運算式」。在「Topic regex」(主題規則運算式) 欄位中，輸入規則運算式。
按一下「Dataset」(資料集)，然後指定 BigQuery 資料集。您可以選擇現有資料集，也可以建立新的資料集。
選用：在「Configurations」(設定) 方塊中，新增設定屬性或編輯預設屬性。詳情請參閱「設定連接器」。
選取「Task restart policy」(任務重新啟動政策)。詳情請參閱「工作重新啟動政策」。
點選「建立」。

gcloud

執行 gcloud managed-kafka connectors create 指令：

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

更改下列內容：

CONNECTOR_ID：連接器的 ID 或名稱。
CONNECT_CLUSTER_ID：建立連接器的 Connect 叢集 ID。
LOCATION：Connect 叢集的位置。
CONFIG_FILE：本機 YAML 或 JSON 設定檔的路徑。

以下是 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"

更改下列內容：

BQ_SINK_CONNECTOR_ID：BigQuery Sink 連接器的 ID 或名稱。如要查看連接器命名準則，請參閱「Managed Service for Apache Kafka 資源命名指南」。連接器名稱無法變更。
GCP_PROJECT_ID：BigQuery 資料集所在 Google Cloud專案的 ID。
GMK_TOPIC_ID：Managed Service for Apache Kafka 主題的 ID，資料會從該主題流向 BigQuery 接收器連接器。
BQ_DATASET_ID：BigQuery 資料集的 ID，做為管道的接收器。

Terraform

您可以使用 Terraform 資源建立連接器。

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
}

如要瞭解如何套用或移除 Terraform 設定，請參閱「基本 Terraform 指令」。

Go

在試用這個範例之前，請先按照「安裝用戶端程式庫」中的 Go 設定操作說明進行操作。詳情請參閱 Managed Service for Apache Kafka Go API 參考文件。

如要向 Managed Service for Apache Kafka 進行驗證，請設定應用程式預設憑證(ADC)。詳情請參閱「為本機開發環境設定 ADC」。

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

在試用這個範例之前，請先按照「安裝用戶端程式庫」中的 Java 設定操作說明進行操作。詳情請參閱 Managed Service for Apache Kafka Java API 參考文件。

如要向 Managed Service for Apache Kafka 進行驗證，請設定應用程式預設憑證。詳情請參閱「為本機開發環境設定 ADC」。


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

試用這個範例前，請先按照「安裝用戶端程式庫」中的 Python 設定操作說明進行。詳情請參閱 Managed Service for Apache Kafka Python API 參考文件。

如要向 Managed Service for Apache Kafka 進行驗證，請設定應用程式預設憑證。詳情請參閱「為本機開發環境設定 ADC」。

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}")

建立連接器後，您可以編輯、刪除、暫停、停止或重新啟動連接器。

設定連接器

本節說明您可以在連接器上設定的部分設定屬性。如需這個連接器專屬屬性的完整清單，請參閱「BigQuery Sink 連接器設定」。

資料表名稱

根據預設，連接器會使用主題名稱做為 BigQuery 資料表名稱。如要使用其他資料表名稱，請按照下列格式設定 topic2TableMap 屬性：

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

建立資料表

如果目的地資料表不存在，BigQuery 接收器連接器可以建立這些資料表。

如果 autoCreateTables=true，連接器會嘗試建立不存在的 BigQuery 資料表。這是預設行為。
如果為 autoCreateTables=false，連接器不會建立任何資料表。如果目的地資料表不存在，則會發生錯誤。

當 autoCreateTables 為 true 時，您可以使用下列設定屬性，更精細地控管連接器建立及設定新資料表的方式：

allBQFieldsNullable
clusteringPartitionFieldNames
convertDoubleSpecialValues
partitionExpirationMs
sanitizeFieldNames
sanitizeTopics
timestampPartitionFieldName

如要瞭解這些屬性，請參閱「BigQuery 接收器連接器設定」。

Kafka 中繼資料

您可以分別設定 kafkaDataFieldName 和 kafkaKeyFieldName 欄位，將 Kafka 的其他資料 (例如中繼資料資訊和金鑰資訊) 對應至 BigQuery 資料表。中繼資料資訊的例子包括 Kafka 主題、分區、偏移和插入時間。

後續步驟

Apache Kafka® 是 Apache Software Foundation 或其關係企業在美國和/或其他國家/地區的註冊商標。

建立 BigQuery Sink 連接器 透過集合功能整理內容 你可以依據偏好儲存及分類內容。