Criar um conector de gravador do Cloud Storage

Os conectores de coletor do Cloud Storage transmitem dados do Kafka para buckets do Cloud Storage, o que permite armazenar e processar grandes volumes de dados de maneira econômica e escalonável.

Os casos de uso dos conectores de coletor do Cloud Storage incluem o seguinte:

  • Ingestão de data lake. Armazene dados do Kafka em um data lake para arquivamento de longo prazo e processamento em lote.

  • Arquive dados para atender aos requisitos regulatórios.

Antes de começar

Antes de criar um conector de coletor do Cloud Storage, verifique se você tem o seguinte:

Papéis e permissões necessárias

Para ter as permissões necessárias para criar um conector, peça ao administrador para conceder a você o papel do IAM de editor de conector do Kafka gerenciado (roles/managedkafka.connectorEditor) no projeto. Para mais informações sobre a concessão de papéis, consulte Gerenciar o acesso a projetos, pastas e organizações.

Esse papel predefinido contém as permissões necessárias para criar um conector. Para acessar as permissões exatas que são necessárias, expanda a seção Permissões necessárias:

Permissões necessárias

As seguintes permissões são necessárias para criar um conector:

  • Criar um conector: managedkafka.connectors.create

Essas permissões também podem ser concedidas com papéis personalizados ou outros papéis predefinidos.

Conceder permissões para gravar no bucket do Cloud Storage

A conta de serviço do Kafka gerenciado precisa ter as seguintes permissões para gravar mensagens no bucket do Cloud Storage:

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

Conceda o papel Usuário de objetos do Storage (roles/storage.objectUser) à conta de serviço no projeto que contém o bucket do Cloud Storage.

A conta de serviço do Kafka gerenciado tem o seguinte formato: service-PROJECT_NUMBER@gcp-sa-managedkafka.iam.gserviceaccount.com, em que PROJECT_NUMBER é o número do projeto do cluster do Connect.

Se o cluster do Connect estiver em um projeto diferente do cluster do Serviço Gerenciado para Apache Kafka cluster, consulte Criar um cluster do Connect em um projeto diferente.

Criar um conector de coletor do Cloud Storage

Para criar um conector de coletor do Cloud Storage, siga estas etapas.

Console

  1. No Google Cloud console, acesse a página Clusters do Connect.

    Acessar clusters do Connect

  2. Clique no cluster do Connect para o qual você quer criar o conector.

    A página Detalhes do cluster do Connect é exibida.

  3. Clique em Criar conector.

    A página Criar conector do Kafka é exibida.

  4. Para o nome do conector, insira uma string.

    Para conferir as diretrizes de nomeação de um conector, consulte Diretrizes de nomeação de recursos do Serviço Gerenciado para Apache Kafka.

  5. Em Plug-in do conector, selecione Coletor do Cloud Storage.

  6. Especifique os tópicos de que você pode transmitir dados.

  7. Escolha o bucket de armazenamento para armazenar os dados.

  8. (Opcional) Defina outras configurações na seção Configuração.

  9. Selecione a Política de reinicialização da tarefa. Para mais informações, consulte Política de reinicialização da tarefa.

  10. Clique em Criar.

gcloud

  1. No Google Cloud console, ative o Cloud Shell.

    Ativar o Cloud Shell

    Na parte de baixo do Google Cloud console, uma sessão do Cloud Shell é iniciada e exibe um prompt de linha de comando. O Cloud Shell é um ambiente shell com a Google Cloud CLI já instalada e com valores já definidos para o projeto atual. A inicialização da sessão pode levar alguns segundos.

  2. Execute o gcloud managed-kafka connectors create comando:

    gcloud managed-kafka connectors create CONNECTOR_ID \
    --location=LOCATION \
    --connect-cluster=CONNECT_CLUSTER_ID \
    --config-file=CONFIG_FILE

    Substitua:

    • CONNECTOR_ID: o ID ou nome do conector. Para conferir as diretrizes de nomeação de um conector, consulte Diretrizes de nomeação de recursos do Serviço Gerenciado para Apache Kafka. O nome de um conector não pode ser modificado.

    • LOCATION: o local em que você cria o conector. Precisa ser o mesmo local em que você criou o cluster do Connect.

    • CONNECT_CLUSTER_ID: o ID do cluster do Connect em que o conector é criado.

    • CONFIG_FILE: o caminho para o arquivo de configuração YAML do conector.

    Confira um exemplo de arquivo de configuração do conector de coletor do Cloud Storage:

    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"

    Substitua:

    • BUCKET_NAME: o nome do bucket do Cloud Storage em que você quer gravar.

    • TOPIC_LIST: uma lista separada por vírgulas de tópicos do Kafka para ler. A lista também pode incluir expressões regulares.

Terraform

É possível usar um recurso do Terraform para criar um conector.

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
}

Para saber como aplicar ou remover uma configuração do Terraform, consulte Comandos básicos do Terraform.

Go

Antes de testar esta amostra, siga as instruções de configuração do Go em Instalar as bibliotecas de cliente. Para mais informações, consulte a documentação de referência da API Go do Serviço Gerenciado para Apache Kafka.

Para autenticar o Serviço Gerenciado para Apache Kafka, configure o Application Default Credentials(ADC). Para mais informações, consulte Configurar o ADC para um ambiente de desenvolvimento local.

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

