Google verwendet KI-Technologie, um Inhalte in Ihre bevorzugte Sprache zu übersetzen. KI-Übersetzungen können Fehler enthalten.

BigQuery-Sink-Connector erstellen

BigQuery-Senken-Connectors streamen Daten von Kafka in BigQuery-Tabellen und ermöglichen so die Echtzeit-Datenaufnahme und -Analyse in BigQuery.

Anwendungsfälle für BigQuery-Senken-Connectors:

Data-Warehousing Streamingdaten für Analysen und Berichte in BigQuery laden.
BigQuery-Tabellen mit Daten füllen, die für Echtzeit-Dashboards verwendet werden.

Hinweis

Bevor Sie einen BigQuery-Senken-Connector erstellen, müssen Sie Folgendes haben:

Erstellen Sie einen Managed Service for Apache Kafka-Cluster für Ihren Connect-Cluster. Dieser Cluster ist der primäre Kafka-Cluster, der mit dem Connect-Cluster verknüpft ist. Dieser Cluster ist auch der Quellcluster, der ein Ende der BigQuery-Senken-Connector-Pipeline bildet.
Connect-Cluster erstellen, um Ihren BigQuery-Senken-Connector zu hosten.
Erstellen Sie ein BigQuery-Dataset zum Speichern der aus Kafka gestreamten Daten.
Erstellen und konfigurieren Sie ein Kafka-Thema im Quellcluster. Daten werden von diesem Kafka-Thema in das BigQuery-Ziel-Dataset übertragen.

Erforderliche Rollen und Berechtigungen

Bitten Sie Ihren Administrator, Ihnen die IAM-Rolle Managed Kafka Connector Editor (roles/managedkafka.connectorEditor) für Ihr Projekt zuzuweisen, um die Berechtigungen zu erhalten, die Sie zum Erstellen eines Connectors benötigen. Weitere Informationen zum Zuweisen von Rollen finden Sie unter Zugriff auf Projekte, Ordner und Organisationen verwalten.

Diese vordefinierte Rolle enthält die Berechtigungen, die zum Erstellen eines Connectors erforderlich sind. Maximieren Sie den Abschnitt Erforderliche Berechtigungen, um die notwendigen Berechtigungen anzuzeigen:

Erforderliche Berechtigungen

Die folgenden Berechtigungen sind zum Erstellen eines Connectors erforderlich:

Connector erstellen: managedkafka.connectors.create

Sie können diese Berechtigungen auch mit benutzerdefinierten Rollen oder anderen vordefinierten Rollen erhalten.

Berechtigungen zum Schreiben in die BigQuery-Tabelle gewähren

Das Managed Kafka-Dienstkonto muss die Berechtigung haben, Nachrichten in die BigQuery-Tabelle zu schreiben. Weisen Sie dem Dienstkonto die Rolle BigQuery-Dateneditor (roles/bigquery.dataEditor) für das Projekt zu, das die Tabelle enthält.

Das Managed Kafka-Dienstkonto hat das folgende Format: service-PROJECT_NUMBER@gcp-sa-managedkafka.iam.gserviceaccount.com, wobei PROJECT_NUMBER die Projektnummer des Connect-Clusters ist.

Wenn sich Ihr Connect-Cluster in einem anderen Projekt als Ihr Managed Service for Apache Kafka-Cluster befindet, lesen Sie den Abschnitt Connect-Cluster in einem anderen Projekt erstellen.

Wichtige Konfigurationsparameter

Im Folgenden finden Sie Empfehlungen für Konfigurationen für den BigQuery-Senken-Connector.

Warteschlangengröße

Mit der Einstellung queueSize wird die maximale Anzahl von Schreibanfragen an BigQuery gesteuert. Standardmäßig ist dieser Wert nicht konfiguriert, was zu Fehlern aufgrund von unzureichendem Arbeitsspeicher im Connect-Cluster führen kann. Wir empfehlen, diesen Wert zu konfigurieren, um die Wahrscheinlichkeit solcher Fehler zu verringern. Die Elemente in der Warteschlange sind Schreibanfragen und entsprechen nicht direkt einer Anzahl von Nachrichten oder Bytes. Daher ist es schwierig, den empfohlenen Wert im Voraus zu ermitteln.

