Apache Kafka to Kafka 템플릿

Apache Kafka to Apache Kafka 템플릿은 Apache Kafka 소스에서 데이터를 바이트로 수집한 후 바이트 데이터를 Apache Kafka 싱크에 쓰는 스트리밍 파이프라인을 만듭니다.

파이프라인 요구사항

• Apache Kafka 소스 주제가 있어야 합니다.
• Apache Kafka 소스 및 싱크 브로커 서버가 실행 중이며 Dataflow 작업자 머신에서 연결될 수 있어야 합니다.
• Apache Kafka용 Google Cloud 관리 서비스를 소스나 싱크로 사용하는 경우 템플릿을 실행하기 전에 주제가 있어야 합니다.

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을 사용합니다. BigQuery용 Apache Kafka에서는 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: SASL_PLAIN 인증에서 사용할 Kafka 사용자 이름이 포함된 Google Cloud Secret Manager 보안 비밀 ID입니다. 예를 들면 projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>입니다. 기본값은 빈 값입니다.
• kafkaReadPasswordSecretId: SASL_PLAIN 인증에서 사용할 Kafka 비밀번호가 포함된 Google Cloud Secret Manager 보안 비밀 ID입니다. 예를 들면 projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>입니다. 기본값은 빈 값입니다.
• kafkaReadKeystoreLocation: Kafka 클러스터로 인증할 때 사용할 TLS 인증서와 비공개 키가 포함된 Java 키 저장소 (JKS) 파일의 Google Cloud Storage 경로입니다. 예를 들면 gs://your-bucket/keystore.jks입니다.
• kafkaReadTruststoreLocation: Kafka 브로커 ID를 확인하는 데 사용할 신뢰할 수 있는 인증서가 포함된 Java 트러스트 저장소 (JKS) 파일의 Google Cloud Storage 경로입니다.
• kafkaReadTruststorePasswordSecretId: Kafka TLS 인증을 위해 Java 트러스트 저장소(JKS) 파일에 액세스하는 데 사용할 비밀번호가 포함된 Google Cloud Secret Manager 보안 비밀 ID입니다(예: projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>).
• kafkaReadKeystorePasswordSecretId: Kafka TLS 인증을 위해 Java 키 저장소 (JKS) 파일에 액세스하는 데 사용할 비밀번호가 포함된 Google Cloud Secret Manager 보안 비밀 ID입니다. 예를 들면 projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>입니다.
• kafkaReadKeyPasswordSecretId: Kafka TLS 인증을 위해 Java 키 저장소 (JKS) 파일 내 비공개 키에 액세스하는 데 사용할 비밀번호가 포함된 Google Cloud Secret Manager 보안 비밀 ID입니다. 예를 들면 projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>입니다.
• kafkaWriteUsernameSecretId: 대상 Kafka 클러스터와의 SASL_PLAIN 인증을 위한 Kafka 사용자 이름이 포함된 Google Cloud Secret Manager 보안 비밀 ID입니다. 예를 들면 projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>입니다. 기본값은 빈 값입니다.
• kafkaWritePasswordSecretId: 대상 Kafka 클러스터와의 SASL_PLAIN 인증에 사용할 Kafka 비밀번호가 포함된 Google Cloud Secret Manager 보안 비밀 ID입니다. 예를 들면 projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>입니다. 기본값은 빈 값입니다.
• kafkaWriteKeystoreLocation: 대상 Kafka 클러스터를 인증하기 위한 TLS 인증서와 비공개 키가 포함된 Java 키 저장소 (JKS) 파일의 Google Cloud Storage 경로입니다. 예를 들면 gs://<BUCKET>/<KEYSTORE>.jks입니다.
• kafkaWriteTruststoreLocation: 대상 Kafka 브로커 ID를 확인하는 데 사용할 신뢰할 수 있는 인증서가 포함된 Java 트러스트 저장소 (JKS) 파일의 Google Cloud Storage 경로입니다.
• kafkaWriteTruststorePasswordSecretId: 대상 Kafka 클러스터와의 TLS 인증을 위해 Java 트러스트 저장소 (JKS) 파일에 액세스하는 데 사용할 비밀번호가 포함된 Google Cloud Secret Manager 보안 비밀 ID입니다. 예를 들면 projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>입니다.
• kafkaWriteKeystorePasswordSecretId: 대상 Kafka 클러스터와의 TLS 인증에 사용할 Java 키 저장소 (JKS) 파일에 액세스하는 비밀번호가 포함된 Google Cloud Secret Manager 보안 비밀 ID입니다. 예를 들면 projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>입니다.
• kafkaWriteKeyPasswordSecretId: 대상 Kafka 클러스터와의 TLS 인증을 위해 Java 키 저장소 (JKS) 파일 내 비공개 키에 액세스하는 데 사용할 비밀번호가 포함된 Google Cloud Secret Manager 보안 비밀 ID입니다. 예를 들면 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 이상이 필요합니다.

셸 또는 터미널에서 템플릿을 실행합니다.

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: 사용할 JavaScript 사용자 정의 함수(UDF)를 정의하는 .js 파일의 Cloud Storage URI입니다. 예를 들면 gs://my-bucket/my-udfs/my_file.js입니다.
• JAVASCRIPT_FUNCTION: 사용할 자바스크립트 사용자 정의 함수(UDF)의 이름입니다.

  예를 들어 자바스크립트 함수가 myTransform(inJson) { /*...do stuff...*/ }이면 함수 이름은 myTransform입니다. 샘플 자바스크립트 UDF는 UDF 예시를 참조하세요.

• KAFKA_SERVER_ADDRESSES: Apache Kafka 브로커 서버 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: 사용할 JavaScript 사용자 정의 함수(UDF)를 정의하는 .js 파일의 Cloud Storage URI입니다. 예를 들면 gs://my-bucket/my-udfs/my_file.js입니다.
• JAVASCRIPT_FUNCTION: 사용할 자바스크립트 사용자 정의 함수(UDF)의 이름입니다.

  예를 들어 자바스크립트 함수가 myTransform(inJson) { /*...do stuff...*/ }이면 함수 이름은 myTransform입니다. 샘플 자바스크립트 UDF는 UDF 예시를 참조하세요.

• KAFKA_SERVER_ADDRESSES: Apache Kafka 브로커 서버 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 제공 템플릿 목록 참조