Plantilla de Apache Kafka a Kafka

La plantilla de Apache Kafka a Apache Kafka crea un flujo de procesamiento en streaming que ingiere datos como bytes de una fuente de Apache Kafka y, a continuación, escribe los bytes en un receptor de Apache Kafka.

Requisitos del flujo de procesamiento

• El tema de origen de Apache Kafka debe existir.
• Los servidores de los brokers de origen y de receptor de Apache Kafka deben estar en funcionamiento y se debe poder acceder a ellos desde las máquinas de trabajador de Dataflow.
• Si usas Google Cloud Managed Service para Apache Kafka como origen o como receptor, el tema debe existir antes de iniciar la plantilla.

Formato de mensaje de Kafka

Los mensajes de origen de Apache Kafka se leen como bytes y los bytes se escriben en el receptor de Apache Kafka.

Autenticación

La plantilla de Apache Kafka a Apache Kafka admite la autenticación SASL/PLAIN y TLS en los brokers de Kafka.

Parámetros de plantilla

Parámetros obligatorios

• readBootstrapServerAndTopic servidor de arranque y tema de Kafka para leer la entrada. Por ejemplo, localhost:9092;topic1,topic2.
• kafkaReadAuthenticationMode el modo de autenticación que se va a usar con el clúster de Kafka. Usa KafkaAuthenticationMethod.NONE para no usar autenticación, KafkaAuthenticationMethod.SASL_PLAIN para usar el nombre de usuario y la contraseña de SASL/PLAIN, KafkaAuthenticationMethod.SASL_SCRAM_512 para usar la autenticación SASL_SCRAM_512 y KafkaAuthenticationMethod.TLS para usar la autenticación basada en certificados. KafkaAuthenticationMethod.APPLICATION_DEFAULT_CREDENTIALS solo se debe usar en el clúster de Google Cloud Apache Kafka para BigQuery. Permite autenticarte mediante las credenciales predeterminadas de la aplicación.
• writeBootstrapServerAndTopic: tema de Kafka en el que se escribirá la salida.
• kafkaWriteAuthenticationMethod el modo de autenticación que se va a usar con el clúster de Kafka. Usa NONE si no quieres usar autenticación, SASL_PLAIN para el nombre de usuario y la contraseña de SASL/PLAIN, SASL_SCRAM_512 para la autenticación basada en SASL_SCRAM_512 y TLS para la autenticación basada en certificados. El valor predeterminado es APPLICATION_DEFAULT_CREDENTIALS.

Parámetros opcionales

