Modelo do Apache Kafka para Cloud Storage

O modelo Apache Kafka para Cloud Storage é um pipeline de streaming que ingere dados de texto do Apache Kafka para BigQuery e envia os registros para o Cloud Storage.

Também é possível usar o modelo Apache Kafka para BigQuery com o Kafka autogerenciado ou externo.

Requisitos de pipeline

  • O bucket de saída do Cloud Storage precisa existir.
  • O servidor do agente do Apache Kafka precisa estar em execução e acessível em máquinas de trabalho do Dataflow.
  • Os tópicos do Apache Kafka precisam existir.

Formato de mensagem do Kafka

O modelo do Apache Kafka para Cloud Storage oferece suporte à leitura de mensagens do Kafka nos seguintes formatos: CONFLUENT_AVRO_WIRE_FORMAT e JSON.

Formato de arquivo de saída

O formato do arquivo de saída é igual à mensagem de entrada do Kafka. Por exemplo, se você selecionar JSON para o formato de mensagem do Kafka, os arquivos JSON serão gravados no bucket de saída do Cloud Storage.

Autenticação

O modelo Apache Kafka para Cloud Storage oferece suporte à autenticação SASL/PLAIN para agentes Kafka.

Parâmetros do modelo

Parâmetros obrigatórios

  • readBootstrapServerAndTopic: o tópico do Kafka a ser usado para ler a entrada.
  • kafkaReadAuthenticationMode: o modo de autenticação a ser usado com o cluster do Kafka. Use NONE para nenhuma autenticação e SASL_PLAIN para o nome de usuário e senha SASL/PLAIN. O Apache Kafka para BigQuery só oferece suporte ao modo de autenticação SASL_PLAIN. O padrão é: SASL_PLAIN.
  • outputDirectory: o caminho e o prefixo do nome do arquivo para gravar arquivos de saída. Precisa terminar com uma barra. Exemplo: gs://your-bucket/your-path/.
  • messageFormat: o formato das mensagens do Kafka a serem lidas. Os valores aceitos são AVRO_CONFLUENT_WIRE_FORMAT (Avro codificado do Confluent Schema Registry), AVRO_BINARY_ENCODING (Avro binário simples) e JSON. O padrão é: AVRO_CONFLUENT_WIRE_FORMAT.

Parâmetros opcionais

  • windowDuration: a duração/tamanho da janela em que os dados serão gravados no Cloud Storage. Os formatos permitidos são: Ns (para segundos, exemplo: "5s"), Nm (para minutos, exemplo: "12m") e Nh (para horas, exemplo: "2h"). (Exemplo: 5 min). O padrão é de 5 minutos.
  • outputFilenamePrefix: o prefixo a ser colocado em cada arquivo em janela. (Exemplo: output-). O padrão é: saída.
  • numShards: o número máximo de fragmentos de saída produzidos durante a gravação. Um número maior de fragmentos significa maior capacidade de gravação no Cloud Storage, mas um custo de agregação de dados potencialmente maior entre os fragmentos ao processar os arquivos de saída do Cloud Storage. O valor padrão é decidido pelo Dataflow.
  • enableCommitOffsets: deslocamentos de confirmação de mensagens processadas para o Kafka. Se ativado, isso minimizará as lacunas ou o processamento duplicado de mensagens ao reiniciar o pipeline. Exige especificar o ID do grupo de consumidores. O padrão é: falso.
  • consumerGroupId: o identificador exclusivo do grupo de consumidores ao qual esse pipeline pertence. Obrigatório se os deslocamentos de confirmação para Kafka estiverem ativados. O padrão é vazio.
  • kafkaReadOffset: o ponto de partida para ler mensagens quando não há deslocamentos confirmados. O mais antigo começa no início, o mais recente a partir da mensagem mais recente. O padrão é: mais recente.
  • kafkaReadUsernameSecretId: o ID do secret do Secret Manager do Google Cloud que contém o nome de usuário do Kafka a ser usado com a autenticação SASL_PLAIN. (Exemplo: projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>). O padrão é vazio.
  • kafkaReadPasswordSecretId: o ID do secret do Secret Manager do Google Cloud que contém a senha do Kafka a ser usada com a autenticação SASL_PLAIN. (Exemplo: projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<SECRET_VERSION>). O padrão é vazio.
  • schemaFormat: o formato de esquema do Kafka. Pode ser fornecido como SINGLE_SCHEMA_FILE ou SCHEMA_REGISTRY. Se SINGLE_SCHEMA_FILE for especificado, todas as mensagens vão precisar ter o esquema mencionado no arquivo de esquema avro. Se SCHEMA_REGISTRY for especificado, as mensagens poderão ter um único esquema ou vários esquemas. O padrão é SINGLE_SCHEMA_FILE.
  • confluentAvroSchemaPath: o caminho do Google Cloud Storage para o único arquivo de esquema Avro usado para decodificar todas as mensagens em um tópico. O padrão é vazio.
  • schemaRegistryConnectionUrl: o URL da instância do Confluent Schema Registry usada para gerenciar esquemas Avro para decodificação de mensagens. O padrão é vazio.
  • binaryAvroSchemaPath: o caminho do Google Cloud Storage para o arquivo de esquema do Avro usado para decodificar mensagens Avro codificadas em binário. O padrão é vazio.

