O Google usa tecnologia de IA na tradução de conteúdos para seu idioma de preferência. As traduções com IA podem ter erros.

Criar um conector de coletor do BigQuery

Com um conector de coletor do BigQuery, é possível transmitir dados do Serviço Gerenciado para Apache Kafka para tabelas do BigQuery em tempo real. Você pode usar esse conector para ingerir dados de streaming no BigQuery para armazenamento em data warehouse, relatórios analíticos e painéis em tempo real.

Antes de começar

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

Crie um cluster do Serviço Gerenciado para Apache Kafka para seu cluster do Connect. Esse cluster é o cluster principal do Kafka associado ao cluster do Connect. Esse cluster também é o cluster de origem que forma uma extremidade do pipeline do conector de coletor do BigQuery.
Crie um cluster do Connect para hospedar o conector de coletor do BigQuery.
Crie um conjunto de dados do BigQuery para armazenar os dados transmitidos do Kafka.
Crie e configure um tópico do Kafka no cluster de origem. Os dados são movidos desse tópico do Kafka para o conjunto de dados de destino do BigQuery.

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 na tabela do BigQuery

O agente de serviço do Kafka gerenciado precisa ter permissão para gravar mensagens na tabela do BigQuery. Conceda ao agente de serviço o papel de Editor de dados do BigQuery (roles/bigquery.dataEditor) no projeto que contém a tabela.

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.

Principais parâmetros de configuração

Veja a seguir recomendações de configurações para o conector de coletor do BigQuery.

Tamanho da fila

A configuração queueSize controla o número máximo de solicitações de gravação para o BigQuery. Por padrão, esse valor não é configurado, o que pode resultar em erros de falta de memória no cluster do Connect. Recomendamos configurar esse valor para reduzir a chance de erros desse tipo. Os itens na fila são solicitações de gravação e não correspondem diretamente a um número de mensagens ou bytes. Portanto, é difícil saber o valor recomendado com antecedência.

Quando as mensagens são extraídas do Kafka, o conector divide o trabalho com base na tabela de destino. Se todas as mensagens em uma pesquisa forem gravadas em uma única tabela, apenas um item de trabalho será adicionado. Mas se as mensagens em uma enquete estiverem sendo gravadas em três tabelas, por exemplo, três itens de trabalho serão adicionados. O tamanho de um item de trabalho pode ser estimado com base nas configurações de sondagem de mensagens, no tamanho das mensagens do Kafka e no número de tabelas em que os dados estão sendo gravados.

Esquemas para um conector de coletor do BigQuery

O conector do BigQuery Sink usa o conversor de valor configurado (value.converter) para analisar os valores de registro do Kafka em campos. Em seguida, ele grava os campos em colunas com o mesmo nome na tabela do BigQuery.

O conector precisa de um esquema para funcionar. O esquema pode ser fornecido das seguintes maneiras:

Esquema baseado em mensagens: o esquema é incluído como parte de cada mensagem.
Esquema baseado em tabela: o conector infere o esquema de mensagens do esquema de tabela do BigQuery.
Registro de esquema: o conector lê o esquema de um registro de esquema, como o registro de esquema do Serviço Gerenciado para Apache Kafka (pré-lançamento).

As próximas seções descrevem essas opções.

Esquema baseado em mensagens

Nesse modo, cada registro do Kafka inclui um esquema JSON. O conector usa o esquema para gravar os dados do registro como uma linha da tabela do BigQuery.

Para usar esquemas baseados em mensagens, defina as seguintes propriedades no conector:

value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=true

Exemplo de valor de registro do Kafka:

{
  "schema": {
    "type": "struct",
    "fields": [
      {
        "field": "user",
        "type": "string",
        "optional": false
      },
      {
        "field": "age",
        "type": "int64",
        "optional": false
      }
    ]
  },
  "payload": {
    "user": "userId",
    "age": 30
  }
}