Wenn Nachrichten aus Kafka abgerufen werden, teilt der Connector die Arbeit basierend auf der Zieltabelle auf. Wenn alle Nachrichten in einer Umfrage in eine einzelne Tabelle geschrieben werden sollen, wird nur ein Arbeitselement hinzugefügt. Wenn die Nachrichten in einer Umfrage jedoch in drei Tabellen geschrieben werden, werden drei Arbeitselemente hinzugefügt. Die Größe eines Arbeitselements kann anhand der Konfigurationen für das Abrufen von Nachrichten, der Größe der Kafka-Nachrichten und der Anzahl der Tabellen, in die geschrieben wird, geschätzt werden.

Schemas für einen BigQuery-Senken-Connector

Der BigQuery-Senken-Connector verwendet den konfigurierten Wertkonverter (value.converter), um Kafka-Datensatzwerte in Felder zu parsen. Anschließend werden die Felder in Spalten mit demselben Namen in der BigQuery-Tabelle geschrieben.

Für den Connector ist ein Schema erforderlich. Das Schema kann auf folgende Weise bereitgestellt werden:

Nachrichtenbasiertes Schema: Das Schema ist Teil jeder Nachricht.
Tabellenbasiertes Schema: Der Connector leitet das Nachrichtenschema aus dem BigQuery-Tabellenschema ab.
Schema-Registry: Der Connector liest das Schema aus einer Schema-Registry, z. B. der Schema-Registry von Managed Service for Apache Kafka (Vorabversion).

In den nächsten Abschnitten werden diese Optionen beschrieben.

Nachrichtenbasiertes Schema

In diesem Modus enthält jeder Kafka-Eintrag ein JSON-Schema. Der Connector verwendet das Schema, um die Datensatzdaten als BigQuery-Tabellenzeile zu schreiben.

Wenn Sie nachrichtenbasierte Schemas verwenden möchten, legen Sie die folgenden Eigenschaften für den Connector fest:

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

Beispiel für einen Kafka-Eintragswert:

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

Wenn die Zieltabelle bereits vorhanden ist, muss das BigQuery-Tabellenschema mit dem Schema der eingebetteten Nachricht kompatibel sein. Wenn autoCreateTables=true, erstellt der Connector bei Bedarf automatisch die Zieltabelle. Weitere Informationen finden Sie unter Tabellen erstellen.

Wenn der Connector das BigQuery-Tabellenschema aktualisieren soll, wenn sich die Nachrichtenschemas ändern, legen Sie allowNewBigQueryFields, allowSchemaUnionization oder allowBigQueryRequiredFieldRelaxation auf true fest.

Tabellenbasiertes Schema

In diesem Modus enthalten die Kafka-Datensätze einfache JSON-Daten ohne explizites Schema. Der Connector leitet das Schema aus der Zieltabelle ab.

Voraussetzungen:

Die BigQuery-Tabelle muss bereits vorhanden sein.
Die Kafka-Eintragsdaten müssen mit dem Tabellenschema kompatibel sein.
Dieser Modus unterstützt keine dynamischen Schemaaktualisierungen basierend auf eingehenden Nachrichten.

Wenn Sie tabellenbasierte Schemas verwenden möchten, legen Sie die folgenden Attribute für den Connector fest:

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

Wenn für die BigQuery-Tabelle zeitbasierte Partitionierung mit täglicher Partitionierung verwendet wird, kann bigQueryPartitionDecorator true sein. Andernfalls legen Sie diese Eigenschaft auf false fest.

Beispiel für einen Kafka-Eintragswert:

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

Schema-Registry

In diesem Modus enthält jeder Kafka-Datensatz Apache Avro-Daten und das Nachrichtenschema wird in einer Schema-Registry gespeichert.

Wenn Sie den BigQuery-Senken-Connector mit einer Schemaregistrierung verwenden möchten, legen Sie die folgenden Eigenschaften für den Connector fest:

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

Ersetzen Sie SCHEMA_REGISTRY_URL durch die URL der Schemaregistry.

Wenn Sie den Connector mit der Schema-Registry von Managed Service for Apache Kafka verwenden möchten, legen Sie die folgende Eigenschaft fest:

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

Weitere Informationen finden Sie unter Kafka Connect mit Schemaregister verwenden.

Verwaltete Apache Iceberg-Tabellen

Der BigQuery Sink-Connector unterstützt verwaltete Apache Iceberg-Tabellen (im Folgenden verwaltete Iceberg-Tabellen) als Senkenziel.

