Apache Kafka to Kafka 模板

Apache Kafka to Apache Kafka 模板会创建一个流处理流水线，可从 Apache Kafka 来源以字节形式注入数据，然后将字节写入 Apache Kafka 接收器。

流水线要求

• Apache Kafka 来源主题必须已存在。
• Apache Kafka 来源和接收器代理服务器必须已在运行且可从 Dataflow 工作器机器进行访问。
• 如果您将 Google Cloud Managed Service for Apache Kafka 用作来源或接收器，则该主题必须已存在才能启动模板。

Kafka 消息格式

Apache Kafka 来源消息会以字节形式读取，然后字节会写入 Apache Kafka 接收器。

身份验证

Apache Kafka to Apache Kafka 模板支持对 Kafka 代理进行 SASL/PLAIN 和 TLS 身份验证。

模板参数

必需参数

• readBootstrapServerAndTopic：要从中读取输入的 Kafka 引导服务器和主题。例如 localhost:9092;topic1,topic2。
• kafkaReadAuthenticationMode：与 Kafka 集群搭配使用的身份验证模式。如果不进行身份验证，请使用 NONE；如果使用 SASL/PLAIN 用户名和密码，请使用 SASL_PLAIN；如果使用基于证书的身份验证，请使用 TLS。Apache Kafka for BigQuery 仅支持 SASL_PLAIN 身份验证模式。默认值为：SASL_PLAIN。
• writeBootstrapServerAndTopic：要将输出写入的 Kafka 主题。
• kafkaWriteAuthenticationMethod：与 Kafka 集群搭配使用的身份验证模式。如果不进行身份验证，请使用 NONE；如果使用 SASL/PLAIN 用户名和密码，请使用 SASL_PLAIN；如果使用基于证书的身份验证，请使用 TLS。默认值为：APPLICATION_DEFAULT_CREDENTIALS。

可选参数

• enableCommitOffsets：将已处理消息的偏移量提交到 Kafka。如果启用此参数，则在重启流水线时，消息处理的间隔或重复处理会降到最低。需要指定使用方群组 ID。默认值为：false。
• consumerGroupId：此流水线所属的使用方群组的唯一标识符。如果已启用“将偏移量提交到 Kafka”，则必须使用此参数。默认值为空。
• kafkaReadOffset：在没有提交偏移量的情况下读取消息的起点。最早的从最开始算起，最新的从最新消息算起。默认值为：latest。
• kafkaReadUsernameSecretId：Google Cloud Secret Manager Secret ID，其中包含要与 SASL_PLAIN 身份验证搭配使用的 Kafka 用户名。例如 projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>。默认值为空。
• kafkaReadPasswordSecretId：Google Cloud Secret Manager Secret ID，其中包含要与 SASL_PLAIN 身份验证搭配使用的 Kafka 密码。例如 projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>。默认值为空。
• kafkaReadKeystoreLocation：Java KeyStore (JKS) 文件的 Google Cloud Storage 路径，该文件包含在向 Kafka 集群进行身份验证时使用的 TLS 证书和私钥。例如 gs://your-bucket/keystore.jks。
• kafkaReadTruststoreLocation：Java TrustStore (JKS) 文件的 Google Cloud Storage 路径，该文件包含用于验证 Kafka 代理身份的受信任证书。
• kafkaReadTruststorePasswordSecretId：Google Cloud Secret Manager Secret ID，其中包含用于访问 Java TrustStore (JKS) 文件以进行 Kafka TLS 身份验证的密码，例如 projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>。
• kafkaReadKeystorePasswordSecretId：Google Cloud Secret Manager Secret ID，其中包含用于访问 Java KeyStore (JKS) 文件以进行 Kafka TLS 身份验证的密码。例如 projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>。
• kafkaReadKeyPasswordSecretId：Google Cloud Secret Manager Secret ID，其中包含用于访问 Java KeyStore (JKS) 文件中的私钥以进行 Kafka TLS 身份验证的密码。例如 projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>。
• kafkaWriteUsernameSecretId：Google Cloud Secret Manager Secret ID，其中包含用于向目标 Kafka 集群进行 SASL_PLAIN 身份验证的 Kafka 用户名。例如 projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>。默认值为空。
• kafkaWritePasswordSecretId：Google Cloud Secret Manager Secret ID，其中包含用于向目标 Kafka 集群进行 SASL_PLAIN 身份验证的 Kafka 密码。例如 projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>。默认值为空。
• kafkaWriteKeystoreLocation：Java KeyStore (JKS) 文件的 Google Cloud Storage 路径，该文件包含用于向目标 Kafka 集群进行身份验证的 TLS 证书和私钥。例如 gs://<BUCKET>/<KEYSTORE>.jks。
• kafkaWriteTruststoreLocation：Java TrustStore (JKS) 文件的 Google Cloud Storage 路径，该文件包含用于验证目标 Kafka 代理身份的受信任证书。
• kafkaWriteTruststorePasswordSecretId：Google Cloud Secret Manager Secret ID，其中包含用于访问 Java TrustStore (JKS) 文件以向目标 Kafka 集群进行 TLS 身份验证的密码。例如 projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>。
• kafkaWriteKeystorePasswordSecretId：Google Cloud Secret Manager Secret ID，其中包含用于访问 Java KeyStore (JKS) 文件以向目标 Kafka 集群进行 TLS 身份验证的密码。例如 projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>。
• kafkaWriteKeyPasswordSecretId：Google Cloud Secret Manager Secret ID，其中包含用于访问 Java KeyStore (JKS) 文件中的私钥以向目标 Kafka 集群进行 TLS 身份验证的密码。例如 projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>。

