Auf Kafka-Daten in Cloud Storage zugreifen

Wenn Sie Daten aus einem Google Cloud Managed Service for Apache Kafka-Thema in einen Cloud Storage-Bucket laden müssen, können Sie dazu eine Dataflow-Vorlage verwenden. Sie können die Google Cloud Console, die REST API oder die Google Cloud CLI verwenden.

In diesem Dokument wird beschrieben, wie Sie die Kafka to Cloud Storage Dataflow Vorlage mit der Google Cloud Console konfigurieren.

Google Cloud Verwendete Produkte

Die Vorlage Kafka to Cloud Storage Dataflow verwendet die folgenden kostenpflichtigen Google Cloud Produkte. Sie können mithilfe des Preisrechners eine Kostenschätzung für Ihre voraussichtliche Nutzung erstellen.

  • Dataflow: Dataflow ist ein vollständig verwalteter Dienst zur Datenverarbeitung. Die Vorlage Kafka to Cloud Storage Dataflow verwendet Dataflow, um eine Pipeline zu erstellen, die Daten aus Ihrem Kafka-Thema liest, alle erforderlichen Transformationen ausführt und sie in Cloud Storage schreibt. Die Autoscaling- und Selbstheilungsfunktionen von Dataflow sorgen dafür, dass Ihre Pipeline zuverlässig und effizient ausgeführt wird.
  • Cloud Storage: Dient als Ziel für Ihre Kafka-Daten. Sie benötigen einen Cloud Storage-Bucket, um die von der Dataflow-Pipeline übertragenen Daten zu speichern.

Außerdem wird in der Lösung auch Google Cloud Managed Service for Apache Kafka verwendet.

  • Google Cloud Managed Service for Apache Kafka: Ein Google Cloud Dienst , mit dem Sie Apache Kafka ausführen können. Stellt die Quelldaten für die Pipeline bereit. Sie benötigen einen vorhandenen Managed Service for Apache Kafka-Cluster und ein Thema mit Daten, die Sie zu Cloud Storage übertragen möchten. Weitere Informationen zu den Preisen für Google Cloud Managed Service for Apache Kafka finden Sie unter dem Preisleitfaden.

Hinweis

Bevor Sie die Vorlage Kafka to Cloud Storage Dataflow starten, müssen Sie Folgendes erledigt haben:

  1. Erstellen Sie einen Managed Service for Apache Kafka-Cluster und ein Thema.

    Eine Möglichkeit, einen Cluster und ein Thema zu erstellen, ist die Schnellstartanleitung für Managed Service for Apache Kafka zu befolgen.

    Wenn Ihr Thema Avro-Datensätze enthält, finden Sie weitere Informationen zu den Ressourcen anforderungen unter Nachrichtenformat angeben.

  2. Aktivieren Sie die folgenden Google Cloud APIs:

    • Dataflow

    • Cloud Storage

    gcloud services enable dataflow.googleapis.com storage-api.googleapis.com \

  3. Erstellen Sie einen Cloud Storage-Bucket.

    Weitere Informationen zum Erstellen eines Cloud Storage-Bucket finden Sie unter Bucket erstellen.

Dem Dataflow-Worker-Dienstkonto die Rolle „Managed Kafka-Client“ gewähren

Wenn Sie Ihren Dataflow-Job mit Managed Service for Apache Kafka verbinden möchten, müssen Sie dem Dataflow-Worker-Dienstkonto bestimmte Berechtigungen gewähren. Dieses Dienstkonto ist die Identität, die für alle Worker-VMs in Ihrem Dataflow-Job verwendet wird. Alle Anfragen, die von diesen VMs stammen, verwenden dieses Konto.

Damit Sie auf Ihre Kafka-Ressourcen zugreifen können, müssen Sie dem Dataflow-Worker-Dienstkonto die Rolle roles/managedkafka.client gewähren. Diese Rolle enthält die erforderliche Berechtigung managedkafka.clusters.connect zum Herstellen von Verbindungen.

Weitere Informationen zum Worker-Dienstkonto finden Sie unter Sicherheit und Berechtigungen für Pipelines in Google Cloud.

So gewähren Sie dem Dataflow-Dienstkonto die Rolle „Managed Kafka-Client“:

Console

  1. Rufen Sie in der Google Cloud Console die Seite IAM auf.
    IAM aufrufen
  2. Prüfen Sie, ob das Projekt auf das Verbraucherprojekt festgelegt ist, auf das der Managed Service for Apache Kafka-Client zugreifen soll.
  3. Klicken Sie auf Zugriff erlauben.
  4. Geben Sie auf der neuen Seite unter Hauptkonten hinzufügen die E‑Mail-Adresse des Dataflow-Worker-Dienstkontos ein, das Sie verwenden.
  5. Wählen Sie unter Rollen zuweisen die Rolle Managed Kafka Client aus.
  6. Klicken Sie auf Speichern.