Executar o modelo

Console

  1. Acesse a página Criar job usando um modelo do Dataflow.
  2. Acesse Criar job usando um modelo
  3. No campo Nome do job, insira um nome exclusivo.
  4. Opcional: em Endpoint regional, selecione um valor no menu suspenso. A região padrão é us-central1.

    Para ver uma lista de regiões em que é possível executar um job do Dataflow, consulte Locais do Dataflow.

  5. No menu suspenso Modelo do Dataflow, selecione the Kafka to Cloud Storage template.
  6. Nos campos de parâmetro fornecidos, insira os valores de parâmetro.
  7. Opcional: para alternar do processamento "Exatamente uma vez" para o modo de streaming "Pelo menos uma vez", selecione Pelo menos uma vez.
  8. Cliquem em Executar job.

gcloud

No shell ou no terminal, execute o modelo:

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

Substitua:

  • PROJECT_ID: o ID do projeto do Google Cloud em que você quer executar o job do Dataflow
  • JOB_NAME: um nome de job de sua escolha
  • REGION_NAME: a região onde você quer implantar o job do Dataflow, por exemplo, us-central1
  • VERSION: a versão do modelo que você quer usar

    Use estes valores:

  • BIGQUERY_TABLE: nome da tabela do Cloud Storage
  • KAFKA_TOPICS: a lista de tópicos do Apache Kakfa. Se vários tópicos forem fornecidos, use vírgulas. Consulte gcloud topic escaping.
  • PATH_TO_JAVASCRIPT_UDF_FILE: O URI do Cloud Storage do arquivo .js que define a função definida pelo usuário (UDF) do JavaScript que você quer usar, por exemplo, gs://my-bucket/my-udfs/my_file.js
  • JAVASCRIPT_FUNCTION: o nome da função definida pelo usuário (UDF) do JavaScript que você quer usar

    Por exemplo, se o código de função do JavaScript for myTransform(inJson) { /*...do stuff...*/ }, o nome da função será myTransform. Para amostras de UDFs do JavaScript, consulte os exemplos de UDF.

  • KAFKA_SERVER_ADDRESSES: a lista de endereços IP do servidor de gerenciamento do Apache Kafka. Cada endereço IP precisa ter o número da porta que o servidor pode acessar. Por exemplo, 35.70.252.199:9092. Se vários endereços forem fornecidos, use vírgulas para separá-los. Consulte gcloud topic escaping.

API

Para executar o modelo usando a API REST, envie uma solicitação HTTP POST. Para mais informações sobre a API e os respectivos escopos de autorização, consulte 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_Cloud Storage",
   }
}
  

Substitua:

  • PROJECT_ID: o ID do projeto do Google Cloud em que você quer executar o job do Dataflow
  • JOB_NAME: um nome de job de sua escolha
  • LOCATION: a região onde você quer implantar o job do Dataflow, por exemplo, us-central1
  • VERSION: a versão do modelo que você quer usar

    Use estes valores:

  • BIGQUERY_TABLE: nome da tabela do Cloud Storage
  • KAFKA_TOPICS: a lista de tópicos do Apache Kakfa. Se vários tópicos forem fornecidos, use vírgulas. Consulte gcloud topic escaping.
  • PATH_TO_JAVASCRIPT_UDF_FILE: O URI do Cloud Storage do arquivo .js que define a função definida pelo usuário (UDF) do JavaScript que você quer usar, por exemplo, gs://my-bucket/my-udfs/my_file.js
  • JAVASCRIPT_FUNCTION: o nome da função definida pelo usuário (UDF) do JavaScript que você quer usar

    Por exemplo, se o código de função do JavaScript for myTransform(inJson) { /*...do stuff...*/ }, o nome da função será myTransform. Para amostras de UDFs do JavaScript, consulte os exemplos de UDF.

  • KAFKA_SERVER_ADDRESSES: a lista de endereços IP do servidor de gerenciamento do Apache Kafka. Cada endereço IP precisa ter o número da porta que o servidor pode acessar. Por exemplo, 35.70.252.199:9092. Se vários endereços forem fornecidos, use vírgulas para separá-los. Consulte gcloud topic escaping.

Para mais informações, consulte Gravar dados do Kafka no Cloud Storage com Dataflow.

A seguir