Allgemeinen PostgreSQL-Quellconnector erstellen

Mit einem Connector vom Typ „Allgemeine PostgreSQL-Quelle“ können Sie Änderungen auf Zeilenebene aus einer PostgreSQL-Datenbank erfassen und in Themen in einem Managed Service for Apache Kafka-Cluster streamen.

Typische Anwendungsfälle für diesen Connector:

  • Datenbankänderungen auf Zeilenebene in Echtzeit überwachen.
  • Datenbankänderungsereignisse in eine ereignisgesteuerte Architektur einbinden
  • Auf Datenbankereignisse wie das Einfügen oder Löschen von Zeilen reagieren.
  • Datenbankänderungen in andere Systeme kopieren.
  • PostgreSQL-Tabellen replizieren oder wiederherstellen.

Wenn Sie Änderungen aus einer Cloud SQL for PostgreSQL-Datenbank lesen möchten, verwenden Sie den Cloud SQL for PostgreSQL-Quellconnector.

Hinweis

Bevor Sie einen Connector vom Typ „Allgemeine PostgreSQL-Quelle“ erstellen, benötigen Sie Folgendes:

  • Eine PostgreSQL-Datenbank.

  • Ein Connect-Cluster, der mit dem Kafka-Cluster verknüpft ist.

  • Erstellen Sie ein Secret Manager-Secret, in dem das Datenbankpasswort gespeichert wird. Wenn Ihre Konfiguration Datenbank-SSL verwendet, erstellen Sie auch ein Secret für das Datenbank-SSL-Passwort. Konfigurieren Sie Ihren Connect-Cluster mit den Secrets. Weitere Informationen finden Sie unter Secret Manager-Ressourcen.

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 für den Zugriff auf Secret Manager-Secrets gewähren

Das Managed Kafka-Dienstkonto benötigt die Berechtigung, die in Secret Manager gespeicherten Secrets aufzurufen und darauf zuzugreifen. Weisen Sie dem Dienstkonto die folgenden IAM-Rollen zu:

  • Secret Manager-Betrachter (roles/secretmanager.viewer)
  • Zugriffsperson für Secret Manager-Secret (roles/secretmanager.secretAccessor)

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 Hilfeartikel Connect-Cluster in einem anderen Projekt erstellen.

PostgreSQL-Datenbank konfigurieren

Konfigurieren Sie die folgenden Einstellungen, damit der Connector Datenänderungsereignisse aus Ihrer Datenbank lesen kann.

  1. Setzen Sie den wal_level des Servers auf logical.

    ALTER SYSTEM SET wal_level = logical;

    Starten Sie den Server neu, um die Einstellung zu übernehmen.

  2. Erstellen Sie einen Datenbanknutzer, damit sich der Connector bei PostgreSQL authentifizieren kann. Der Datenbanknutzer muss eine Replikationsrolle haben, damit er im Replikationsmodus eine Verbindung zum Server herstellen kann.

    CREATE ROLE ROLE_NAME WITH REPLICATION LOGIN PASSWORD 'ROLE_PASSWORD';

    Ersetzen Sie Folgendes:

    • ROLE_NAME: Der Name des Nutzers, z. B. debezium_user.
    • ROLE_PASSWORD: Das Passwort des Nutzers.

  3. Erstellen Sie eine Publikation für die Tabellen, die Sie erfassen möchten. Der Connector abonniert die Publikation, um Ereignisse zu Datenänderungen zu erhalten.

    CREATE PUBLICATION dbz_publication FOR TABLE "SCHEMA_NAME"."TABLE_NAME";

    Ersetzen Sie Folgendes:

    • SCHEMA_NAME: Das Schema der Tabelle.

    • TABLE_NAME: der Name der Tabelle

    Wir empfehlen, die Schema- und Tabellennamen wie gezeigt in doppelte Anführungszeichen zu setzen, um Syntaxfehler zu vermeiden, wenn die Namen Sonderzeichen oder Großbuchstaben enthalten.

    Alternativ können Sie eine Publikation erstellen, die Änderungen für alle Tabellen in der Datenbank repliziert:

    CREATE PUBLICATION dbz_publication FOR ALL TABLES;

    Je nach Einstellung publication.autocreate.mode des Connectors können Sie die Publikation manuell erstellen oder vom Connector automatisch erstellen lassen. Weitere Informationen finden Sie unter Veröffentlichungsmodus.

  4. Gewähren Sie dem Datenbanknutzer für jede Tabelle die Berechtigungen SELECT für die Tabelle.

    GRANT SELECT ON TABLE "SCHEMA_NAME"."TABLE_NAME" TO ROLE_NAME;

    Alternativ können Sie die Auswahl für alle Tabellen in einem Schema gewähren:

    GRANT SELECT ON ALL TABLES IN SCHEMA "SCHEMA_NAME" TO ROLE_NAME;

  5. Gewähren Sie dem Datenbanknutzer für jede Tabelle USAGE-Berechtigungen für das Tabellenschema. Sie können diesen Schritt überspringen, wenn sich die Tabelle im Standardschema public befindet.

    GRANT USAGE ON SCHEMA "SCHEMA_NAME" TO ROLE_NAME;