• enableCommitOffsets confirma los desplazamientos de los mensajes procesados en Kafka. Si se habilita, se minimizarán los huecos o el procesamiento duplicado de mensajes al reiniciar la canalización. Requiere que se especifique el ID de grupo de consumidores. Valor predeterminado: false.
• consumerGroupId identificador único del grupo de consumidores al que pertenece esta canalización. Obligatorio si la opción Confirmar desfases en Kafka está habilitada. El valor predeterminado es una cadena vacía.
• kafkaReadOffset el punto de partida para leer mensajes cuando no hay offsets confirmados. La más antigua empieza desde el principio y la más reciente, desde el mensaje más reciente. El valor predeterminado es latest.
• kafkaReadUsernameSecretId el ID del secreto de Google Cloud Secret Manager que contiene el nombre de usuario de Kafka que se va a usar con la autenticación SASL_PLAIN. Por ejemplo, projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>. El valor predeterminado es una cadena vacía.
• kafkaReadPasswordSecretId el ID del secreto de Google Cloud Secret Manager que contiene la contraseña de Kafka que se va a usar con la autenticación SASL_PLAIN. Por ejemplo, projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>. El valor predeterminado es una cadena vacía.
• kafkaReadKeystoreLocation ruta de Google Cloud Storage al archivo Java KeyStore (JKS) que contiene el certificado TLS y la clave privada que se usarán para autenticarte con el clúster de Kafka. Por ejemplo, gs://your-bucket/keystore.jks.
• kafkaReadTruststoreLocation la ruta de Google Cloud Storage al archivo Java TrustStore (JKS) que contiene los certificados de confianza que se usarán para verificar la identidad del broker de Kafka.
• kafkaReadTruststorePasswordSecretId el ID del secreto de Google Cloud Secret Manager que contiene la contraseña que se usa para acceder al archivo Java TrustStore (JKS) para la autenticación TLS de Kafka. Por ejemplo, projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>.
• kafkaReadKeystorePasswordSecretId el ID del secreto de Google Cloud Secret Manager que contiene la contraseña que se usa para acceder al archivo Java KeyStore (JKS) para la autenticación TLS de Kafka. Por ejemplo, projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>.
• kafkaReadKeyPasswordSecretId el ID de secreto de Google Cloud Secret Manager que contiene la contraseña que se usa para acceder a la clave privada del archivo Java KeyStore (JKS) para la autenticación TLS de Kafka. Por ejemplo, projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>.
• kafkaReadSaslScramUsernameSecretId el ID del secreto de Google Cloud Secret Manager que contiene el nombre de usuario de Kafka que se va a usar con la autenticación SASL_SCRAM. Por ejemplo, projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>.
• kafkaReadSaslScramPasswordSecretId el ID del secreto de Secret Manager de Google Cloud que contiene la contraseña de Kafka que se va a usar con la autenticación SASL_SCRAM. Por ejemplo, projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>.
• kafkaReadSaslScramTruststoreLocation la ruta de Google Cloud Storage al archivo Java TrustStore (JKS) que contiene los certificados de confianza que se usarán para verificar la identidad del broker de Kafka.
• kafkaReadSaslScramTruststorePasswordSecretId el ID del secreto de Google Cloud Secret Manager que contiene la contraseña que se usa para acceder al archivo Java TrustStore (JKS) para la autenticación SASL_SCRAM de Kafka. Por ejemplo, projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>.
• kafkaWriteUsernameSecretId el ID secreto de Google Cloud Secret Manager que contiene el nombre de usuario de Kafka para la autenticación SASL_PLAIN con el clúster de Kafka de destino. Por ejemplo, projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>. El valor predeterminado es una cadena vacía.
• kafkaWritePasswordSecretId el ID secreto de Secret Manager de Google Cloud que contiene la contraseña de Kafka que se va a usar para la autenticación SASL_PLAIN con el clúster de Kafka de destino. Por ejemplo, projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>. El valor predeterminado es una cadena vacía.
• kafkaWriteKeystoreLocation la ruta de Google Cloud Storage al archivo de almacén de claves de Java (JKS) que contiene el certificado TLS y la clave privada para autenticarte en el clúster de Kafka de destino. Por ejemplo, gs://<BUCKET>/<KEYSTORE>.jks.
• kafkaWriteTruststoreLocation la ruta de Google Cloud Storage al archivo Java TrustStore (JKS) que contiene los certificados de confianza que se usarán para verificar la identidad del broker de Kafka de destino.
• kafkaWriteTruststorePasswordSecretId el ID del secreto de Google Cloud Secret Manager que contiene la contraseña que se usa para acceder al archivo Java TrustStore (JKS) para la autenticación TLS con el clúster de Kafka de destino. Por ejemplo, projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>.
• kafkaWriteKeystorePasswordSecretId el ID del secreto de Google Cloud Secret Manager que contiene la contraseña para acceder al archivo Java KeyStore (JKS) que se va a usar para la autenticación TLS con el clúster de Kafka de destino. Por ejemplo, projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>.
• kafkaWriteKeyPasswordSecretId el ID de secreto de Google Cloud Secret Manager que contiene la contraseña que se debe usar para acceder a la clave privada del archivo de almacén de claves de Java (JKS) para la autenticación TLS con el clúster de Kafka de destino. Por ejemplo, projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>.

Ejecutar la plantilla

Consola

1. Ve a la página Crear tarea a partir de plantilla de Dataflow.
Ir a Crear tarea a partir de plantilla
2. En el campo Nombre de la tarea, introduce un nombre único.
3. Opcional: En Endpoint regional, seleccione un valor en el menú desplegable. La región predeterminada es us-central1.

   Para ver una lista de las regiones en las que puedes ejecutar una tarea de Dataflow, consulta Ubicaciones de Dataflow.

4. En el menú desplegable Plantilla de flujo de datos, seleccione the Kafka to Cloud Storage template.
5. En los campos de parámetros proporcionados, introduzca los valores de los parámetros.
6. Opcional: Para cambiar del procesamiento una sola vez al modo de streaming al menos una vez, selecciona Al menos una vez.
7. Haz clic en Ejecutar trabajo.

gcloud

Nota: Para usar la CLI de Google Cloud y ejecutar plantillas de flex, debes tener la versión CLI de Google Cloud 284.0.0 o una posterior.

En tu shell o terminal, ejecuta la plantilla:

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 \
readBootstrapServerAndTopic=READ_BOOTSTRAP_SERVER_AND_TOPIC,\
kafkaReadAuthenticationMode=APPLICATION_DEFAULT_CREDENTIALS,\
writeBootstrapServerAndTopic=WRITE_BOOTSTRAP_SERVER_AND_TOPIC,\
kafkaWriteAuthenticationMethod=APPLICATION_DEFAULT_CREDENTIALS
  

