创建 Cloud Storage Sink 连接器

Cloud Storage 接收器连接器会将数据从 Managed Service for Apache Kafka 流式传输到 Cloud Storage，以便您以经济实惠且可伸缩的方式存储和处理大量数据。

此连接器的典型应用场景包括：

  • 数据湖注入。将 Kafka 数据存储在数据湖中，以便进行长期归档和批处理。

  • 归档数据以满足监管要求。

准备工作

在创建 Cloud Storage Sink 连接器之前，请确保您具备以下条件：

所需的角色和权限

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

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

所需权限

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

  • 创建连接器： managedkafka.connectors.create

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

授予对 Cloud Storage 存储桶的写入权限

Managed Kafka Service Agent 必须具有以下权限才能将消息写入 Cloud Storage 存储桶：

  • storage.objects.create
  • storage.objects.delete

如需授予这些权限，请向服务代理授予包含 Cloud Storage 存储桶的项目中的 Storage Object User (roles/storage.objectUser) 角色。

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

创建 Cloud Storage 接收器连接器

如需创建 Cloud Storage 接收器连接器，请执行以下步骤。

控制台

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

    前往“连接集群”

  2. 点击要为其创建连接器的 Connect 集群。

    系统会显示连接集群详情页面。

  3. 点击创建连接器。

    系统会显示创建 Kafka 连接器页面。

  4. 对于连接器名称，请输入一个字符串。

    有关如何命名连接器的指南，请参阅 Managed Service for Apache Kafka 资源命名指南。

  5. 在连接器插件部分，选择 Cloud Storage 接收器。

  6. 指定可从中流式传输数据的主题。

  7. 选择用于存储数据的存储桶。

  8. （可选）在配置部分中配置其他设置。

  9. 选择任务重启政策。如需了解详情，请参阅任务重启政策。

  10. 点击创建。

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 配置文件的路径。

以下是 Cloud Storage Sink 连接器的配置文件示例：

connector.class: "io.aiven.kafka.connect.gcs.GcsSinkConnector"
gcs.bucket.name: "BUCKET_NAME"
gcs.credentials.default: "true"
format.output.type: "json"
key.converter: "org.apache.kafka.connect.storage.StringConverter"
value.converter: "org.apache.kafka.connect.json.JsonConverter"
value.converter.schemas.enable: "false"
topics: "TOPIC_LIST"
tasks.max: "3"

替换以下内容：

  • BUCKET_NAME：要写入的 Cloud Storage 存储桶的名称。

  • TOPIC_LIST：要从中读取的 Kafka 主题的英文逗号分隔列表。该列表还可以包含正则表达式。

Terraform

您可以使用 Terraform 资源创建连接器。

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

  configs = {
    "connector.class"                = "io.aiven.kafka.connect.gcs.GcsSinkConnector"
    "tasks.max"                      = "3"
    "topics"                         = "GMK_TOPIC_ID"
    "gcs.bucket.name"                = "GCS_BUCKET_NAME"
    "gcs.credentials.default"        = "true"
    "format.output.type"             = "json"
    "name"                           = "my-gcs-sink-connector"
    "value.converter"                = "org.apache.kafka.connect.json.JsonConverter"
    "value.converter.schemas.enable" = "false"
    "key.converter"                  = "org.apache.kafka.connect.storage.StringConverter"
  }
  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"
)

