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

Cloud Storage-Senkenconnector erstellen

Cloud Storage-Senken-Connectors streamen Daten von Kafka zu Cloud Storage-Buckets. So können Sie große Datenmengen kostengünstig und skalierbar speichern und verarbeiten.

Anwendungsfälle für Cloud Storage-Senken-Connectors:

Data-Lake-Aufnahme Kafka-Daten in einem Data Lake für die langfristige Archivierung und Batchverarbeitung speichern
Daten archivieren, um regulatorische Anforderungen zu erfüllen

Hinweis

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

Erstellen Sie einen Managed Service for Apache Kafka-Cluster für Ihren Connect-Cluster. Dies ist der primäre Kafka-Cluster, der mit dem Connect-Cluster verknüpft ist. Dies ist auch der Quellcluster, der ein Ende der Connector-Pipeline bildet.
Erstellen Sie einen Connect-Cluster um Ihren Cloud Storage-Senken-Connector zu hosten.
Erstellen Sie einen Cloud Storage-Bucket, um die von Kafka gestreamten Daten zu speichern.
Erstellen und konfigurieren Sie ein Kafka-Thema im Quellcluster. Daten werden von diesem Kafka-Thema zum Ziel-Cloud Storage-Bucket verschoben.

Erforderliche Rollen und Berechtigungen

Bitten Sie Ihren Administrator, Ihnen die Managed Kafka Connector Editor (roles/managedkafka.connectorEditor) IAM-Rolle für Ihr Projekt zu gewähren, 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, die erforderlich sind:

Erforderliche Berechtigungen

Die folgenden Berechtigungen sind zum Erstellen eines Connectors erforderlich:

Connector erstellen: managedkafka.connectors.create

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

Berechtigungen zum Schreiben in den Cloud Storage-Bucket gewähren

Das Managed Kafka-Dienstkonto muss die folgenden Berechtigungen haben, um Nachrichten in den Cloud Storage-Bucket zu schreiben:

storage.objects.create
storage.objects.delete

Gewähren Sie dem Dienstkonto die Rolle Storage Object User (roles/storage.objectUser) für das Projekt, das den Cloud Storage-Bucket 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 Connect-Cluster in einem anderen Projekt erstellen.

Cloud Storage-Senken-Connector erstellen

So erstellen Sie einen Cloud Storage-Senken-Connector:

Console

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

Zu den Connect-Clustern
Klicken Sie auf den Connect-Cluster, für den Sie den Connector erstellen möchten.

Die Seite Details zum Connect-Cluster wird angezeigt.
Klicken Sie auf Connector erstellen.

Die Seite Kafka-Connector erstellen wird angezeigt.
Geben Sie einen String für den Namen des Connectors ein.

Tipps zum Benennen von Connectors finden Sie unter Richtlinien zum Benennen einer Ressource von Managed Service for Apache Kafka.
Wählen Sie unter Connector-Plug-in die Option Cloud Storage-Senke aus.
Geben Sie die Themen an, aus denen Sie Daten streamen können.
Wählen Sie den Storage-Bucket aus, in dem die Daten gespeichert werden sollen.
Optional: Konfigurieren Sie weitere Einstellungen im Abschnitt Konfiguration.
Wählen Sie die Richtlinie für Task-Neustart aus. Weitere Informationen finden Sie unter Richtlinie für Task-Neustart.
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.
Führen Sie den gcloud managed-kafka connectors create Befehl 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 Standort, 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 Connector.
Hier ist ein Beispiel für eine Konfigurationsdatei für den Cloud Storage-Senken-Connector:
```
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"
```
Ersetzen Sie Folgendes:
- BUCKET_NAME: Der Name des Cloud Storage-Bucket, in den geschrieben werden soll.
- TOPIC_LIST: Eine durch Kommas getrennte Liste von Kafka-Themen, aus denen gelesen werden soll. Die Liste kann auch reguläre Ausdrücke enthalten.

Terraform

Sie können eine Terraform-Ressource zum Erstellen eines Connectors verwenden.

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
}

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

Go

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

Richten Sie zur Authentifizierung bei Managed Service for Apache Kafka 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"
)

// 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

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

Richten Sie zur Authentifizierung bei Managed Service for Apache Kafka Standardanmeldedaten für Anwendungen ein. 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 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

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

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

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

Connector konfigurieren

Eine vollständige Liste der Attribute, die für diesen Connector spezifisch sind, finden Sie unter Konfiguration in der Dokumentation zum Cloud Storage-Senken-Connector.

Ausgabefelder

Standardmäßig schreibt der Connector nur den Datensatzwert aus jedem Kafka-Datensatz. Wenn Sie Datensatzschlüssel oder -header einbeziehen möchten, legen Sie das Attribut format.output.fields auf eine durch Kommas getrennte Liste der einzubeziehenden Felder fest. Die unterstützten Felder sind key, value, headers und timestamp. Beispiel:

format.output.fields=key,value,headers

Ausgabeformate