Antes de testar esta amostra, siga as instruções de configuração do Java em Instalar as bibliotecas de cliente. Para mais informações, consulte a documentação de referência da API Java do Serviço Gerenciado para Apache Kafka.

Para autenticar o Serviço Gerenciado para Apache Kafka, configure o Application Default Credentials. Para mais informações, consulte Configurar o ADC para um ambiente de desenvolvimento local.


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

Antes de testar esta amostra, siga as instruções de configuração do Python em Instalar as bibliotecas de cliente. Para mais informações, consulte a documentação de referência da API Python do Serviço Gerenciado para Apache Kafka.

Para autenticar o Serviço Gerenciado para Apache Kafka, configure o Application Default Credentials. Para mais informações, consulte Configurar o ADC para um ambiente de desenvolvimento local.

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

Depois de criar um conector, é possível editar, excluir, pausar, interromper ou reiniciar o conector.

Configurar o conector

Para uma lista completa das propriedades específicas desse conector, consulte Configuração na documentação do conector de coletor do Cloud Storage.

Campos de saída

Por padrão, o conector grava apenas o valor do registro de cada registro do Kafka. Para incluir chaves ou cabeçalhos de registro, defina a propriedade format.output.fields como uma lista separada por vírgulas de campos a serem incluídos. Os campos aceitos são key, value, headers e timestamp. Exemplo:

format.output.fields=key,value,headers

Formatos de saída

O conector oferece suporte à gravação de registros do Kafka nos formatos CSV, JSON e Parquet. As seções a seguir descrevem como configurar o formato de saída.

Formato CSV

Para gravar arquivos CSV no Cloud Storage, defina format.output.type=csv. Use os seguintes conversores:

  • Conversor de chaves: org.apache.kafka.connect.storage.StringConverter ou org.apache.kafka.connect.json.JsonConverter

  • Conversor de valores: org.apache.kafka.connect.converters.ByteArrayConverter

Formato JSON

Para gravar arquivos formatados em JSON no Cloud Storage, defina format.output.type como um dos seguintes valores:

  • json: cada arquivo de saída contém uma única matriz de objetos JSON, em que cada objeto JSON representa um registro do Kafka.

  • jsonl: o conector grava cada registro do Kafka como um objeto JSON em uma linha separada.

O exemplo a seguir mostra como o conector grava um registro do Kafka no formato JSON:

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

Se você definir o conversor de chaves ou valores como org.apache.kafka.connect.converters.ByteArrayConverter, o conector gravará esse campo como um valor codificado em base64.

O conector não inclui um esquema JSON na saída, mesmo que value.converter.schemas.enable seja true.

Formato Parquet

Para gravar arquivos Parquet no Cloud Storage, defina format.output.type=parquet.

Se o conversor de chaves ou valores for org.apache.kafka.connect.json.JsonConverter, os dados JSON precisarão incluir um esquema. Defina a propriedade schemas.enable do conversor como true.

Para dados JSON, o conector grava os dados como um tipo struct no arquivo Parquet. Por exemplo, se o valor do registro do Kafka tiver um campo chamado userid, o conector gravará uma coluna chamada value.userid. Para nivelar os nomes das colunas, defina format.output.envelope=false. Se format.output.envelope for false, então format.output.fields precisará especificar um único campo de saída.

Configurações de arquivo do Cloud Storage

As configurações a seguir controlam como o conector grava arquivos no Cloud Storage.

  • gcs.bucket.name: obrigatório. Especifica o nome do bucket do Cloud Storage em que o conector grava dados.

  • file.compression.type: especifica o tipo de compactação dos arquivos de saída. Exemplos são gzip, snappy, zstd e none. O valor padrão é none.

  • file.name.prefix: especifica um prefixo a ser adicionado ao nome de cada arquivo. O valor padrão está vazio ("").

  • file.name.template: especifica um modelo para o nome do arquivo. O conector oferece suporte a variáveis de marcador de posição para o modelo, como {{topic}} para o nome do tópico e {{partition}} para o número da partição. Para mais informações, consulte a documentação do conector de coletor do Cloud Storage.

  • file.max.records: o número máximo de registros a serem gravados em um único arquivo. Se o valor for 0, não haverá um máximo específico. O valor padrão é 0.

  • file.max.bytes: o tamanho máximo do arquivo a ser gravado, em bytes. Se o valor for 0, não haverá um máximo específico. O valor padrão é 0.

Agrupamento de registros

Quando o conector de coletor do Cloud Storage grava registros do Kafka, ele agrupa os registros por tópico e partição por padrão. Por exemplo, os registros da partição 0 do tópico my-topic são gravados em arquivos chamados my-topic-0-OFFSET, em que OFFSET é o deslocamento inicial.

Se o parâmetro file.name.template incluir a variável de marcador de posição {{key}}, o conector agrupa os registros por chave. Nesse modo, o conector grava um arquivo separado para cada chave de registro. O conector substitui cada arquivo pelo registro mais recente dessa chave.

Credenciais

A propriedade gcs.credentials.default precisa ser definida como true. Essa configuração permite que o conector descubra as credenciais automaticamente. Google Cloud

A seguir

Apache Kafka® é uma marca registrada da The Apache Software Foundation ou afiliadas nos Estados Unidos e/ou em outros países.