Google 会使用 AI 技术将内容翻译成您偏好的语言。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 数据集。

所需的角色和权限

如需获得创建连接器所需的权限，请让您的管理员为您授予项目的 Managed Kafka Connector Editor (roles/managedkafka.connectorEditor) IAM 角色。如需详细了解如何授予角色，请参阅管理对项目、文件夹和组织的访问权限。

此预定义角色可提供创建连接器所需的权限。如需查看所需的确切权限，请展开所需权限部分：

所需权限

创建连接器需要以下权限：

创建连接器： managedkafka.connectors.create

您也可以使用自定义角色或其他预定义角色来获取这些权限。

授予向 BigQuery 表写入数据的权限

Managed Kafka 服务代理必须具有向 BigQuery 表写入消息的权限。向服务代理授予对包含表的项目的 BigQuery Data Editor (roles/bigquery.dataEditor) 角色。

如果您的 Connect 集群与 Managed Service for Apache Kafka 集群位于不同的项目中，请参阅在其他项目中创建 Connect 集群。

关键配置参数

以下是 BigQuery 接收器连接器的配置建议。

队列大小

queueSize 设置用于控制向 BigQuery 发出的写入请求数上限。默认情况下，此值未配置，这可能会导致 Connect 集群中出现内存不足错误。我们建议您配置此值，以降低发生此类错误的可能性。队列中的内容是写入请求，并不直接对应于消息数或字节数。因此，很难提前知道建议值。

从 Kafka 轮询消息时，连接器会根据目标表拆分工作。如果轮询中的所有消息都要写入单个表，则只会添加一个工作项。但如果轮询中的消息被写入三个表，则会添加三个工作项。工作项的大小可以根据消息轮询配置、Kafka 消息的大小以及要写入的表的数量进行大致估算。

BigQuery 接收器连接器的架构

BigQuery Sink 连接器使用配置的值转换器 (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 将数据存储在客户拥有的存储桶中，以便与 Apache Iceberg 开放表格式进行互操作。

如需了解如何创建 Apache Iceberg 表，请参阅创建 Apache Iceberg 表。

创建 BigQuery 接收器连接器

控制台

在 Google Cloud 控制台中，前往连接集群页面。

前往“连接集群”
点击要在其中创建连接器的 Connect 集群。
点击创建连接器。
对于连接器名称，请输入一个字符串。

有关如何命名连接器的指南，请参阅 Managed Service for Apache Kafka 资源命名指南。
对于连接器插件，选择 BigQuery 接收器。
在主题部分中，指定要从中读取数据的 Kafka 主题。您可以指定主题列表或与主题名称匹配的正则表达式。
- 选项 1：选择选择一个 Kafka 主题列表。在 Kafka 主题列表中，选择一个或多个主题。点击确定。
- 方法 2：选择使用主题正则表达式。在主题正则表达式字段中，输入正则表达式。
点击数据集，然后指定一个 BigQuery 数据集。您可以选择现有数据集，也可以创建新数据集。
可选：在配置框中，添加配置属性或修改默认属性。如需了解详情，请参阅配置连接器。
选择任务重启政策。如需了解详情，请参阅任务重启政策。
点击创建。

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 Sink 连接器。
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 Sink 连接器配置。

Kafka 元数据

您可以通过分别配置 kafkaDataFieldName 和 kafkaKeyFieldName 字段，将 Kafka 中的其他数据（例如元数据信息和键信息）映射到 BigQuery 表中。元数据信息的示例包括 Kafka 主题、分区、偏移量和插入时间。

接下来怎么做？

Apache Kafka® 是 Apache Software Foundation 或其关联公司在美国和/或其他国家/地区的注册商标。

创建 BigQuery Sink 连接器 使用集合让一切井井有条 根据您的偏好保存内容并对其进行分类。