Haz los cambios siguientes:

• PROJECT_ID: el ID del proyecto Google Cloud en el que quieres ejecutar la tarea de Dataflow
• JOB_NAME: un nombre de trabajo único que elijas
• REGION_NAME: la región en la que quieras desplegar tu trabajo de Dataflow. Por ejemplo, us-central1
• VERSION: la versión de la plantilla que quieres usar

  Puedes usar los siguientes valores:

  • latest para usar la última versión de la plantilla, que está disponible en la carpeta principal sin fecha del contenedor: gs://dataflow-templates-REGION_NAME/latest/
  • el nombre de la versión, como 2023-09-12-00_RC00, para usar una versión específica de la plantilla, que se encuentra anidada en la carpeta principal correspondiente con la fecha en el bucket: gs://dataflow-templates-REGION_NAME/
  Precaución: La última versión de las plantillas puede actualizarse con cambios importantes. Tus entornos de producción deben usar plantillas que se encuentren en la carpeta principal con fecha más reciente para evitar que estos cambios afecten a tus flujos de trabajo de producción.
• READ_BOOTSTRAP_SERVER_AND_TOPIC: la dirección del servidor de arranque de Apache Kafka y el tema del que se va a leer
• WRITE_BOOTSTRAP_SERVER_AND_TOPIC: la dirección del servidor de arranque de Apache Kafka y el tema en el que escribir

  El formato de la dirección del servidor de arranque y del tema depende del tipo de clúster:

  • Clúster de Managed Service para Apache Kafka: projects/PROJECT_ID/locations/REGION_NAME/clusters/CLUSTER_NAME/topics/TOPIC_NAME
  • Clúster de Kafka externo: BOOTSTRAP_SERVER_ADDRESS;TOPIC_NAME

API

Para ejecutar la plantilla mediante la API REST, envía una solicitud HTTP POST. Para obtener más información sobre la API y sus ámbitos de autorización, consulta projects.templates.launch.

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/flexTemplates:launch
{
   "launch_parameter": {
      "jobName": "JOB_NAME",
      "parameters": {
          "readBootstrapServerAndTopic": "READ_BOOTSTRAP_SERVER_AND_TOPIC",
          "kafkaReadAuthenticationMode": "APPLICATION_DEFAULT_CREDENTIALS",
          "writeBootstrapServerAndTopic": "WRITE_BOOTSTRAP_SERVER_AND_TOPIC",
          "kafkaWriteAuthenticationMethod": "APPLICATION_DEFAULT_CREDENTIALS
      },
      "containerSpecGcsPath": "gs://dataflow-templates-LOCATION/VERSION/flex/Kafka_to_Kafka",
   }
}
  

Haz los cambios siguientes:

• PROJECT_ID: el ID del proyecto Google Cloud en el que quieres ejecutar la tarea de Dataflow
• JOB_NAME: un nombre de trabajo único que elijas
• LOCATION: la región en la que quieras desplegar tu trabajo de Dataflow. Por ejemplo, us-central1
• VERSION: la versión de la plantilla que quieres usar

  Puedes usar los siguientes valores:

  • latest para usar la última versión de la plantilla, que está disponible en la carpeta principal sin fecha del contenedor: gs://dataflow-templates-REGION_NAME/latest/
  • el nombre de la versión, como 2023-09-12-00_RC00, para usar una versión específica de la plantilla, que se encuentra anidada en la carpeta principal correspondiente con la fecha en el bucket: gs://dataflow-templates-REGION_NAME/
  Precaución: La última versión de las plantillas puede actualizarse con cambios importantes. Tus entornos de producción deben usar plantillas que se encuentren en la carpeta principal con fecha más reciente para evitar que estos cambios afecten a tus flujos de trabajo de producción.
• READ_BOOTSTRAP_SERVER_AND_TOPIC: la dirección del servidor de arranque de Apache Kafka y el tema del que se va a leer
• WRITE_BOOTSTRAP_SERVER_AND_TOPIC: la dirección del servidor de arranque de Apache Kafka y el tema en el que escribir

  El formato de la dirección del servidor de arranque y del tema depende del tipo de clúster:

  • Clúster de Managed Service para Apache Kafka: projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_NAME/topics/TOPIC_NAME
  • Clúster de Kafka externo: BOOTSTRAP_SERVER_ADDRESS;TOPIC_NAME

Código fuente de la plantilla

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();
  }
}

Siguientes pasos

• Consulta información sobre las plantillas de Dataflow.
• Consulta la lista de plantillas proporcionadas por Google.