Allgemeinen PostgreSQL-Quellconnector erstellen

In diesem Dokument wird beschrieben, wie Sie einen Connector „Allgemeine PostgreSQL-Quelle“ erstellen.

Ein Connector „Allgemeine PostgreSQL-Quelle“ ist eine Instanz eines Debezium PostgreSQL-Connectors. Er liest Änderungen auf Zeilenebene aus einer PostgreSQL-Datenbank und schreibt sie in Themen in einem Managed Service for Apache Kafka-Cluster.

Anwendungsfälle für diesen Connector sind unter anderem:

  • Änderungen auf Zeilenebene in der Datenbank in Echtzeit beobachten.
  • 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.

Hinweis

Bevor Sie einen Connector „Allgemeine PostgreSQL-Quelle“ erstellen, müssen die folgenden Voraussetzungen erfüllt sein:

  • 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 ist. 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 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 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 anzusehen 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 Artikel Connect-Cluster in einem anderen Projekt erstellen.

PostgreSQL-Datenbank konfigurieren

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

  1. Setzen Sie wal_level des Servers auf logical.

    ALTER SYSTEM SET wal_level = logical;

    Starten Sie den Server neu, um die Einstellung anzuwenden.

  2. Erstellen Sie einen Datenbanknutzer, damit sich der Connector bei PostgreSQL authentifizieren kann. Der Datenbanknutzer muss eine Replikationsrolle sein, 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 Datenänderungsereignisse 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 Publikationsmodus.

  4. Gewähren Sie dem Datenbanknutzer für jede Tabelle SELECT-Berechtigungen 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;

Connector „Allgemeine PostgreSQL-Quelle“ erstellen

Führen Sie die folgenden Schritte aus, um einen Connector „Allgemeine PostgreSQL-Quelle“ zu erstellen.

Beim Initialisieren des Connectors werden die folgenden Aktionen ausgeführt:

  1. Erstellt einen ersten Snapshot der Datenbank.
  2. Erstellt ein Kafka-Thema für jede Tabelle mit Zeilen.
  3. Sendet für jede Datenbankzeile ein Änderungsereignis an das entsprechende Thema.

Während der Connector ausgeführt wird, werden weiterhin Änderungsereignisse an die Themen gesendet. 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 den Connect-Clustern

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

  3. Klicken Sie auf Connector erstellen.

  4. Geben Sie eine String für den Namen des Connectors ein.

    Richtlinien zum Benennen eines Connectors finden Sie unter Richtlinien zum Benennen einer Managed Service for Apache Kafka-Ressource.

  5. Wählen Sie unter Connector-Plug-in die Option Allgemeine PostgreSQL-Quelle aus.

  6. Geben Sie im Feld Hostname der Datenbank 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 für die Namen der Kafka-Themen ein.

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

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

  12. Optional: Wählen Sie die Richtlinie für den Neustart von Aufgaben 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 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 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. Richtlinien zum Benennen eines Connectors finden Sie unter Richtlinien zum Benennen einer Managed Service for Apache Kafka-Ressource. Der Name eines Connectors ist unveränderlich.

    • 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 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 für die Namen der Kafka-Themen.

Connector konfigurieren

In diesem Abschnitt werden einige Konfigurationseigenschaften 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.

Konfigurationen für Passwort und SSL-Passwort

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

IP-Adresstypen

Die Eigenschaft driver.ipTypes gibt den Typ der IP-Adresse an, die der Connector verwendet, um eine Verbindung zur Datenbank herzustellen:

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

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

Publikationsmodus

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

Die publication.autocreate.mode Einstellung gibt an, 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 CREATE-Berechtigungen für die Datenbank haben und der Eigentümer der enthaltenen Tabellen sein.

    Wenn die Publikation bereits vorhanden ist, ändert der Connector sie so, dass die erfassten Tabellen enthalten sind. Um eine vorhandene Publikation zu ändern, muss der Datenbanknutzer der Eigentümer der Publikation und der Eigentümer 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 zu gewähren. Erstellen Sie stattdessen die Publikation 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 namens 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 namens debezium. Wenn Sie einen anderen Slotnamen verwenden möchten, legen Sie die Eigenschaft slot.name fest.

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

Standardmäßig setzt der Connector die slot.drop.on.stop Eigenschaft auf false, 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 der slot.name Eigenschaft anders konfiguriert wurde.

Wir empfehlen, Benachrichtigungen einzurichten, um die WAL-Festplattennutzung auf Ihrem PostgreSQL-Quelldatenbankserver zu beobachten und alle nicht verwendeten Replikationsslots zu löschen.

Tabellenfilter

Standardmäßig erfasst der Connector Änderungsdaten aus jeder Nicht-Systemtabelle 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 eingeschlossen werden sollen.
  • schema.exclude.list: Eine Liste der Schemas, die ausgeschlossen werden sollen. Kann nicht mit schema.include.list verwendet werden.
  • table.include.list: Eine Liste der Tabellen, die eingeschlossen werden sollen.
  • table.exclude.list: Eine Liste der Tabellen, die ausgeschlossen werden sollen. 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 Konfiguration topic.prefix ist.

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

Nächste Schritte