运行模板

控制台

1. 转到 Dataflow 基于模板创建作业页面。
转到“基于模板创建作业”
2. 在作业名称字段中，输入唯一的作业名称。
3. 可选：对于区域性端点，从下拉菜单中选择一个值。默认区域为 us-central1。

   如需查看可以在其中运行 Dataflow 作业的区域列表，请参阅 Dataflow 位置。

4. 从 Dataflow 模板下拉菜单中，选择 the Kafka to Cloud Storage template。
5. 在提供的参数字段中，输入您的参数值。
6. 可选：如需从“正好一次”处理切换到“至少一次”流处理模式，请选择至少一次。
7. 点击运行作业。

gcloud

注意：如需使用 Google Cloud CLI 运行 flex模板，您必须具有 Google Cloud CLI 284.0.0 或更高版本。

在 shell 或终端中，运行模板：

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_Kafka \
    --parameters \
outputTableSpec=BIGQUERY_TABLE,\
inputTopics=KAFKA_TOPICS,\
javascriptTextTransformGcsPath=PATH_TO_JAVASCRIPT_UDF_FILE,\
javascriptTextTransformFunctionName=JAVASCRIPT_FUNCTION,\
bootstrapServers=KAFKA_SERVER_ADDRESSES
  

替换以下内容：

