Google utiliza tecnología de IA para traducir contenido a tu idioma preferido. Las traducciones realizadas con IA pueden contener errores.

Crea un conector de BigQuery Sink

Un conector de receptor de BigQuery te permite transmitir datos de Managed Service para Apache Kafka a tablas de BigQuery en tiempo real. Puedes usar este conector para transferir datos de transmisión a BigQuery para el almacenamiento de datos, los informes analíticos y la potenciación de paneles en tiempo real.

Antes de comenzar

Antes de crear un conector de receptor de BigQuery, asegúrate de tener lo siguiente:

Crea un clúster de Managed Service para Apache Kafka para tu clúster de Connect. Este clúster es el clúster principal de Kafka asociado con el clúster de Connect. Este clúster también es el clúster de origen que forma un extremo de la canalización del conector de receptor de BigQuery.
Crea un clúster de Connect para alojar tu conector de receptor de BigQuery.
Crea un conjunto de datos de BigQuery para almacenar los datos transmitidos desde Kafka.
Crea y configura un tema de Kafka dentro del clúster de origen. Los datos se mueven desde este tema de Kafka al conjunto de datos de BigQuery de destino.

Roles y permisos requeridos

Para obtener los permisos que necesitas para crear un conector, pídele a tu administrador que te otorgue el rol de IAM de editor de conectores de Kafka administrados (roles/managedkafka.connectorEditor) en tu proyecto. Para obtener más información sobre cómo otorgar roles, consulta Administra el acceso a proyectos, carpetas y organizaciones.

Este rol predefinido contiene los permisos necesarios para crear un conector. Para ver los permisos exactos que son necesarios, expande la sección Permisos requeridos:

Permisos necesarios

Se requieren los siguientes permisos para crear un conector:

Crea un conector: managedkafka.connectors.create

También puedes obtener estos permisos con roles personalizados o con otros roles predefinidos.

Otorga permisos para escribir en la tabla de BigQuery

El agente de servicio de Kafka administrado debe tener permiso para escribir mensajes en la tabla de BigQuery. Otorga al agente de servicio el rol de editor de datos de BigQuery (roles/bigquery.dataEditor) en el proyecto que contiene la tabla.

Si tu clúster de Connect está en un proyecto diferente al de tu clúster de Managed Service para Apache Kafka, consulta Crea un clúster de Connect en un proyecto diferente.

Parámetros de configuración clave

A continuación, se incluyen recomendaciones para configurar el conector de receptor de BigQuery.

Tamaño de la cola

El parámetro de configuración queueSize controla la cantidad máxima de solicitudes de escritura en BigQuery. De forma predeterminada, este valor no está configurado, lo que puede generar errores de memoria insuficiente en el clúster de Connect. Te recomendamos que configures este valor para reducir la probabilidad de que se produzcan esos errores. Los elementos de la cola son solicitudes de escritura y no corresponden directamente a una cantidad de mensajes o bytes. Por lo tanto, es difícil conocer el valor recomendado de antemano.

Cuando se sondean los mensajes de Kafka, el conector divide el trabajo según su tabla de destino. Si todos los mensajes de un sondeo se deben escribir en una sola tabla, solo se agrega un elemento de trabajo. Sin embargo, si los mensajes de una encuesta se escriben en tres tablas, por ejemplo, se agregan tres elementos de trabajo. El tamaño de un elemento de trabajo se puede estimar de forma aproximada según las configuraciones de sondeo de mensajes, el tamaño de los mensajes de Kafka y la cantidad de tablas en las que se escribe.

Esquemas para un conector de receptor de BigQuery

El conector de BigQuery Sink usa el convertidor de valores configurado (value.converter) para analizar los valores de los registros de Kafka en campos. Luego, escribe los campos en las columnas con el mismo nombre en la tabla de BigQuery.

El conector requiere un esquema para funcionar. El esquema se puede proporcionar de las siguientes maneras:

Esquema basado en mensajes: El esquema se incluye como parte de cada mensaje.
Esquema basado en tablas: El conector infiere el esquema del mensaje a partir del esquema de la tabla de BigQuery.
Registro de esquemas: El conector lee el esquema de un registro de esquemas, como el registro de esquemas de Managed Service para Apache Kafka (versión preliminar).

En las siguientes secciones, se describen estas opciones.

Esquema basado en mensajes

