Google utilizza la tecnologia AI per tradurre i contenuti nella tua lingua preferita. Le traduzioni generate dall'AI potrebbero contenere errori.

Crea un connettore BigQuery Sink

Un connettore di sink BigQuery ti consente di trasmettere i dati da Managed Service per Apache Kafka alle tabelle BigQuery in tempo reale. Puoi utilizzare questo connettore per importare dati in modalità flusso in BigQuery per il data warehousing, i report analitici e l'alimentazione di dashboard in tempo reale.

Prima di iniziare

Prima di creare un connettore sink BigQuery, assicurati di disporre di quanto segue:

Crea un cluster Managed Service per Apache Kafka per il tuo cluster Connect. Questo cluster è il cluster Kafka principale associato al cluster di connessione. Questo cluster è anche il cluster di origine che costituisce un'estremità della pipeline del connettore di sink BigQuery.
Crea un cluster Connect per ospitare il connettore sink BigQuery.
Crea un set di dati BigQuery per archiviare i dati trasmessi in streaming da Kafka.
Crea e configura un argomento Kafka all'interno del cluster di origine. I dati vengono spostati da questo argomento Kafka al set di dati BigQuery di destinazione.

Ruoli e autorizzazioni richiesti

Per ottenere le autorizzazioni necessarie per creare un connettore, chiedi all'amministratore di concederti il ruolo IAM Managed Kafka Connector Editor (roles/managedkafka.connectorEditor) nel progetto. Per saperne di più sulla concessione dei ruoli, consulta Gestisci l'accesso a progetti, cartelle e organizzazioni.

Questo ruolo predefinito contiene le autorizzazioni necessarie per creare un connettore. Per vedere quali sono esattamente le autorizzazioni richieste, espandi la sezione Autorizzazioni obbligatorie:

Autorizzazioni obbligatorie

Per creare un connettore sono necessarie le seguenti autorizzazioni:

Crea un connettore: managedkafka.connectors.create

Potresti anche ottenere queste autorizzazioni con ruoli personalizzati o altri ruoli predefiniti.

Concedi le autorizzazioni per scrivere nella tabella BigQuery

L'agente di servizio Managed Kafka deve disporre dell'autorizzazione per scrivere messaggi nella tabella BigQuery. Concedi all'agente di servizio il ruolo BigQuery Data Editor (roles/bigquery.dataEditor) nel progetto che contiene la tabella.

Se il cluster di connessione si trova in un progetto diverso da quello del cluster Managed Service per Apache Kafka, consulta Crea un cluster di connessione in un progetto diverso.

Parametri di configurazione principali

Di seguito sono riportati i consigli per le configurazioni del connettore di sink BigQuery.

Dimensione della coda

L'impostazione queueSize controlla il numero massimo di richieste di scrittura in BigQuery. Per impostazione predefinita, questo valore non è configurato, il che può causare errori di memoria insufficiente nel cluster Connect. Ti consigliamo di configurare questo valore per ridurre la probabilità di questi errori. Gli elementi nella coda sono richieste di scrittura e non corrispondono direttamente a un numero di messaggi o byte. Pertanto, è difficile conoscere in anticipo il valore consigliato.

Quando i messaggi vengono sottoposti a polling da Kafka, il connettore suddivide il lavoro in base alla tabella di destinazione. Se tutti i messaggi di un sondaggio devono essere scritti in una singola tabella, viene aggiunto un solo elemento di lavoro. Tuttavia, se i messaggi di un sondaggio vengono scritti in tre tabelle, ad esempio, vengono aggiunti tre elementi di lavoro. La dimensione di un elemento di lavoro può essere stimata approssimativamente in base alle configurazioni di polling dei messaggi, alla dimensione dei messaggi Kafka e al numero di tabelle in cui vengono scritti i dati.

Schemi per un connettore sink BigQuery

Il connettore di sink BigQuery utilizza il convertitore di valori configurato (value.converter) per analizzare i valori dei record Kafka nei campi. Quindi scrive i campi nelle colonne con lo stesso nome nella tabella BigQuery.

Per funzionare, il connettore richiede uno schema. Lo schema può essere fornito nei seguenti modi:

Schema basato su messaggi: lo schema è incluso in ogni messaggio.
Schema basato su tabella: il connettore deduce lo schema dei messaggi dallo schema della tabella BigQuery.
Registro di schema: il connettore legge lo schema da un registro di schema, ad esempio il registro di schema Managed Service per Apache Kafka (anteprima).