Allgemeinen PostgreSQL-Quellconnector erstellen

So erstellen Sie einen Connector für eine allgemeine PostgreSQL-Quelle:

Wenn der Connector initialisiert wird, führt er die folgenden Aktionen aus:

  1. Erstellt einen ersten Snapshot der Datenbank.
  2. Erstellt ein Kafka-Thema für jede Tabelle, die Zeilen enthält.
  3. Für jede Datenbankzeile wird ein Änderungsereignis an das entsprechende Thema gesendet.

Während der Connector ausgeführt wird, sendet er weiterhin Änderungsereignisse an die Themen. Weitere Informationen zum ersten Snapshot finden Sie in der Debezium-Dokumentation unter Snapshots.

Console

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

    Zu „Connect-Cluster“

  2. Klicken Sie auf den Connect-Cluster, in dem Sie den Connector erstellen möchten.

  3. Klicken Sie auf Connector erstellen.

  4. Geben Sie für den Connector-Namen einen String ein.

    Tipps zum Benennen von Connectors finden Sie in den Richtlinien zum Benennen einer Ressource von Managed Service for Apache Kafka.

  5. Wählen Sie als Connector-Plug-in die Option Generic PostgreSQL Source aus.

  6. Geben Sie im Feld Datenbank-Hostname den Hostnamen oder die IP-Adresse des PostgreSQL-Servers ein.

  7. Geben Sie im Feld Datenbankname den Namen der Datenbank ein.

  8. Geben Sie im Feld Datenbanknutzer den Namen der Replikatrolle ein. Der Connector authentifiziert sich mit dieser Rolle beim PostgreSQL-Server.

  9. Geben Sie im Feld Themenpräfix ein Präfix ein, das für die Kafka-Themennamen verwendet werden soll.

  10. Wählen Sie in der Liste Secret das Secret aus, das das Datenbankpasswort enthält.

  11. Optional: Fügen Sie im Feld Konfigurationen Konfigurationsattribute hinzu oder bearbeiten Sie die Standardattribute. Weitere Informationen finden Sie unter Connector konfigurieren.

  12. Optional: Wählen Sie die Richtlinie für Task-Neustart aus. Weitere Informationen finden Sie unter Richtlinie für den Neustart von Aufgaben.

  13. Klicken Sie auf Erstellen.

gcloud

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

  2. 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 in den 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 sehen Sie ein Beispiel für eine Konfigurationsdatei für den Connector „Allgemeine PostgreSQL-Quelle“:

    connector.class: io.debezium.connector.postgresql.PostgresConnector
database.dbname: DATABASE_NAME
database.hostname: HOSTNAME
database.password: CREDENTIALS
database.user: DATABASE_USER
key.converter: org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable: "false"
plugin.name: pgoutput
topic.prefix: TOPIC_PREFIX
value.converter: org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable: "true"

    Ersetzen Sie Folgendes:

    • HOSTNAME: Der Hostname der PostgreSQL-Datenbank, aus der gelesen werden soll.

    • DATABASE_NAME: Der Name der PostgreSQL-Datenbank, aus der gelesen werden soll.

    • DATABASE_USER: Der PostgreSQL-Datenbanknutzer, der für die Authentifizierung bei der Datenbank verwendet werden soll.

    • CREDENTIALS: Ein Pfad zum Secret Manager-Secret, das das Datenbankpasswort enthält. Geben Sie das Secret im folgenden Format an:

      ${directory:/var/secrets:PROJECT_ID-SECRET_NAME-SECRET_VERSION}

    • TOPIC_PREFIX: Ein Präfix, das für die Kafka-Themennamen verwendet werden soll.

Connector konfigurieren

In diesem Abschnitt werden einige Konfigurationsattribute beschrieben, die Sie für den Connector festlegen können. Eine vollständige Liste finden Sie in der Debezium-Dokumentation unter Debezium-Connector für PostgreSQL.

Passwort- und SSL-Passwortkonfigurationen

In den Konfigurationen database.password und database.sslpassword werden nur Secret-Pfade unterstützt. Das Backend erwartet, dass diese Konfigurationen das folgende Format verwenden: ${directory:/var/secrets:PROJECT_ID-SECRET_NAME-SECRET_VERSION}.

IP-Adresstypen