• PROJECT_ID：您要在其中运行 Dataflow 作业的 Google Cloud 项目的 ID
• JOB_NAME：您选择的唯一性作业名称
• REGION_NAME：要在其中部署 Dataflow 作业的区域，例如 us-central1
• VERSION：您要使用的模板的版本

  您可使用以下值：

  • latest，以使用模板的最新版本，该模板在存储桶的未标示日期的父文件夹 (gs://dataflow-templates-REGION_NAME/latest/) 中可用
  • 版本名称（如 2023-09-12-00_RC00），以使用模板的特定版本，该版本嵌套在存储桶的相应日期父文件夹 (gs://dataflow-templates-REGION_NAME/) 中
  注意：最新版模板可能会随着重大更改而更新。为了防止这些重大更改影响您的生产工作流程，生产环境应使用有最近标示日期的父文件夹中保存的模板。
• BIGQUERY_TABLE：您的 Cloud Storage 表名称
• KAFKA_TOPICS：Apache Kakfa 主题列表。如果提供了多个主题，您需要转义英文逗号。请参阅 gcloud topic escaping。
• PATH_TO_JAVASCRIPT_UDF_FILE： .js 文件的 Cloud Storage URI，用于定义您要使用的 JavaScript 用户定义的函数 (UDF)，例如 gs://my-bucket/my-udfs/my_file.js
• JAVASCRIPT_FUNCTION： 您要使用的 JavaScript 用户定义的函数 (UDF) 的名称

  例如，如果您的 JavaScript 函数代码为 myTransform(inJson) { /*...do stuff...*/ }，则函数名称为 myTransform。如需查看 JavaScript UDF 示例，请参阅 UDF 示例。

• KAFKA_SERVER_ADDRESSES：Apache Kafka broker 服务器 IP 地址列表。每个 IP 地址都需要可访问服务器的端口号。例如：35.70.252.199:9092。如果提供了多个地址，您需要转义英文逗号。请参阅 gcloud topic escaping。

API

如需使用 REST API 来运行模板，请发送 HTTP POST 请求。如需详细了解 API 及其授权范围，请参阅 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_Kafka",
   }
}
  

替换以下内容：