gcloud CLI

  1. 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 Befehlszeilenaufforderung 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.

  2. Führen Sie den gcloud projects add-iam-policy-binding Befehl aus:

    gcloud projects add-iam-policy-binding PROJECT_ID \
  --member serviceAccount:SERVICE_ACCOUNT_EMAIL \
  --role roles/managedkafka.client

    Ersetzen Sie Folgendes:

    • PROJECT_ID ist die Projekt-ID.

    • SERVICE_ACCOUNT_EMAIL ist die E‑Mail-Adresse des Dataflow-Worker-Dienstkontos.

Vorlage „Kafka to Cloud Storage Dataflow“ starten

Sie können die Vorlage Kafka to Cloud Storage Dataflow auf der Seite mit den Clusterdetails in der Console starten.

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

    Zu den Clustern

    Die in einem Projekt erstellten Cluster werden aufgelistet.

  2. Klicken Sie auf einen Clusternamen, um die Seite mit den Clusterdetails aufzurufen.
  3. Klicken Sie auf der Seite mit den Clusterdetails auf Daten importieren.

    Die Seite Dataflow-Job mit der Vorlage „Kafka to Kafka“ erstellen wird geöffnet.

  4. Aktualisieren Sie in der Vorlage unter Dataflow-Vorlage die Vorlage auf Kafka to Cloud Storage.

Konfigurieren Sie die Felder in der Vorlage gemäß den Informationen in den folgenden Abschnitten.

Jobname eingeben

Geben Sie im Feld Jobname einen Namen für Ihren Dataflow-Job ein.

Der Name muss unter allen aktuell ausgeführten Jobs im Projekt eindeutig sein.

Regionalen Endpunkt für die Pipeline auswählen

Legen Sie im Feld Regionaler Endpunkt den regionalen Endpunkt auf den Standort Ihres Kafka-Clusters fest, um die Gebühren für die Datenübertragung zwischen Regionen zu minimieren.

Die Dataflow-Worker können unabhängig von der Region Ihres Kafka-Clusters ausgeführt werden. Wenn Sie Worker außerhalb der Region Ihres Kafka-Clusters starten, fallen jedoch Kosten für ausgehenden Traffic zwischen Regionen an.

Die Schritte zum Aufrufen des Standorts des Clusters finden Sie unter Managed Service for Apache Kafka-Cluster auflisten.

Quelle konfigurieren

  1. Behalten Sie für Quelle den Standardwert Managed Service for Apache Kafka bei.

  2. Behalten Sie für Kafka-Cluster und Modus der Kafka-Quellauthentifizierung die Standardwerte bei.

  3. Wählen Sie unter Kafka-Thema ein Thema aus der Liste der verfügbaren Themen aus.

Kafka-Nachrichtenformat konfigurieren

Die Dataflow-Vorlage unterstützt die folgenden drei Nachrichtenformate:

  • Avro Confluent-Wire-Format: Jede Kafka-Nachricht enthält ein Magic Byte, eine Schema-ID und den binärcodierten Avro-Datensatz.

    Für Avro-Formate (Confluent-Wire-Format) können Sie entweder ein einzelnes Schema oder mehrere Schemas verwenden:

    • Einzelnes Schema: Alle Nachrichten entsprechen einem einzelnen vordefinierten Avro-Schema.

    • Mehrere Schemas: Nachrichten können verschiedene Schemas verwenden. Dies wird nur für das Avro-Format (Confluent-Wire-Format) unterstützt.

  • Avro (binärcodiert) : Nachrichten enthalten nur die Nutzlast des Datensatzes ohne Metadaten. Sie müssen eine Avro-Schemadatei (.avsc) bereitstellen, die in Cloud Storage hochgeladen wurde. Alle Nachrichten müssen diesem einzelnen Schema entsprechen.

  • JSON: Für Datensätze ist kein vordefiniertes Schema erforderlich. Datensätze, die nicht dem Schema entsprechen, werden an die Dead-Letter-Warteschlange gesendet (falls konfiguriert) oder es wird eine Fehlermeldung protokolliert. Das unterstützte Format ist {"field": "value"} Format. Das Format [{"name": "field", "value": "value"}] wird nicht unterstützt.

Google Cloud Managed Service for Apache Kafka bietet keine Schema-Registry. Die Vorlage unterstützt nur die Übergabe von Anmeldedaten für die Authentifizierung an Schema-Registries, die mit dem Confluent-Wire-Format kompatibel sind.

Avro Confluent-Wire-Format

Wenn Sie diese Option als das Kafka-Nachrichtenformat auswählen, konfigurieren Sie die folgenden zusätzlichen Einstellungen:

Schemaquelle: In diesem Feld wird der Pipeline mitgeteilt, wo das Schema zu finden ist. Wählen Sie eine der folgenden Optionen aus:

  • Schema-Registry: Ihre Schemas werden in einer Confluent-Schema-Registry gespeichert. Dies ist nützlich für die Entwicklung von Schemas und die Verwaltung mehrerer Versionen. Achten Sie darauf, dass die Schema-Registry für das Netzwerk des Managed Service for Apache Kafka-Clusters zugänglich ist und in derselben Region wie Ihre Dataflow-Worker gehostet wird. Sie können eine Schema-Registry sowohl für einzelne als auch für mehrere Schemas verwenden. Konfigurieren Sie die folgenden zusätzlichen Einstellungen:

    • URL der Schema-Registry-Verbindung: Geben Sie die URL ein, um eine Verbindung zu Ihrer Schema-Registry herzustellen.

    • Authentifizierungsmodus: Wenn für Ihre Registry eine Authentifizierung erforderlich ist, wählen Sie OAuth oder TLS aus. Wählen Sie andernfalls Keine aus.

  • Einzelne Schemadatei: Wählen Sie diese Option aus, wenn alle Ihre Nachrichten einem einzelnen, festen Schema folgen, das in einer Datei definiert ist.

    • Cloud Storage-Datei zur Avro-Schemadatei: Der Pfad zur Avro Schemadatei, die zum Decodieren aller Nachrichten in einem Thema verwendet wird.

Binärcodierung von Avro

Wenn Sie diese Option als das Kafka-Nachrichtenformat auswählen, konfigurieren Sie die folgenden zusätzlichen Einstellungen:

  • Cloud Storage-Datei zur Avro-Schemadatei: Der Pfad zur Avro Schemadatei, die zum Decodieren aller Nachrichten in einem Thema verwendet wird.

JSON

Wenn Sie diese Option als das Kafka-Nachrichtenformat auswählen, sind keine weiteren Konfigurationen erforderlich.

Kafka-Offset angeben

  1. Wenn Sie die Verarbeitung von Nachrichten vermeiden möchten, wenn einzelne Worker oder die gesamte Pipeline neu gestartet werden müssen, wählen Sie die Option Offsets an Kafka übertragen aus. So wird sichergestellt, dass die Pipeline die Verarbeitung dort fortsetzt, wo sie unterbrochen wurde. Dadurch werden doppelte Verarbeitungen und potenzielle Dateninkonsistenzen vermieden.

  2. Geben Sie im Feld Consumer-Gruppen-ID eingeben einen eindeutigen Namen für die Gruppe dieser Pipeline ein. In den meisten Fällen soll die Pipeline jede Nachricht einmal lesen und neu gestartet werden können.

  3. Für das Feld Standard-Kafka-Start-Offset bietet die Dataflow-Pipeline zwei Optionen für das Start-Offset. Wählen Sie eine der folgenden Optionen aus:

    • Frühester: Verarbeitet Nachrichten vom Anfang des Kafka-Themas.

    • Neueste: Verarbeitet Nachrichten ab dem neuesten verfügbaren Offset.

Ziel konfigurieren

Mit diesen Optionen wird gesteuert, wie die Datenpipeline Daten in Cloud Storage schreibt.

  1. Geben Sie unter Ziel den Bucket-Pfad ein und fügen Sie das Dateinamenpräfix für Ihre Ausgabedateien ein. Das Dateipräfix muss mit einem Schrägstrich enden. Beispiel: gs://test-bucket/test-prefix/

  2. Geben Sie unter Fensterdauer das Zeitfenster für das Schreiben von Daten in Cloud Storage ein. Wählen Sie je nach Ihren Anforderungen an die Datenverarbeitung das entsprechende Format aus (Ns für Sekunden, Nm für Minuten, Nh für Stunden).

  3. Geben Sie unter Ausgabedateipräfix der zu schreibenden Dateien ein Präfix an, das jeder Ausgabedatei hinzugefügt werden soll, um die Organisation und Identifizierung zu verbessern.

  4. Setzen Sie die Zahl für Maximale Anzahl von Ausgabeshards auf null. Sie können die Anzahl der Shards angeben, die beim Schreiben von Dateien erstellt werden sollen. Wenn Sie die Anzahl erhöhen, kann ein höherer Durchsatz erzielt werden, aber es entstehen auch höhere Kosten aufgrund höherer Shuffle-Kosten. Wenn Sie die Zahl auf null setzen, wählt der Dienst eine optimale Anzahl aus.

Dead-Letter-Warteschlange konfigurieren

Manchmal können Nachrichten aufgrund von Beschädigungen, inkompatiblen Datentypen oder Schema-Abweichungen nicht verarbeitet werden.