Le sezioni successive descrivono queste opzioni.

Schema basato su messaggi

In questa modalità, ogni record Kafka include uno schema JSON. Il connettore utilizza lo schema per scrivere i dati del record come riga di una tabella BigQuery.

Per utilizzare gli schemi basati su messaggi, imposta le seguenti proprietà sul connettore:

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

Valore di esempio del record Kafka:

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

Se la tabella di destinazione esiste già, lo schema della tabella BigQuery deve essere compatibile con lo schema dei messaggi incorporati. Se autoCreateTables=true, il connettore crea automaticamente la tabella di destinazione, se necessario. Per saperne di più, consulta Creazione di tabelle.

Se vuoi che il connettore aggiorni lo schema della tabella BigQuery man mano che gli schemi dei messaggi cambiano, imposta allowNewBigQueryFields, allowSchemaUnionization o allowBigQueryRequiredFieldRelaxation su true.

Schema basato su tabelle

In questa modalità, i record Kafka contengono dati JSON non elaborati senza uno schema esplicito. Il connettore deduce lo schema dalla tabella di destinazione.

Requisiti:

La tabella BigQuery deve già esistere.
I dati dei record Kafka devono essere compatibili con lo schema della tabella.
Questa modalità non supporta gli aggiornamenti dinamici dello schema in base ai messaggi in arrivo.

Per utilizzare gli schemi basati su tabelle, imposta le seguenti proprietà sul connettore:

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

Se la tabella BigQuery utilizza il partizionamento basato sul tempo con partizionamento giornaliero, bigQueryPartitionDecorator può essere true. In caso contrario, imposta questa proprietà su false.

Valore di esempio del record Kafka:

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

Registro di schema

In questa modalità, ogni record Kafka contiene dati Apache Avro e lo schema dei messaggi viene archiviato in un registro degli schemi.

Per utilizzare il connettore sink BigQuery con un registro degli schemi, imposta le seguenti proprietà sul connettore:

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

Sostituisci SCHEMA_REGISTRY_URL con l'URL del registry dello schema.

Per utilizzare il connettore con il registry dello schema di Managed Service per Apache Kafka, imposta la seguente proprietà:

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

Per saperne di più, consulta Utilizza Kafka Connect con il registry degli schemi.

Tabelle gestite di Apache Iceberg

Il connettore sink BigQuery supporta le tabelle gestite Apache Iceberg (di seguito, tabelle gestite Iceberg) come destinazione sink.

Le tabelle gestite Iceberg forniscono le basi per la creazione di lakehouse in formato aperto su Google Cloud. Le tabelle gestite da Iceberg offrono la stessa esperienza completamente gestita delle tabelle BigQuery, ma archiviano i dati in bucket di archiviazione di proprietà del cliente utilizzando Parquet per essere interoperabili con i formati di tabella aperti Apache Iceberg.

Per informazioni su come creare una tabella Apache Iceberg, consulta Crea una tabella Apache Iceberg.

Crea un connettore di sink BigQuery

Console

Nella console Google Cloud , vai alla pagina Connetti i cluster.

Vai a Cluster connessi
Fai clic sul cluster di connessione in cui vuoi creare il connettore.
Fai clic su Crea connettore.
Inserisci una stringa per il nome del connettore.

Per maggiori informazioni su come assegnare un nome a un connettore, consulta le linee guida per assegnare un nome a una risorsa Managed Service per Apache Kafka.
In Plug-in connettore, seleziona BigQuery Sink.
Nella sezione Argomenti, specifica gli argomenti Kafka da cui leggere. Puoi specificare un elenco di argomenti o un'espressione regolare per trovare corrispondenze con i nomi degli argomenti.
- Opzione 1: scegli Seleziona un elenco di argomenti Kafka. Nell'elenco Argomenti Kafka, seleziona uno o più argomenti. Fai clic su OK.
- Opzione 2: scegli Utilizza un'espressione regolare per gli argomenti. Nel campo Regex argomento, inserisci un'espressione regolare.
Fai clic su Set di dati e specifica un set di dati BigQuery. Puoi scegliere un set di dati esistente o crearne uno nuovo.
(Facoltativo) Nella casella Configurazioni, aggiungi le proprietà di configurazione o modifica le proprietà predefinite. Per saperne di più, consulta Configura il connettore.
Seleziona la Policy di riavvio attività. Per saperne di più, consulta Policy di riavvio delle attività.
Fai clic su Crea.

gcloud