• PROJECT_ID：您要在其中运行 Dataflow 作业的 Google Cloud 项目的 ID
• JOB_NAME：您选择的唯一性作业名称
• LOCATION：要在其中部署 Dataflow 作业的区域，例如 us-central1
• VERSION：您要使用的模板的版本

  您可使用以下值：

  • latest，以使用模板的最新版本，该模板在存储桶的未标示日期的父文件夹 (gs://dataflow-templates-REGION_NAME/latest/) 中可用
  • 版本名称（如 2023-09-12-00_RC00），以使用模板的特定版本，该版本嵌套在存储桶的相应日期父文件夹 (gs://dataflow-templates-REGION_NAME/) 中
  注意：最新版模板可能会随着重大更改而更新。为了防止这些重大更改影响您的生产工作流程，生产环境应使用有最近标示日期的父文件夹中保存的模板。
• BIGQUERY_TABLE：您的 Cloud Storage 表名称
• KAFKA_TOPICS：Apache Kakfa 主题列表。如果提供了多个主题，您需要转义英文逗号。请参阅 gcloud topic escaping。
• PATH_TO_JAVASCRIPT_UDF_FILE： .js 文件的 Cloud Storage URI，用于定义您要使用的 JavaScript 用户定义的函数 (UDF)，例如 gs://my-bucket/my-udfs/my_file.js
• JAVASCRIPT_FUNCTION： 您要使用的 JavaScript 用户定义的函数 (UDF) 的名称

  例如，如果您的 JavaScript 函数代码为 myTransform(inJson) { /*...do stuff...*/ }，则函数名称为 myTransform。如需查看 JavaScript UDF 示例，请参阅 UDF 示例。

• KAFKA_SERVER_ADDRESSES：Apache Kafka broker 服务器 IP 地址列表。每个 IP 地址都需要可访问服务器的端口号。例如：35.70.252.199:9092。如果提供了多个地址，您需要转义英文逗号。请参阅 gcloud topic escaping。

如需了解详情，请参阅使用 Dataflow 将数据从 Kafka 写入 Cloud Storage。

模板源代码

Java

/*
 * Copyright (C) 2024 Google LLC
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package com.google.cloud.teleport.v2.templates;

import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkArgument;

import com.google.cloud.teleport.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.v2.common.UncaughtExceptionLogger;
import com.google.cloud.teleport.v2.kafka.options.KafkaReadOptions;
import com.google.cloud.teleport.v2.kafka.options.KafkaWriteOptions;
import com.google.cloud.teleport.v2.kafka.transforms.KafkaTransform;
import com.google.cloud.teleport.v2.kafka.utils.FileAwareProducerFactoryFn;
import com.google.cloud.teleport.v2.kafka.utils.KafkaConfig;
import com.google.cloud.teleport.v2.kafka.utils.KafkaTopicUtils;
import com.google.cloud.teleport.v2.kafka.values.KafkaAuthenticationMethod;
import java.io.IOException;
import java.util.Collections;
import java.util.List;
import org.apache.beam.runners.dataflow.options.DataflowPipelineOptions;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.kafka.KafkaIO;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.kafka.common.serialization.ByteArraySerializer;

@Template(
    name = "Kafka_to_Kafka",
    category = TemplateCategory.STREAMING,
    displayName = "Kafka to Kafka",
    description = "A pipeline that writes data to a kafka destination from another kafka source",
    optionsClass = KafkaToKafka.KafkaToKafkaOptions.class,
    flexContainerName = "kafka-to-kafka",
    contactInformation = "https://cloud.google.com/support")
public class KafkaToKafka {
  public interface KafkaToKafkaOptions
      extends PipelineOptions, KafkaReadOptions, KafkaWriteOptions, DataflowPipelineOptions {}

  public static void main(String[] args) throws IOException {
    UncaughtExceptionLogger.register();
    KafkaToKafkaOptions options =
        PipelineOptionsFactory.fromArgs(args).withValidation().as(KafkaToKafkaOptions.class);
    run(options);
  }

  public static PipelineResult run(KafkaToKafkaOptions options) {

    if (options.getKafkaReadAuthenticationMode().equals(KafkaAuthenticationMethod.SASL_PLAIN)) {
      checkArgument(
          options.getKafkaReadUsernameSecretId().trim().length() > 0,
          "KafkaReadUsernameSecretId required to access username for source Kafka");
      checkArgument(
          options.getKafkaReadPasswordSecretId().trim().length() > 0,
          "KafkaReadPasswordSecretId required to access password for source kafka");
    } else if (options.getKafkaReadAuthenticationMode().equals(KafkaAuthenticationMethod.TLS)) {
      checkArgument(
          options.getKafkaReadTruststoreLocation().trim().length() > 0,
          "KafkaReadTruststoreLocation for trust store certificate required for ssl authentication");
      checkArgument(
          options.getKafkaReadTruststorePasswordSecretId().trim().length() > 0,
          "KafkaReadTruststorePassword for trust store password required for accessing truststore");
      checkArgument(
          options.getKafkaReadKeystoreLocation().trim().length() > 0,
          "KafkaReadKeystoreLocation for key store location required for ssl authentication");
      checkArgument(
          options.getKafkaReadKeystorePasswordSecretId().trim().length() > 0,
          "KafkaReadKeystorePassword for key store password required to access key store");
      checkArgument(
          options.getKafkaReadKeyPasswordSecretId().trim().length() > 0,
          "KafkaReadKeyPasswordSecretId version for key password required for SSL authentication");
    } else if (options.getKafkaReadAuthenticationMode().equals(KafkaAuthenticationMethod.NONE)
        || (options
            .getKafkaReadAuthenticationMode()
            .equals(KafkaAuthenticationMethod.APPLICATION_DEFAULT_CREDENTIALS))) {
    } else {
      throw new UnsupportedOperationException(
          "Authentication method not supported: " + options.getKafkaReadAuthenticationMode());
    }

    if (options.getKafkaWriteAuthenticationMethod().equals(KafkaAuthenticationMethod.SASL_PLAIN)) {
      checkArgument(
          options.getKafkaWriteUsernameSecretId().trim().length() > 0,
          "KafkaWriteUsernameSecretId required to access username for source Kafka");
      checkArgument(
          options.getKafkaWritePasswordSecretId().trim().length() > 0,
          "KafkaWritePasswordSecretId required to access password for destination Kafka");
    } else if (options.getKafkaWriteAuthenticationMethod().equals(KafkaAuthenticationMethod.TLS)) {
      checkArgument(
          options.getKafkaWriteTruststoreLocation().trim().length() > 0,
          "KafkaWriteTruststoreLocation for trust store certificate required for ssl authentication");
      checkArgument(
          options.getKafkaWriteTruststorePasswordSecretId().trim().length() > 0,
          "KafkaWriteTruststorePasswordSecretId for trust store password required for accessing truststore");
      checkArgument(
          options.getKafkaWriteKeystoreLocation().trim().length() > 0,
          "KafkaWriteKeystoreLocation for key store location required for ssl authentication");
      checkArgument(
          options.getKafkaWriteKeystorePasswordSecretId().trim().length() > 0,
          "KafkaWriteKeystorePasswordSecretId for key store password required to access key store");
      checkArgument(
          options.getKafkaWriteKeyPasswordSecretId().trim().length() > 0,
          "KafkaWriteKeyPasswordSecretId for source key password secret id version required for SSL authentication");
    } else if (options.getKafkaWriteAuthenticationMethod().equals(KafkaAuthenticationMethod.NONE)
        || options
            .getKafkaWriteAuthenticationMethod()
            .equals(KafkaAuthenticationMethod.APPLICATION_DEFAULT_CREDENTIALS)) {
      // No additional validation is required for these auth mechanisms since they don't depend on
      // any specific pipeline options.
    } else {
      throw new UnsupportedOperationException(
          "Authentication method not supported: " + options.getKafkaWriteAuthenticationMethod());
    }

    String sourceTopic;
    String sourceBootstrapServers;
    if (options.getReadBootstrapServerAndTopic() != null) {
      List<String> sourceBootstrapServerAndTopicList =
          KafkaTopicUtils.getBootstrapServerAndTopic(
              options.getReadBootstrapServerAndTopic(), options.getProject());
      sourceTopic = sourceBootstrapServerAndTopicList.get(1);
      sourceBootstrapServers = sourceBootstrapServerAndTopicList.get(0);
    } else {
      throw new IllegalArgumentException(
          "Please provide a valid bootstrap server which matches `[,:a-zA-Z0-9._-]+` and a topic which matches `[,a-zA-Z0-9._-]+`");
    }

    String destinationTopic;
    String destinationBootstrapServers;
    if (options.getWriteBootstrapServerAndTopic() != null) {
      List<String> destinationBootstrapServerAndTopicList =
          KafkaTopicUtils.getBootstrapServerAndTopic(
              options.getWriteBootstrapServerAndTopic(), options.getProject());
      destinationBootstrapServers = destinationBootstrapServerAndTopicList.get(0);
      destinationTopic = destinationBootstrapServerAndTopicList.get(1);
    } else {
      throw new IllegalArgumentException(
          "Please provide a valid bootstrap server which matches `[,:a-zA-Z0-9._-]+` and a topic which matches `[,a-zA-Z0-9._-]+`");
    }

    Pipeline pipeline = Pipeline.create(options);
    pipeline
        .apply(
            "Read from Kafka",
            KafkaTransform.readBytesFromKafka(
                    sourceBootstrapServers,
                    Collections.singletonList(sourceTopic),
                    KafkaConfig.fromReadOptions(options),
                    options.getEnableCommitOffsets())
                .withoutMetadata())
        .apply(
            "Write to Kafka",
            KafkaIO.<byte[], byte[]>write()
                .withBootstrapServers(destinationBootstrapServers)
                .withTopic(destinationTopic)
                .withKeySerializer(ByteArraySerializer.class)
                .withValueSerializer(ByteArraySerializer.class)
                .withProducerConfigUpdates(KafkaConfig.fromWriteOptions(options))
                .withProducerFactoryFn(new FileAwareProducerFactoryFn()));

    return pipeline.run();
  }
}

后续步骤

• 了解 Dataflow 模板。
• 参阅 Google 提供的模板列表。