Der Connector unterstützt das Schreiben von Kafka-Datensätzen im CSV-, JSON- und Parquet-Format. In den folgenden Abschnitten wird beschrieben, wie Sie das Ausgabeformat konfigurieren.

CSV-Format

Wenn Sie CSV-Dateien in Cloud Storage schreiben möchten, legen Sie format.output.type=csv fest. Verwenden Sie die folgenden Converter:

Schlüssel-Converter: org.apache.kafka.connect.storage.StringConverter oder org.apache.kafka.connect.json.JsonConverter
Wert-Converter: org.apache.kafka.connect.converters.ByteArrayConverter

JSON-Format

Wenn Sie Dateien im JSON-Format in Cloud Storage schreiben möchten, legen Sie format.output.type auf einen der folgenden Werte fest:

json: Jede Ausgabedatei enthält ein einzelnes Array von JSON-Objekten, wobei jedes JSON-Objekt einen Kafka-Datensatz darstellt.
jsonl: Der Connector schreibt jeden Kafka-Datensatz als JSON-Objekt in eine separate Zeile.

Das folgende Beispiel zeigt, wie der Connector einen Kafka-Datensatz im JSON-Format schreibt:

{"headers":[{"key":"k1","value":"v1"},{"key":"k2","value":"v2"}],"value":"Hello world","key":"my-key"}

Wenn Sie den Schlüssel- oder Wert-Converter auf org.apache.kafka.connect.converters.ByteArrayConverter festlegen, schreibt der Connector dieses Feld als base64-codierten Wert.

Der Connector enthält kein JSON-Schema in der Ausgabe, auch wenn value.converter.schemas.enable auf true gesetzt ist.

Parquet-Format

Wenn Sie Parquet-Dateien in Cloud Storage schreiben möchten, legen Sie format.output.type=parquet fest.

Wenn der Schlüssel- oder Wert-Converter org.apache.kafka.connect.json.JsonConverter ist, müssen die JSON-Daten ein Schema enthalten. Legen Sie das Attribut schemas.enable des Converters auf true fest.

Bei JSON-Daten schreibt der Connector die Daten als struct-Typ in die Parquet-Datei. Wenn der Kafka-Datensatzwert beispielsweise ein Feld mit dem Namen userid hat, schreibt der Connector eine Spalte mit dem Namen value.userid. Wenn Sie die Spaltennamen reduzieren möchten, legen Sie format.output.envelope=false fest. Wenn format.output.envelope auf false gesetzt ist, muss format.output.fields ein einzelnes Ausgabefeld angeben.

Cloud Storage-Dateieinstellungen

Mit den folgenden Einstellungen wird gesteuert, wie der Connector Dateien in Cloud Storage schreibt.

gcs.bucket.name: Erforderlich. Gibt den Namen des Cloud Storage-Buckets an, in den der Connector Daten schreibt.
file.compression.type: Gibt den Komprimierungstyp für die Ausgabedateien an. Beispiele sind gzip, snappy, zstd und none. Der Standardwert ist none.
file.name.prefix: Gibt ein Präfix an, das dem Namen jeder Datei hinzugefügt werden soll. Der Standardwert ist leer ("").
file.name.template: Gibt eine Vorlage für den Dateinamen an. Der Connector unterstützt Platzhaltervariablen für die Vorlage, z. B. {{topic}} für den Namen des Themas und {{partition}} für die Partitionsnummer. Weitere Informationen finden Sie in der Dokumentation zum Cloud Storage-Senken-Connector.
file.max.records: Die maximale Anzahl von Datensätzen, die in eine einzelne Datei geschrieben werden sollen. Wenn der Wert 0 ist, gibt es kein bestimmtes Maximum. Der Standardwert ist 0.
file.max.bytes: Die maximale Dateigröße in Byte. Wenn der Wert 0 ist, gibt es kein bestimmtes Maximum. Der Standardwert ist 0.

Datensätze gruppieren

Wenn der Cloud Storage-Senken-Connector Kafka-Datensätze schreibt, werden die Datensätze standardmäßig nach Thema und Partition gruppiert. Datensätze aus Partition 0 des Themas my-topic werden beispielsweise in Dateien mit dem Namen my-topic-0-OFFSET geschrieben, wobei OFFSET der Start-Offset ist.

Wenn der file.name.template Parameter die {{key}} Platzhaltervariable enthält, gruppiert der Connector Datensätze nach Schlüssel. In diesem Modus schreibt der Connector für jeden Datensatzschlüssel eine separate Datei. Der Connector überschreibt jede Datei mit dem neuesten Datensatz für diesen Schlüssel.

Anmeldedaten

Das Attribut gcs.credentials.default muss auf true gesetzt sein. Mit dieser Einstellung kann der Connector Anmeldedaten automatisch erkennen Google Cloud .

Nächste Schritte

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

Cloud Storage-Senkenconnector erstellen Mit Sammlungen den Überblick behalten Sie können Inhalte basierend auf Ihren Einstellungen speichern und kategorisieren.