Das Attribut driver.ipTypes gibt den Typ der IP-Adresse an, die der Connector für die Verbindung mit der Datenbank verwendet:

  • PRIVATE: Private IP-Adresse
  • PSC: Private Service Connect
  • PUBLIC: Öffentliche IP-Adresse

Das Attribut driver.ipTypes enthält eine durch Kommas getrennte Liste von IP-Typen in der bevorzugten Reihenfolge, z. B. driver.ipTypes=PRIVATE,PUBLIC.

Veröffentlichungsmodus

Ein Connector „Allgemeine PostgreSQL-Quelle“ streamt Änderungsereignisse aus einer Veröffentlichung in der Datenbank. Sie können die Publikation manuell erstellen oder vom Connector automatisch erstellen lassen.

Mit der Einstellung publication.autocreate.mode wird festgelegt, wie und ob der Connector eine Publikation erstellen soll.

  • filtered. Wenn die Publikation nicht vorhanden ist, erstellt der Connector eine neue Publikation, die nur die erfassten Tabellen enthält. Der Datenbanknutzer muss die Berechtigungen CREATE für die Datenbank haben und der Inhaber der enthaltenen Tabellen sein.

    Wenn die Publikation bereits vorhanden ist, ändert der Connector sie so, dass die erfassten Tabellen enthalten sind. Wenn Sie eine vorhandene Veröffentlichung ändern möchten, muss der Datenbanknutzer der Inhaber der Veröffentlichung und der enthaltenen Tabellen sein.

  • all_tables. Wenn die Publikation nicht vorhanden ist, erstellt der Connector eine neue Publikation mit dem Parameter FOR ALL TABLES. Der Datenbanknutzer muss ein Superuser sein.

    Superuser-Rollen umgehen alle Berechtigungsprüfungen in einer Datenbank. Daher wird nicht empfohlen, dem Datenbanknutzer SUPERUSER zuzuweisen. Erstellen Sie die Publikation stattdessen manuell oder legen Sie publication.autocreate.mode=filtered fest.

  • disabled. Wenn die Publikation nicht vorhanden ist, tritt ein Fehler auf. Der Connector erstellt keine neue Publikation.

Der Standardwert ist all_tables.

Name der Publikation

Standardmäßig versucht der Connector, aus einer Publikation mit dem Namen dbz_publication zu streamen. Wenn Sie eine andere Publikation angeben möchten, fügen Sie der Konfiguration publication.name=PUBLICATION_NAME hinzu, wobei PUBLICATION_NAME der Name der Publikation ist. Beispiel: publication.name=my_publication.

Replikationsslots

PostgreSQL verwendet Replikationsslots, um Änderungen an Datenbanktabellen zu streamen. Standardmäßig erstellt der Connector einen Replikationsslot mit dem Namen debezium. Wenn Sie einen anderen Slotnamen verwenden möchten, legen Sie das Attribut slot.name fest.

Wenn Sie zwei Instanzen des Connectors für dieselbe Datenbank erstellen, müssen Sie für jeden Connector einen eindeutigen Slot-Namen angeben.

Standardmäßig legt der Connector das Attribut slot.drop.on.stop auf false fest, um Datenverlust zu vermeiden. Wenn Sie einen Connector endgültig löschen, müssen Sie den Replikationsslot, den der Connector verwendet hat, manuell löschen. Der Name des Replikationsslots ist standardmäßig debezium, sofern er nicht mit dem Attribut slot.name anders konfiguriert wurde.

Wir empfehlen, Benachrichtigungen einzurichten, um die WAL-Laufwerksnutzung auf Ihrem PostgreSQL-Quelldatenbankserver zu überwachen und nicht verwendete Replikationsslots zu löschen.

Tabellenfilter

Standardmäßig erfasst der Connector Änderungsdaten aus allen Nicht-Systemtabellen in der Datenbank. Wenn Sie filtern möchten, welche Tabellen erfasst werden, geben Sie eine oder mehrere der folgenden Einstellungen an:

  • schema.include.list: Eine Liste der Schemas, die einbezogen werden sollen.
  • schema.exclude.list: Eine Liste der auszuschließenden Schemas. Kann nicht mit schema.include.list verwendet werden.
  • table.include.list: Eine Liste der Tabellen, die einbezogen werden sollen.
  • table.exclude.list: Eine Liste der auszuschließenden Tabellen. Kann nicht mit table.include.list verwendet werden.

Themennamen

Standardmäßig erstellt der Connector Kafka-Themen mit der folgenden Namenskonvention: topic_prefix.schema.table_name, wobei topic.prefix der Wert der topic.prefix-Konfiguration ist.

Weitere Informationen finden Sie in der Debezium-Dokumentation unter Themennamen.

Nächste Schritte

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