Von Google verwaltete Iceberg-Tabellen bilden die Grundlage für die Erstellung von Lakehouses im offenen Format auf Google Cloud. Verwaltete Iceberg-Tabellen bieten dieselbe vollständig verwaltete Umgebung wie BigQuery-Tabellen, speichern Daten jedoch in kundeneigenen Speicher-Buckets mit Parquet, um mit offenen Apache Iceberg-Tabellenformaten kompatibel zu sein.

Informationen zum Erstellen einer Apache Iceberg-Tabelle finden Sie unter Apache Iceberg-Tabelle erstellen.

BigQuery-Senken-Connector erstellen

Console

Rufen Sie in der Google Cloud Console die Seite Connect Clusters auf.

Zu „Connect Clusters“
Klicken Sie auf den Connect-Cluster, in dem Sie den Connector erstellen möchten.
Klicken Sie auf Connector erstellen.
Geben Sie für den Connectornamen einen String ein.

Tipps zum Benennen von Connectors finden Sie unter Richtlinien zum Benennen einer Ressource von Managed Service for Apache Kafka.
Wählen Sie für Connector-Plug-in die Option BigQuery Sink aus.
Geben Sie im Bereich Themen die Kafka-Themen an, aus denen gelesen werden soll. Sie können eine Liste von Themen oder einen regulären Ausdruck angeben, der mit den Namen der Themen abgeglichen werden soll.
- Option 1: Wählen Sie Liste mit Kafka-Themen auswählen aus. Wählen Sie in der Liste Kafka-Themen ein oder mehrere Themen aus. Klicken Sie auf OK.
- Option 2: Wählen Sie Thema-Regex verwenden aus. Geben Sie im Feld Themen-Regex einen regulären Ausdruck ein.
Klicken Sie auf Dataset und geben Sie ein BigQuery-Dataset an. Sie können ein vorhandenes Dataset auswählen oder ein neues erstellen.
Optional: Fügen Sie im Feld Konfigurationen Konfigurationseigenschaften hinzu oder bearbeiten Sie die Standardeigenschaften. Weitere Informationen finden Sie unter Connector konfigurieren.
Wählen Sie die Richtlinie für Task-Neustart aus. Weitere Informationen finden Sie unter Richtlinie zum Neustart von Aufgaben.
Klicken Sie auf Erstellen.

gcloud

Aktivieren Sie Cloud Shell in der Google Cloud Console.

Cloud Shell aktivieren

Unten in der Google Cloud Console wird eine Cloud Shell-Sitzung gestartet und eine Eingabeaufforderung angezeigt. Cloud Shell ist eine Shell-Umgebung, in der das Google Cloud CLI bereits installiert ist und Werte für Ihr aktuelles Projekt bereits festgelegt sind. Das Initialisieren der Sitzung kann einige Sekunden dauern.
Zum Abrufen der aktuellen Richtlinie führen Sie den Befehl gcloud managed-kafka connectors create aus:
```
gcloud managed-kafka connectors create CONNECTOR_ID \
    --location=LOCATION \
    --connect-cluster=CONNECT_CLUSTER_ID \
    --config-file=CONFIG_FILE
```
Ersetzen Sie Folgendes:
- CONNECTOR_ID: Die ID oder der Name des Connectors. Tipps zum Benennen von Connectors finden Sie unter Richtlinien zum Benennen einer Ressource von Managed Service for Apache Kafka. Der Name eines Connectors kann nicht geändert werden.
- LOCATION: Der Ort, an dem Sie den Connector erstellen. Dies muss derselbe Standort sein, an dem Sie den Connect-Cluster erstellt haben.
- CONNECT_CLUSTER_ID: Die ID des Connect-Clusters, in dem der Connector erstellt wird.
- CONFIG_FILE: Der Pfad zur YAML-Konfigurationsdatei für den BigQuery Sink-Connector.
Hier sehen Sie ein Beispiel für eine Konfigurationsdatei für den BigQuery Sink Connector:
```
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"
```
Ersetzen Sie Folgendes:
- BQ_SINK_CONNECTOR_ID: Die ID oder der Name des BigQuery Sink-Connectors. Hinweise zum Benennen eines Connectors finden Sie unter Richtlinien zum Benennen einer Managed Service for Apache Kafka-Ressource. Der Name eines Connectors kann nicht geändert werden.
- GCP_PROJECT_ID: Die ID des Google Cloud-Projekts, in dem sich Ihr BigQuery-Dataset befindet.
- GMK_TOPIC_ID: Die ID des Managed Service for Apache Kafka-Themas, aus dem die Daten in den BigQuery Sink-Connector fließen.
- BQ_DATASET_ID: Die ID des BigQuery-Datasets, das als Senke für die Pipeline dient.