Se a tabela de destino já existir, o esquema da tabela do BigQuery precisará ser compatível com o esquema de mensagens incorporado. Se autoCreateTables=true, o conector criará automaticamente a tabela de destino, se necessário. Para mais informações, consulte Criação de tabelas.

Se você quiser que o conector atualize o esquema da tabela do BigQuery à medida que os esquemas de mensagens mudam, defina allowNewBigQueryFields, allowSchemaUnionization ou allowBigQueryRequiredFieldRelaxation como true.

Esquema baseado em tabela

Nesse modo, os registros do Kafka contêm dados JSON simples sem um esquema explícito. O conector infere o esquema da tabela de destino.

Requisitos:

A tabela do BigQuery já precisa existir.
Os dados do registro do Kafka precisam ser compatíveis com o esquema da tabela.
Esse modo não é compatível com atualizações dinâmicas de esquema com base nas mensagens recebidas.

Para usar esquemas baseados em tabelas, defina as seguintes propriedades no conector:

value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false
bigQueryPartitionDecorator=false

Se a tabela do BigQuery usar particionamento baseado em tempo com particionamento diário, bigQueryPartitionDecorator poderá ser true. Caso contrário, defina essa propriedade como false.

Exemplo de valor de registro do Kafka:

{
  "user": "userId",
  "age": 30
}

Registro de esquema

Nesse modo, cada registro do Kafka contém dados do Apache Avro, e o esquema de mensagens é armazenado em um registro de esquemas.

Para usar o conector do coletor do BigQuery com um registro de esquema, defina as seguintes propriedades no conector:

value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=SCHEMA_REGISTRY_URL

Substitua SCHEMA_REGISTRY_URL pelo URL do registro de esquema.

Para usar o conector com o registro de esquema do Serviço Gerenciado para Apache Kafka, defina a seguinte propriedade:

value.converter.bearer.auth.credentials.source=GCP

Para mais informações, consulte Usar o Kafka Connect com o registro de esquema.

Tabelas gerenciadas do Apache Iceberg

O conector do BigQuery Sink é compatível com tabelas gerenciadas do Apache Iceberg (daqui em diante, tabelas gerenciadas do Iceberg) como um destino de coletor.

As tabelas gerenciadas do Iceberg fornecem a base para criar lakehouses de formato aberto no Google Cloud. As tabelas gerenciadas do Iceberg oferecem a mesma experiência totalmente gerenciada que as tabelas do BigQuery, mas armazenam dados em buckets de armazenamento de propriedade do cliente usando o Parquet para serem interoperáveis com os formatos de tabela aberta do Apache Iceberg.

Para saber como criar uma tabela do Apache Iceberg, consulte Criar uma tabela do Apache Iceberg.

Criar um conector de coletor do BigQuery

Console

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

Acessar clusters do Connect
Clique no cluster do Connect em que você quer criar o conector.
Clique em Criar conector.
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.
Em Plug-in do conector, selecione Coletor do BigQuery.
Na seção Tópicos, especifique os tópicos do Kafka de onde os dados serão lidos. É possível especificar uma lista de tópicos ou uma expressão regular para corresponder aos nomes dos tópicos.
- Opção 1: escolha Selecionar uma lista de tópicos do Kafka. Na lista Tópicos do Kafka, selecione um ou mais tópicos. Clique em OK.
- Opção 2: escolha Usar uma regex do tópico. No campo Regex do tópico, insira uma expressão regular.
Clique em Conjunto de dados e especifique um conjunto de dados do BigQuery. É possível escolher um conjunto de dados atual ou criar um novo.
Opcional: na caixa Configurações, adicione propriedades de configuração ou edite as propriedades padrão. Para mais informações, consulte Configurar o conector.
Selecione a Política de reinicialização da tarefa. Para mais informações, consulte Política de reinicialização de tarefas.
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 BigQuery:

name: "BQ_SINK_CONNECTOR_ID"
project: "GCP_PROJECT_ID"
topics: "GMK_TOPIC_ID"
tasks.max: 3
connector.class: "com.wepay.kafka.connect.bigquery.BigQuerySinkConnector"
key.converter: "org.apache.kafka.connect.storage.StringConverter"
value.converter: "org.apache.kafka.connect.json.JsonConverter"
value.converter.schemas.enable: "false"
defaultDataset: "BQ_DATASET_ID"

Substitua:

BQ_SINK_CONNECTOR_ID: o ID ou nome do conector do coletor do BigQuery. Para conferir as diretrizes de nomeação de conectores, acesse Diretrizes de nomeação de recursos do Serviço Gerenciado para Apache Kafka. O nome de um conector é imutável.
GCP_PROJECT_ID: o ID do projeto do Google Cloud em que o conjunto de dados do BigQuery está localizado.
GMK_TOPIC_ID: o ID do tópico do Serviço Gerenciado para Apache Kafka de onde os dados fluem para o conector do coletor do BigQuery.
BQ_DATASET_ID: o ID do conjunto de dados do BigQuery que atua como o coletor do pipeline.

Terraform

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

resource "google_managed_kafka_connector" "example-bigquery-sink-connector" {
  project         = data.google_project.default.project_id
  connector_id    = "my-bigquery-sink-connector"
  connect_cluster = google_managed_kafka_connect_cluster.default.connect_cluster_id
  location        = "us-central1"

  configs = {
    "name"                           = "my-bigquery-sink-connector"
    "project"                        = data.google_project.default.project_id
    "topics"                         = "GMK_TOPIC_ID"
    "tasks.max"                      = "3"
    "connector.class"                = "com.wepay.kafka.connect.bigquery.BigQuerySinkConnector"
    "key.converter"                  = "org.apache.kafka.connect.storage.StringConverter"
    "value.converter"                = "org.apache.kafka.connect.json.JsonConverter"
    "value.converter.schemas.enable" = "false"
    "defaultDataset"                 = "BQ_DATASET_ID"
  }

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

// createBigQuerySinkConnector creates a BigQuery Sink connector.
func createBigQuerySinkConnector(w io.Writer, projectID, region, connectClusterID, connectorID, topics, tasksMax, keyConverter, valueConverter, valueConverterSchemasEnable, defaultDataset 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 := "BQ_SINK_CONNECTOR_ID"
	// topics := "GMK_TOPIC_ID"
	// tasksMax := "3"
	// keyConverter := "org.apache.kafka.connect.storage.StringConverter"
	// valueConverter := "org.apache.kafka.connect.json.JsonConverter"
	// valueConverterSchemasEnable := "false"
	// defaultDataset := "BQ_DATASET_ID"
	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)

	// BigQuery Sink sample connector configuration
	config := map[string]string{
		"name":                           connectorID,
		"project":                        projectID,
		"topics":                         topics,
		"tasks.max":                      tasksMax,
		"connector.class":                "com.wepay.kafka.connect.bigquery.BigQuerySinkConnector",
		"key.converter":                  keyConverter,
		"value.converter":                valueConverter,
		"value.converter.schemas.enable": valueConverterSchemasEnable,
		"defaultDataset":                 defaultDataset,
	}

	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 BigQuery 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 CreateBigQuerySinkConnector {

  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-bigquery-sink-connector";
    String bigqueryProjectId = "my-bigquery-project-id";
    String datasetName = "my-dataset";
    String kafkaTopicName = "kafka-topic";
    String maxTasks = "3";
    String connectorClass = "com.wepay.kafka.connect.bigquery.BigQuerySinkConnector";
    String keyConverter = "org.apache.kafka.connect.storage.StringConverter";
    String valueConverter = "org.apache.kafka.connect.json.JsonConverter";
    String valueSchemasEnable = "false";
    createBigQuerySinkConnector(
        projectId,
        region,
        connectClusterId,
        connectorId,
        bigqueryProjectId,
        datasetName,
        kafkaTopicName,
        maxTasks,
        connectorClass,
        keyConverter,
        valueConverter,
        valueSchemasEnable);
  }

  public static void createBigQuerySinkConnector(
      String projectId,
      String region,
      String connectClusterId,
      String connectorId,
      String bigqueryProjectId,
      String datasetName,
      String kafkaTopicName,
      String maxTasks,
      String connectorClass,
      String keyConverter,
      String valueConverter,
      String valueSchemasEnable)
      throws Exception {

    // Build the connector configuration
    Map<String, String> configMap = new HashMap<>();
    configMap.put("name", connectorId);
    configMap.put("project", bigqueryProjectId);
    configMap.put("topics", kafkaTopicName);
    configMap.put("tasks.max", maxTasks);
    configMap.put("connector.class", connectorClass);
    configMap.put("key.converter", keyConverter);
    configMap.put("value.converter", valueConverter);
    configMap.put("value.converter.schemas.enable", valueSchemasEnable);
    configMap.put("defaultDataset", datasetName);

    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 BigQuery 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 = {
    "name": connector_id,
    "project": project_id,
    "topics": topics,
    "tasks.max": tasks_max,
    "connector.class": "com.wepay.kafka.connect.bigquery.BigQuerySinkConnector",
    "key.converter": key_converter,
    "value.converter": value_converter,
    "value.converter.schemas.enable": value_converter_schemas_enable,
    "defaultDataset": default_dataset,
}

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

Nesta seção, descrevemos algumas propriedades de configuração que podem ser definidas no conector. Para uma lista completa das propriedades específicas desse conector, consulte Configurações do conector de coletor do BigQuery.

Nome da tabela

Por padrão, o conector usa o nome do tópico como o nome da tabela do BigQuery. Para usar um nome de tabela diferente, defina a propriedade topic2TableMap com o seguinte formato:

topic2TableMap=TOPIC_1:TABLE_1,TOPIC_2:TABLE_2,...

Criação de tabelas

O conector de coletor do BigQuery pode criar as tabelas de destino se elas não existirem.

Se for autoCreateTables=true, o conector vai tentar criar as tabelas do BigQuery que não existem. Essa configuração é o comportamento padrão.
Se for autoCreateTables=false, o conector não vai criar tabelas. Se uma tabela de destino não existir, ocorrerá um erro.

Quando autoCreateTables é true, é possível usar as seguintes propriedades de configuração para ter um controle mais refinado sobre como o conector cria e configura novas tabelas:

allBQFieldsNullable
clusteringPartitionFieldNames
convertDoubleSpecialValues
partitionExpirationMs
sanitizeFieldNames
sanitizeTopics
timestampPartitionFieldName

Para informações sobre essas propriedades, consulte Configurações do conector do coletor do BigQuery.

Metadados do Kafka

É possível mapear outros dados do Kafka, como informações de metadados e informações importantes, para a tabela do BigQuery configurando os campos kafkaDataFieldName e kafkaKeyFieldName, respectivamente. Exemplos de informações de metadados incluem o tópico, a partição, o deslocamento e o horário de inserção do Kafka.

A seguir

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

Criar um conector de coletor do BigQuery Mantenha tudo organizado com as coleções Salve e categorize o conteúdo com base nas suas preferências.

Antes de começar

Papéis e permissões necessárias

Permissões necessárias

Conceder permissões para gravar na tabela do BigQuery

Principais parâmetros de configuração

Tamanho da fila

Esquemas para um conector de coletor do BigQuery

Esquema baseado em mensagens

Esquema baseado em tabela

Registro de esquema

Tabelas gerenciadas do Apache Iceberg

Criar um conector de coletor do BigQuery

Console

gcloud

Terraform

Go

Java

Python

Configurar o conector

Nome da tabela

Criação de tabelas

Metadados do Kafka

A seguir

Criar um conector de coletor do BigQuery