Criar um conector de gravador do Cloud Storage

Um conector de coletor do Cloud Storage transmite dados do Serviço Gerenciado para Apache Kafka para um Cloud Storage, o que permite armazenar e processar grandes volumes de dados de maneira econômica e escalonável.

Os casos de uso comuns desse conector incluem:

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

  • Arquivar 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 receber 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 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:

  • Crie um conector: managedkafka.connectors.create

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

Conceder permissões para gravar no bucket do Cloud Storage

O agente 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

Para conceder essas permissões, atribua ao agente de serviço o papel de Usuário de objetos do Storage (roles/storage.objectUser) no projeto que contém o bucket do Cloud Storage.

Se o cluster do Connect estiver em um projeto diferente do cluster do Serviço Gerenciado para Apache Kafka, 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 console do Google Cloud , 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 conectores, acesse 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 onde é possível 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 de tarefas.

  10. Clique em Criar.

gcloud

Execute o comando gcloud managed-kafka connectors create:

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

Substitua:

  • CONNECTOR_ID: o ID ou nome do conector.
  • CONNECT_CLUSTER_ID: o ID do cluster do Connect em que o conector é criado.
  • LOCATION: o local do cluster do Connect.
  • CONFIG_FILE: o caminho para um arquivo de configuração YAML ou JSON local.

Veja um exemplo de arquivo de configuração para o 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 os dados serão gravados.

  • TOPIC_LIST: uma lista separada por vírgulas de tópicos do Kafka para leitura. 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 este exemplo, 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 no 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 se autenticar no 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 este exemplo, 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 se autenticar no 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 editá-lo, excluí-lo, pausá-lo, interrompê-lo ou reiniciá-lo.

Configurar o conector

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

Campos de saída

Por padrão, o conector grava apenas o valor 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 é compatível com a 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 valor: org.apache.kafka.connect.converters.ByteArrayConverter

Formato JSON

Para gravar arquivos no formato 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 chave ou valor como org.apache.kafka.connect.converters.ByteArrayConverter, o conector vai 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 chave ou valor 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 vai gravar uma coluna chamada value.userid. Para nivelar os nomes das colunas, defina format.output.envelope=false. Se format.output.envelope for false, format.output.fields precisará especificar um único campo de saída.

Configurações de arquivos 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. Por exemplo, 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 é 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 do 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, por padrão, ele agrupa os registros por tópico e partiçã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 vai agrupar 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 do Google Cloud automaticamente.

A seguir

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