// createCloudStorageSinkConnector creates a Cloud Storage Sink connector.
func createCloudStorageSinkConnector(w io.Writer, projectID, region, connectClusterID, connectorID, topics, gcsBucketName, tasksMax, formatOutputType, valueConverter, valueConverterSchemasEnable, keyConverter, gcsCredentialsDefault 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 := "GCS_SINK_CONNECTOR_ID"
	// topics := "GMK_TOPIC_ID"
	// gcsBucketName := "GCS_BUCKET_NAME"
	// tasksMax := "3"
	// formatOutputType := "json"
	// valueConverter := "org.apache.kafka.connect.json.JsonConverter"
	// valueConverterSchemasEnable := "false"
	// keyConverter := "org.apache.kafka.connect.storage.StringConverter"
	// gcsCredentialsDefault := "true"
	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)

	config := map[string]string{
		"connector.class":                "io.aiven.kafka.connect.gcs.GcsSinkConnector",
		"tasks.max":                      tasksMax,
		"topics":                         topics,
		"gcs.bucket.name":                gcsBucketName,
		"gcs.credentials.default":        gcsCredentialsDefault,
		"format.output.type":             formatOutputType,
		"name":                           connectorID,
		"value.converter":                valueConverter,
		"value.converter.schemas.enable": valueConverterSchemasEnable,
		"key.converter":                  keyConverter,
	}

	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 Cloud Storage 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 CreateCloudStorageSinkConnector {

  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-gcs-sink-connector";
    String bucketName = "my-gcs-bucket";
    String kafkaTopicName = "kafka-topic";
    String connectorClass = "io.aiven.kafka.connect.gcs.GcsSinkConnector";
    String maxTasks = "3";
    String gcsCredentialsDefault = "true";
    String formatOutputType = "json";
    String valueConverter = "org.apache.kafka.connect.json.JsonConverter";
    String valueSchemasEnable = "false";
    String keyConverter = "org.apache.kafka.connect.storage.StringConverter";
    createCloudStorageSinkConnector(
        projectId,
        region,
        connectClusterId,
        connectorId,
        bucketName,
        kafkaTopicName,
        connectorClass,
        maxTasks,
        gcsCredentialsDefault,
        formatOutputType,
        valueConverter,
        valueSchemasEnable,
        keyConverter);
  }

  public static void createCloudStorageSinkConnector(
      String projectId,
      String region,
      String connectClusterId,
      String connectorId,
      String bucketName,
      String kafkaTopicName,
      String connectorClass,
      String maxTasks,
      String gcsCredentialsDefault,
      String formatOutputType,
      String valueConverter,
      String valueSchemasEnable,
      String keyConverter)
      throws Exception {

    // Build the connector configuration
    Map<String, String> configMap = new HashMap<>();
    configMap.put("connector.class", connectorClass);
    configMap.put("tasks.max", maxTasks);
    configMap.put("topics", kafkaTopicName);
    configMap.put("gcs.bucket.name", bucketName);
    configMap.put("gcs.credentials.default", gcsCredentialsDefault);
    configMap.put("format.output.type", formatOutputType);
    configMap.put("name", connectorId);
    configMap.put("value.converter", valueConverter);
    configMap.put("value.converter.schemas.enable", valueSchemasEnable);
    configMap.put("key.converter", keyConverter);

    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 Cloud Storage 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 = {
    "connector.class": "io.aiven.kafka.connect.gcs.GcsSinkConnector",
    "tasks.max": tasks_max,
    "topics": topics,
    "gcs.bucket.name": gcs_bucket_name,
    "gcs.credentials.default": "true",
    "format.output.type": format_output_type,
    "name": connector_id,
    "value.converter": value_converter,
    "value.converter.schemas.enable": value_converter_schemas_enable,
    "key.converter": key_converter,
}

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

创建连接器后，您可以修改、删除、暂停、停止或重启连接器。

配置连接器

如需查看此连接器特有的属性的完整列表，请参阅 Cloud Storage Sink 连接器文档中的配置。

输出字段

默认情况下，连接器仅写入每个 Kafka 记录中的记录值。如需包含记录键或标头，请将 format.output.fields 属性设置为要包含的字段的逗号分隔列表。支持的字段为 key、value、headers 和 timestamp。例如：

format.output.fields=key,value,headers

输出格式

该连接器支持以 CSV、JSON 和 Parquet 格式写入 Kafka 记录。 以下部分介绍了如何配置输出格式。

CSV 格式

如需将 CSV 文件写入 Cloud Storage，请设置 format.output.type=csv。使用以下转换器：

  • 密钥转换器：org.apache.kafka.connect.storage.StringConverter 或 org.apache.kafka.connect.json.JsonConverter

  • 值转换器：org.apache.kafka.connect.converters.ByteArrayConverter

JSON 格式

如需将 JSON 格式的文件写入 Cloud Storage，请将 format.output.type 设置为以下值之一：

  • json：每个输出文件都包含一个 JSON 对象数组，其中每个 JSON 对象代表一条 Kafka 记录。

  • jsonl：连接器会将每条 Kafka 记录作为 JSON 对象写入单独的一行。

以下示例展示了连接器如何以 JSON 格式写入 Kafka 记录：

{"headers":[{"key":"k1","value":"v1"},{"key":"k2","value":"v2"}],"value":"Hello world","key":"my-key"}

如果您将键或值转换器设置为 org.apache.kafka.connect.converters.ByteArrayConverter，则连接器会将该字段写入为 base64 编码值。

连接器不会在输出中包含 JSON 架构，即使 value.converter.schemas.enable 为 true 也是如此。

Parquet 格式

如需将 Parquet 文件写入 Cloud Storage，请设置 format.output.type=parquet。

如果键或值转换器为 org.apache.kafka.connect.json.JsonConverter，则 JSON 数据必须包含架构。将转换器的 schemas.enable 属性设置为 true。

对于 JSON 数据，连接器会将数据作为 struct 类型写入 Parquet 文件。例如，如果 Kafka 记录值具有名为 userid 的字段，则连接器会写入名为 value.userid 的列。如需展平列名称，请设置 format.output.envelope=false。如果 format.output.envelope 为 false，则 format.output.fields 必须指定单个输出字段。

Cloud Storage 文件设置

以下设置控制连接器如何将文件写入 Cloud Storage。

  • gcs.bucket.name：必填。指定连接器向其中写入数据的 Cloud Storage 存储桶的名称。

  • file.compression.type：指定输出文件的压缩类型。 例如 gzip、snappy、zstd 和 none。默认值为 none。

  • file.name.prefix：指定要添加到每个文件名称的前缀。 默认值为空 ("")。

  • file.name.template：指定文件名的模板。该连接器支持模板的占位符变量，例如 {{topic}} 表示主题名称，{{partition}} 表示分区编号。如需了解详情，请参阅 Cloud Storage Sink 连接器文档。

  • file.max.records：要写入单个文件的记录数上限。如果值为 0，则没有特定的最大值。默认值为 0。

  • file.max.bytes：要写入的文件大小上限（以字节为单位）。如果值为 0，则没有特定的上限。默认值为 0。

记录分组

当 Cloud Storage 接收器连接器写入 Kafka 记录时，默认情况下，它会按主题和分区对记录进行分组。例如，主题 my-topic 的分区 0 中的记录会写入名为 my-topic-0-OFFSET 的文件，其中 OFFSET 是起始偏移量。

如果 file.name.template 参数包含 {{key}} 占位符变量，则连接器会按键对记录进行分组。在此模式下，连接器会为每个记录键写入一个单独的文件。连接器会使用该键的最新记录覆盖每个文件。

凭据

gcs.credentials.default 属性必须设置为 true。此设置使连接器能够自动发现 Google Cloud 凭据。

接下来怎么做？

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