Esegui il comando gcloud managed-kafka connectors create:

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

Sostituisci quanto segue:

CONNECTOR_ID: l'ID o il nome del connettore.
CONNECT_CLUSTER_ID: l'ID del cluster Connect in cui viene creato il connettore.
LOCATION: la località del cluster Connect.
CONFIG_FILE: il percorso di un file di configurazione YAML o JSON locale.

Ecco un esempio di file di configurazione per il connettore sink 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"

Sostituisci quanto segue:

BQ_SINK_CONNECTOR_ID: l'ID o il nome del connettore sink BigQuery. Per maggiori informazioni su come assegnare un nome a un connettore, consulta le linee guida per assegnare un nome a una risorsa Managed Service per Apache Kafka. Il nome di un connettore è immutabile.
GCP_PROJECT_ID: l'ID del progetto Google Cloud in cui si trova il set di dati BigQuery.
GMK_TOPIC_ID: l'ID dell'argomento Managed Service per Apache Kafka da cui i dati fluiscono al connettore sink BigQuery.
BQ_DATASET_ID: l'ID del set di dati BigQuery che funge da sink per la pipeline.

Terraform

Puoi utilizzare una risorsa Terraform per creare un connettore.

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
}

Per scoprire come applicare o rimuovere una configurazione Terraform, consulta Comandi Terraform di base.

Vai

Prima di provare questo esempio, segui le istruzioni di configurazione di Go in Installa le librerie client. Per saperne di più, consulta la documentazione di riferimento dell'API Go di Managed Service per Apache Kafka.

Per eseguire l'autenticazione in Managed Service per Apache Kafka, configura le Credenziali predefinite dell'applicazione(ADC). Per saperne di più, consulta Configura ADC per un ambiente di sviluppo locale.

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

Prima di provare questo esempio, segui le istruzioni di configurazione di Java riportate in Installa le librerie client. Per saperne di più, consulta la documentazione di riferimento dell'API Java di Managed Service per Apache Kafka.

Per eseguire l'autenticazione in Managed Service per Apache Kafka, configura le Credenziali predefinite dell'applicazione. Per saperne di più, consulta Configura ADC per un ambiente di sviluppo locale.


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

Prima di provare questo esempio, segui le istruzioni di configurazione di Python in Installa le librerie client. Per saperne di più, consulta la documentazione di riferimento dell'API Python di Managed Service per 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}")

Dopo aver creato un connettore, puoi modificarlo, eliminarlo, metterlo in pausa, arrestarlo o riavviarlo.

Configura il connettore

Questa sezione descrive alcune proprietà di configurazione che puoi impostare sul connettore. Per un elenco completo delle proprietà specifiche di questo connettore, consulta Configurazioni del connettore sink BigQuery.

Nome tabella

Per impostazione predefinita, il connettore utilizza il nome dell'argomento come nome della tabella BigQuery. Per utilizzare un nome di tabella diverso, imposta la proprietà topic2TableMap con il seguente formato:

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

Creazione della tabella

Il connettore sink BigQuery può creare le tabelle di destinazione se non esistono.

Se autoCreateTables=true, il connettore tenta di creare le tabelle BigQuery che non esistono. Questa è l'impostazione predefinita.
Se autoCreateTables=false, il connettore non crea tabelle. Se una tabella di destinazione non esiste, si verifica un errore.

Quando autoCreateTables è true, puoi utilizzare le seguenti proprietà di configurazione per un controllo più granulare sul modo in cui il connettore crea e configura le nuove tabelle:

allBQFieldsNullable
clusteringPartitionFieldNames
convertDoubleSpecialValues
partitionExpirationMs
sanitizeFieldNames
sanitizeTopics
timestampPartitionFieldName

Per informazioni su queste proprietà, consulta Configurazioni del connettore sink BigQuery.

Metadati Kafka

Puoi mappare dati aggiuntivi da Kafka, come informazioni sui metadati e informazioni sulle chiavi, nella tabella BigQuery configurando rispettivamente i campi kafkaDataFieldName e kafkaKeyFieldName. Esempi di informazioni sui metadati includono l'argomento Kafka, la partizione, l'offset e l'ora di inserimento.

Passaggi successivi

Apache Kafka® è un marchio registrato di Apache Software Foundation o delle sue società affiliate negli Stati Uniti e/o in altri paesi.

Crea un connettore BigQuery Sink Mantieni tutto organizzato con le raccolte Salva e classifica i contenuti in base alle tue preferenze.