En este modo, cada registro de Kafka incluye un esquema JSON. El conector usa el esquema para escribir los datos del registro como una fila de la tabla de BigQuery.

Para usar esquemas basados en mensajes, establece las siguientes propiedades en el conector:

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

Ejemplo de valor de registro de Kafka:

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

Si la tabla de destino ya existe, el esquema de la tabla de BigQuery debe ser compatible con el esquema del mensaje incorporado. Si es autoCreateTables=true, el conector crea automáticamente la tabla de destino si es necesario. Para obtener más información, consulta Creación de tablas.

Si quieres que el conector actualice el esquema de la tabla de BigQuery a medida que cambian los esquemas de mensajes, establece allowNewBigQueryFields, allowSchemaUnionization o allowBigQueryRequiredFieldRelaxation en true.

Esquema basado en tablas

En este modo, los registros de Kafka contienen datos JSON sin formato y sin un esquema explícito. El conector infiere el esquema de la tabla de destino.

Requisitos:

La tabla de BigQuery ya debe existir.
Los datos de los registros de Kafka deben ser compatibles con el esquema de la tabla.
Este modo no admite actualizaciones dinámicas de esquemas basadas en mensajes entrantes.

Para usar esquemas basados en tablas, establece las siguientes propiedades en el conector:

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

Si la tabla de BigQuery usa partición basada en el tiempo con partición diaria, bigQueryPartitionDecorator puede ser true. De lo contrario, establece esta propiedad en false.

Ejemplo de valor de registro de Kafka:

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

Registro de esquemas

En este modo, cada registro de Kafka contiene datos de Apache Avro, y el esquema del mensaje se almacena en un registro de esquemas.

Para usar el conector de BigQuery Sink con un registro de esquemas, establece las siguientes propiedades en el conector:

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

Reemplaza SCHEMA_REGISTRY_URL por la URL del registro de esquemas.

Para usar el conector con el registro de esquemas de Managed Service para Apache Kafka, establece la siguiente propiedad:

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

Para obtener más información, consulta Usa Kafka Connect con el registro de esquemas.

Tablas administradas de Apache Iceberg

El conector de BigQuery Sink admite tablas administradas de Apache Iceberg (en adelante, tablas administradas de Iceberg) como destino de receptor.

Las tablas administradas de Iceberg proporcionan la base para compilar lakehouses de formato abierto en Google Cloud. Las tablas administradas de Iceberg ofrecen la misma experiencia completamente administrada que las tablas de BigQuery, pero almacenan datos en buckets de almacenamiento que pertenecen al cliente con Parquet para ser interoperables con los formatos de tablas abiertas de Apache Iceberg.

Para obtener información sobre cómo crear una tabla de Apache Iceberg, consulta Crea una tabla de Apache Iceberg.

Crea un conector de receptor de BigQuery

Console

En la consola de Google Cloud , ve a la página Conectar clústeres.

Ir a Conectar clústeres
Haz clic en el clúster de Connect en el que deseas crear el conector.
Haz clic en Crear conector.
Para el nombre del conector, ingresa una cadena.

Si necesitas ayuda para asignarle un nombre a un conector, consulta los Lineamientos para asignarles nombres a los recursos de Managed Service para Apache Kafka.
En Complemento de conector, selecciona Receptor de BigQuery.
En la sección Temas, especifica los temas de Kafka desde los que se leerá. Puedes especificar una lista de temas o una expresión regular para que coincida con los nombres de los temas.
- Opción 1: Elige Seleccionar una lista de temas de Kafka. En la lista Temas de Kafka, selecciona uno o más temas. Haz clic en OK.
- Opción 2: Elige Usar una regex del tema. En el campo Regex del tema, ingresa una expresión regular.
Haz clic en Conjunto de datos y especifica un conjunto de datos de BigQuery. Puedes elegir un conjunto de datos existente o crear uno nuevo.
Opcional: En el cuadro Configuraciones, agrega propiedades de configuración o edita las propiedades predeterminadas. Para obtener más información, consulta Configura el conector.
Selecciona la Política de reinicio de tareas. Para obtener más información, consulta Política de reinicio de tareas.
Haz clic en Crear.

gcloud

Ejecuta el comando gcloud managed-kafka connectors create:

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

Reemplaza lo siguiente:

CONNECTOR_ID: Es el ID o el nombre del conector.
CONNECT_CLUSTER_ID: Es el ID del clúster de Connect en el que se crea el conector.
LOCATION: Es la ubicación del clúster de Connect.
CONFIG_FILE: Es la ruta de acceso a un archivo de configuración YAML o JSON local.

A continuación, se muestra un ejemplo de un archivo de configuración para el conector de receptor de 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"

Reemplaza lo siguiente:

BQ_SINK_CONNECTOR_ID: Es el ID o el nombre del conector de BigQuery Sink. Si necesitas ayuda para asignarle un nombre a un conector, consulta los Lineamientos para asignarles nombres a los recursos de Managed Service para Apache Kafka. El nombre de un conector es inmutable.
GCP_PROJECT_ID: Es el ID del proyecto de Google Cloud en el que reside tu conjunto de datos de BigQuery.
GMK_TOPIC_ID: Es el ID del tema de Managed Service para Apache Kafka desde el que fluyen los datos al conector de BigQuery Sink.
BQ_DATASET_ID: Es el ID del conjunto de datos de BigQuery que actúa como receptor de la canalización.

Terraform

Puedes usar un recurso de Terraform para crear un 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
}

Si deseas obtener más información para aplicar o quitar una configuración de Terraform, consulta los comandos básicos de Terraform.

Go

Antes de probar esta muestra, sigue las instrucciones de configuración de Go en Instala las bibliotecas cliente. Para obtener más información, consulta la documentación de referencia de la API de Go de Managed Service para Apache Kafka.

Para autenticarte en Managed Service para Apache Kafka, configura las credenciales predeterminadas de la aplicación(ADC). Para obtener más información, consulta Configura ADC para un entorno de desarrollo 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 probar esta muestra, sigue las instrucciones de configuración de Java en Instala las bibliotecas cliente. Para obtener más información, consulta la documentación de referencia de la API de Java de Managed Service para Apache Kafka.

Para autenticarte en Managed Service para Apache Kafka, configura las credenciales predeterminadas de la aplicación. Para obtener más información, consulta Configura ADC para un entorno de desarrollo 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 probar este ejemplo, sigue las instrucciones de configuración para Python que encontrarás en Instala las bibliotecas cliente. Para obtener más información, consulta la documentación de referencia de la API de Python de Managed Service para Apache Kafka.

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

Después de crear un conector, puedes editarlo, borrarlo, pausarlo, detenerlo o reiniciarlo.

Configura el conector

En esta sección, se describen algunas propiedades de configuración que puedes establecer en el conector. Para obtener una lista completa de las propiedades específicas de este conector, consulta Configuración del conector de BigQuery Sink.

Nombre de la tabla

De forma predeterminada, el conector usa el nombre del tema como el nombre de la tabla de BigQuery. Para usar un nombre de tabla diferente, configura la propiedad topic2TableMap con el siguiente formato:

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

Creación de tablas

El conector de receptor de BigQuery puede crear las tablas de destino si no existen.

Si es autoCreateTables=true, el conector intenta crear las tablas de BigQuery que no existen. Este parámetro de configuración es el comportamiento predeterminado.
Si es autoCreateTables=false, el conector no crea ninguna tabla. Si no existe una tabla de destino, se produce un error.

Cuando autoCreateTables es true, puedes usar las siguientes propiedades de configuración para tener un control más detallado sobre cómo el conector crea y configura tablas nuevas:

allBQFieldsNullable
clusteringPartitionFieldNames
convertDoubleSpecialValues
partitionExpirationMs
sanitizeFieldNames
sanitizeTopics
timestampPartitionFieldName

Para obtener información sobre estas propiedades, consulta Parámetros de configuración del conector de BigQuery Sink.

Metadatos de Kafka

Puedes asignar datos adicionales de Kafka, como información de metadatos y de claves, a la tabla de BigQuery configurando los campos kafkaDataFieldName y kafkaKeyFieldName, respectivamente. Entre los ejemplos de información de metadatos, se incluyen el tema, la partición, el desplazamiento y la hora de inserción de Kafka.

Próximos pasos

Apache Kafka® es una marca registrada de The Apache Software Foundation o sus afiliados en Estados Unidos y otros países.

Crea un conector de BigQuery Sink Organiza tus páginas con colecciones Guarda y categoriza el contenido según tus preferencias.