Um diese Fälle zu behandeln, aktivieren Sie die Dead-Letter-Warteschlange in der Vorlage und geben Sie einen Tabellennamen an. Die Tabelle wird mit einem standardisierten Schema erstellt.

Verschlüsselung konfigurieren

Standardmäßig werden alle Daten im Ruhezustand und bei der Übertragung mit einem Google-owned and Google-managed encryption keyverschlüsselt. Wenn Sie vom Kunden verwaltete Verschlüsselungsschlüssel (Customer-Managed Encryption Keys, CMEK) haben, können Sie Ihre eigenen Schlüssel auswählen. Weitere Informationen zum Konfigurieren eines CMEK finden Sie unter Nachrichtenverschlüsselung konfigurieren.

Netzwerk konfigurieren

Sie müssen das Netzwerk und das Subnetzwerk des Clusters in der Dataflow-Vorlage angeben. Im Abschnitt Optionale Parameter der Vorlage können Sie das Netzwerk für Ihre Dataflow-Worker definieren.

Standardmäßig werden in der Vorlage Kafka to Cloud Storage Dataflow Dataflow-Worker im Standardnetzwerk Ihres Projekts bereitgestellt. Damit Ihr Managed Service for Apache Kafka-Cluster Daten über Dataflow an Cloud Storage senden kann, müssen Ihre Dataflow-Worker auf das Netzwerk Ihres Clusters zugreifen können.

Wenn Ihr Kafka-Cluster nicht mit einem Subnetzwerk im Standardnetzwerk des Projekts verbunden ist, empfehlen wir, das Standardnetzwerk des Projekts für Ihren Kafka-Cluster zu verwenden.

Weitere Informationen zum Einrichten des Netzwerks mit Ihrer Dataflow-Pipeline finden Sie unter:

Wenn Sie Probleme beim Konfigurieren Ihres Dataflow-Netzwerks haben, lesen Sie den Leitfaden zur Fehlerbehebung bei Dataflow-Netzwerken.

Optionale Dataflow-Parameter konfigurieren

Konfigurieren Sie die optionalen Parameter nur, wenn Sie die Auswirkungen der Konfiguration auf die Dataflow-Worker kennen. Falsche Einstellungen können sich auf die Leistung oder die Kosten auswirken. Ausführliche Erklärungen zu den einzelnen Optionen finden Sie unter Optionale Parameter.

Monitoring

Die Dataflow-Vorlage für Kafka to Cloud Storage bietet eine Monitoring-Funktion, mit der Sie Logs, Messwerte und Fehler in der Console untersuchen können. Diese Monitoring-Suite ist Teil der Dataflow-Benutzeroberfläche.

Auf dem Tab Jobmesswerte können Sie benutzerdefinierte Dashboards erstellen. Für die Dataflow-Vorlage Kafka to Cloud Storage empfehlen wir, ein Jobmesswerte -Dashboard einzurichten, das Folgendes überwacht:

  • Durchsatz: Das Datenvolumen, das zu einem bestimmten Zeitpunkt verarbeitet wird. Dies ist nützlich, um den Datenfluss durch Ihren Job zu überwachen und potenzielle Leistungsprobleme zu erkennen.

    Weitere Informationen finden Sie unter Dataflow-Durchsatzmonitoring.

  • Datenaktualität: Die Differenz in Sekunden zwischen dem Zeitstempel des Datenelements und dem Zeitpunkt, zu dem das Ereignis in Ihrer Pipeline verarbeitet wird. So können Sie Engpässe bei der Leistung und der Datenquelle oder häufige Wiederholungen erkennen.

    Weitere Informationen finden Sie unter Dataflow-Datenaktualitätsmonitoring.

  • Backlog: Die Anzahl der Byte, die auf die Verarbeitung warten. Diese Informationen werden für Autoscaling-Entscheidungen verwendet.

Weitere Informationen zum Dataflow-Monitoring finden Sie unter der Dataflow-Monitoring-Dokumentation.

Fehlerbehebung

Wenn Leistungsprobleme mit Ihrer Dataflow-Pipeline auftreten, bietet Dataflow eine umfassende Reihe von Tools zur Fehlerbehebung und Diagnose.

Hier sind zwei häufige Szenarien und die entsprechenden Leitfäden zur Fehlerbehebung:

Eine allgemeine Übersicht zur Fehlerbehebung bei Dataflow-Pipelines finden Sie unter Fehlerbehebung bei Dataflow-Pipelines.

Bekannte Einschränkungen

  • Die Vorlage unterstützt nicht die Übergabe von Anmeldedaten für die Authentifizierung an Ihre Schema-Registry.

  • Wenn Sie den Kafka to Cloud Storage Dataflow-Job erstellen, muss das Google Cloud Projekt auf dasselbe Projekt festgelegt sein, das den Managed Service for Apache Kafka-Cluster enthält.

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

Nächste Schritte