Terraform

Sie können eine Terraform-Ressource verwenden, um einen Connector zu erstellen.

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
}

Informationen zum Anwenden oder Entfernen einer Terraform-Konfiguration finden Sie unter Grundlegende Terraform-Befehle.

Go

Bevor Sie dieses Beispiel ausprobieren, folgen Sie der Go-Einrichtungsanleitung unter Clientbibliotheken installieren. Weitere Informationen finden Sie in der Referenzdokumentation zur Go API für Managed Service for Apache Kafka.

Richten Sie zur Authentifizierung bei Managed Service for Apache Kafka die Standardanmeldedaten für Anwendungen(Application Default Credentials, ADC) ein. Weitere Informationen finden Sie unter ADC für eine lokale Entwicklungsumgebung einrichten.

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

Bevor Sie dieses Beispiel ausprobieren, folgen Sie der Anleitung zur Einrichtung von Java unter Clientbibliotheken installieren. Weitere Informationen finden Sie in der Referenzdokumentation zur Java API für Managed Service for Apache Kafka.

Richten Sie die Standardanmeldedaten für Anwendungen ein, um sich bei Managed Service for Apache Kafka zu authentifizieren. Weitere Informationen finden Sie unter ADC für eine lokale Entwicklungsumgebung einrichten.


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

Folgen Sie der Einrichtungsanleitung für Python unter Clientbibliotheken installieren, bevor Sie dieses Beispiel ausprobieren. Weitere Informationen finden Sie in der Referenzdokumentation zur Python API für Managed Service for 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}")

Nachdem Sie einen Connector erstellt haben, können Sie ihn bearbeiten, löschen, pausieren, beenden oder neu starten.

Connector konfigurieren

In diesem Abschnitt werden einige Konfigurationseigenschaften beschrieben, die Sie für den Connector festlegen können. Eine vollständige Liste der Eigenschaften, die für diesen Connector spezifisch sind, finden Sie unter BigQuery Sink Connector-Konfigurationen.

Tabellenname

Standardmäßig verwendet der Connector den Themennamen als BigQuery-Tabellennamen. Wenn Sie einen anderen Tabellennamen verwenden möchten, legen Sie die Eigenschaft topic2TableMap im folgenden Format fest:

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

Tabellenerstellung

Mit dem BigQuery-Senken-Connector können die Zieltabelle erstellt werden, falls sie nicht vorhanden sind.

Wenn autoCreateTables=true, versucht der Connector, alle BigQuery-Tabellen zu erstellen, die nicht vorhanden sind. Das ist das Standardverhalten.
Wenn autoCreateTables=false, werden vom Connector keine Tabellen erstellt. Wenn eine Zieltabelle nicht vorhanden ist, tritt ein Fehler auf.

Wenn autoCreateTables gleich true ist, können Sie die folgenden Konfigurationseigenschaften verwenden, um genauer zu steuern, wie der Connector neue Tabellen erstellt und konfiguriert:

allBQFieldsNullable
clusteringPartitionFieldNames
convertDoubleSpecialValues
partitionExpirationMs
sanitizeFieldNames
sanitizeTopics
timestampPartitionFieldName

Weitere Informationen zu diesen Attributen finden Sie unter BigQuery Sink Connector-Konfigurationen.

Kafka-Metadaten

Sie können zusätzliche Daten aus Kafka, z. B. Metadaten und Schlüsselinformationen, der BigQuery-Tabelle zuordnen, indem Sie die Felder kafkaDataFieldName und kafkaKeyFieldName entsprechend konfigurieren. Beispiele für Metadaten sind das Kafka-Thema, die Partition, der Offset und die Einfügezeit.

Nächste Schritte

Apache Kafka® ist eine eingetragene Marke der Apache Software Foundation oder ihrer verbundenen Unternehmen in den USA und/oder anderen Ländern.

BigQuery-Sink-Connector erstellen Mit Sammlungen den Überblick behalten Sie können Inhalte basierend auf Ihren Einstellungen speichern und kategorisieren.