Cloud SQL for PostgreSQL-Quellconnector erstellen

Mit einem Cloud SQL for PostgreSQL-Quellconnector können Sie Änderungen auf Zeilenebene aus einer Cloud SQL for PostgreSQL-Datenbank erfassen und in Themen in einem Managed Service for Apache Kafka-Cluster schreiben.

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.

Hinweis

Bevor Sie einen Cloud SQL for PostgreSQL-Quell-Connector erstellen, müssen Sie Folgendes tun:

• Eine Cloud SQL for PostgreSQL-Instanz mit einer Datenbank. Informationen zum Erstellen dieser Ressourcen finden Sie unter Cloud SQL for PostgreSQL-Datenbank mit der Google Cloud Console erstellen und abfragen.

• Ein Managed Service for Apache Kafka-Cluster.

• Ein Connect-Cluster.

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 Lesen aus Cloud SQL gewähren

Das Managed Kafka-Dienstkonto muss die Berechtigung für den Zugriff auf Cloud SQL for PostgreSQL haben. Weisen Sie dem Dienstkonto die folgenden IAM-Rollen zu:

• Cloud SQL-Client (roles/cloudsql.client)
• Cloud SQL-Instanznutzer (roles/cloudsql.instanceUser)

Das Managed Kafka-Dienstkonto hat das folgende Format: [e-mail], 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.

Datenbank konfigurieren

Bevor Sie den Connector erstellen, müssen Sie die Datenbankreplikation konfigurieren und den Connector für die Authentifizierung bei der Datenbank aktivieren. Diese Schritte werden in den folgenden Abschnitten beschrieben.

Logische Decodierung aktivieren

Ein Cloud SQL for PostgreSQL-Quellconnector basiert auf dem Feature Logische Decodierung von PostgreSQL. Führen Sie die folgenden Schritte aus, um die logische Decodierung in Ihrer Cloud SQL for PostgreSQL-Instanz zu aktivieren.

Console

1. Rufen Sie Cloud SQL > Instanzen auf.

   Zur Seite „VM-Instanzen“

2. Klicken Sie auf den Namen der Instanz.

3. Klicken Sie auf  Bearbeiten.

4. Maximieren Sie Flags und Parameter.

5. Klicken Sie auf Datenbank-Flag hinzufügen.

6. Wählen Sie in der Liste Flag auswählen die Option cloudsql.logical_decoding aus.

7. Wählen Sie für Wert die Option On aus.

8. Klicken Sie auf Fertig.

9. Klicken Sie auf Speichern.

Hinweis: Durch das Aktivieren der logischen Replikation wird die Datenbank neu gestartet.

Weitere Informationen finden Sie unter Logische Replikation und Decodierung einrichten.

Change Data Capture (CDC) konfigurieren

Nachdem Sie die logische Decodierung in Ihrer Instanz aktiviert haben, aktivieren Sie Change Data Capture (CDC) für die Tabellen, die Sie replizieren möchten.

Führen Sie die SQL-Anweisung CREATE PUBLICATION aus, um CDC für eine Tabelle zu aktivieren. Mit dieser Anweisung wird eine Publikation erstellt, die eine Gruppe von Tabellen definiert, die repliziert werden sollen.

• Option 1 Erstellen Sie eine Publikation, die Änderungen für alle Tabellen in der Datenbank repliziert.

  CREATE PUBLICATION dbz_publication FOR ALL TABLES;
  

• Option 2. Erstellen Sie eine Publikation für eine bestimmte Gruppe von Tabellen.

  CREATE PUBLICATION dbz_publication FOR TABLE TABLE_LIST;
  

  Ersetzen Sie TABLE_LIST durch eine durch Kommas getrennte Liste von Tabellen im Format "schema_name"."table_name". Wenn Sie die Schema- und Tabellennamen wie gezeigt in doppelte Anführungszeichen setzen, werden Syntaxfehler vermieden, falls die Namen Sonderzeichen oder Großbuchstaben enthalten.

Standardmäßig verwendet der Connector dbz_publication für den Publikationsnamen. Wenn Sie eine Publikation mit einem anderen Namen verwenden möchten, lesen Sie den Abschnitt Publikationsname.

Nutzerkonto für das Managed Kafka-Dienstkonto erstellen

Der Cloud SQL for PostgreSQL-Quellconnector verwendet die IAM-Datenbankauthentifizierung, um eine Verbindung zur Datenbank herzustellen. Wenn Sie die IAM-Datenbankauthentifizierung aktivieren möchten, fügen Sie das Managed Kafka-Dienstkonto der Cloud SQL-Instanz hinzu:

Console

1. Rufen Sie Cloud SQL > Instanzen auf.

   Zur Seite „VM-Instanzen“

2. Klicken Sie auf den Namen der Instanz.

3. Klicken Sie im Navigationsbereich auf Nutzer.

4. Klicken Sie auf Nutzerkonto hinzufügen.

5. Wählen Sie im Bereich Nutzerkonto hinzufügen die Option Cloud IAM aus.

6. Geben Sie im Feld IAM-Hauptkonto Folgendes ein:

   [e-mail]
   

   Ersetzen Sie PROJECT_NUMBER durch die Projektnummer des Connect-Clusters.

7. Klicken Sie auf Hinzufügen.

gcloud

Zum Abrufen der aktuellen Richtlinie führen Sie den Befehl gcloud sql users create aus:

gcloud sql users create [e-mail] \
  --instance=INSTANCE_NAME \
  --type=cloud_iam_service_account

Ersetzen Sie Folgendes:

• PROJECT_NUMBER: Die Projektnummer des Connect-Clusters.

• INSTANCE_NAME: Der Name der Cloud SQL for PostgreSQL-Instanz.

Aufgrund der Längenbeschränkung für einen Datenbank-Nutzernamen wird das Suffix .gserviceaccount.com aus dem Nutzernamen entfernt, sodass der Nutzername [e-mail] lautet. Wenn Sie SQL-Abfragen ausführen, die auf das IAM-Nutzerkonto verweisen, geben Sie den gekürzten Namen an.

Nutzerkonto konfigurieren

Nachdem Sie das IAM-Nutzerkonto erstellt haben, stellen Sie als Nutzer mit der Rolle cloudsqlsuperuser (z. B. der Standardnutzer postgres) eine Verbindung zur Datenbank her und führen Sie die folgenden SQL-Abfragen aus.

Console

1. Ermöglichen Sie dem Nutzer, das Write-Ahead-Log zu lesen.

   ALTER USER "[e-mail]" WITH REPLICATION;
   

2. Erteilen Sie dem Nutzer die Berechtigung SELECT für die Tabellen.

   GRANT SELECT ON ALL TABLES IN SCHEMA "SCHEMA_NAME"
   TO "[e-mail]";
   

   Alternativ können Sie die Berechtigung SELECT für einzelne Tabellen gewähren. Wenn Sie diese Option auswählen, müssen Sie auch die Konfigurationseigenschaft table.include.list des Connectors auf die Liste der zulässigen Tabellen festlegen. Mit der folgenden SQL-Abfrage wird die Berechtigung SELECT für eine einzelne Tabelle gewährt:

   GRANT SELECT ON TABLE SCHEMA_NAME.TABLE_NAME
   TO "[e-mail]";
   

3. Gewähren Sie dem Nutzer für jede Tabelle Zugriff auf das Schema der Tabelle. Sie können diesen Schritt überspringen, wenn sich die Tabelle im Standardschema public befindet.

   GRANT USAGE ON SCHEMA SCHEMA_NAME
   TO "[e-mail]";
   

Netzwerk konfigurieren

Ein Cloud SQL for PostgreSQL-Quell-Connector kann auf folgende Weise eine Verbindung zur Cloud SQL-Instanz herstellen:

• Private IP-Adresse
• Private Service Connect
• Öffentliche IP-Adresse

Weitere Informationen zu diesen Optionen finden Sie unter Verbindung zu Cloud SQL herstellen. Aus Sicherheitsgründen wird empfohlen, entweder eine private IP-Adresse oder Private Service Connect zu verwenden, da für diese Optionen keine Verbindung zu einer externen IP-Adresse erforderlich ist.

In der folgenden Tabelle sind die Netzwerkanforderungen für die einzelnen Optionen aufgeführt:

IP-Adresstyp	Voraussetzungen
Private IP-Adresse	Konfigurieren Sie eine private IP-Adresse für Ihre Instanz. Weitere Informationen finden Sie unter Private IP-Adresse konfigurieren.
Private Service Connect	Konfigurieren Sie Private Service Connect für Ihre Instanz und rufen Sie den DNS-Namen des Private Service Connect-Endpunkts ab. Weitere Informationen finden Sie unter Verbindung zu einer Instanz über Private Service Connect herstellen. Fügen Sie den DNS-Namen des Endpunkts den auflösbaren DNS-Domains des Connect-Clusters hinzu. Weitere Informationen finden Sie unter Connect-Cluster aktualisieren.
Öffentliche IP-Adresse	Konfigurieren Sie die öffentliche IP-Adresse für Ihre Instanz. Weitere Informationen finden Sie unter Öffentliche IP-Adresse konfigurieren. Richten Sie Public NAT ein, damit die Connect-Cluster-Worker mit dem Internet kommunizieren können. Weitere Informationen finden Sie unter Public NAT einrichten. Geben Sie beim Erstellen des Cloud NAT-Gateways das VPC-Netzwerk an, das das primäre Subnetz des Connect-Clusters enthält.

Quell-Connector für Cloud SQL for PostgreSQL erstellen

So erstellen Sie einen Cloud SQL for PostgreSQL-Quell-Connector:

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 Cloud SQL for PostgreSQL-Quelle aus.

6. Wählen Sie in der Liste Instanz die Cloud SQL-Instanz aus.

7. Wählen Sie in der Liste Datenbank die Cloud SQL-Datenbank aus.

8. Geben Sie im Feld Themenpräfix ein Präfix ein, das für die Kafka-Themennamen verwendet werden soll. Wählen Sie für jeden Cloud SQL for PostgreSQL-Quellconnector ein eindeutiges Präfix aus.

9. Optional: Geben Sie im Feld Tabellennamen eine durch Kommas getrennte Liste der Tabellen ein, aus denen Änderungsdaten gelesen werden sollen, im Format "schema_name"."table_name". Wenn Sie dieses Feld leer lassen, liest der Connector Änderungsdaten aus allen Nicht-Systemtabellen in der Datenbank.

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

    Möglicherweise müssen Sie die Standardwerte für die folgenden Attribute überschreiben:

    • driver.ipTypes: Dieses Attribut muss mit der Netzwerkkonfiguration Ihrer Cloud SQL-Instanz übereinstimmen. Weitere Informationen finden Sie unter IP-Adresstypen.

    • slot.name: Wenn Sie mehrere Instanzen des Connectors für dieselbe Datenbank erstellen, geben Sie für jeden Connector einen eindeutigen Wert an. Weitere Informationen finden Sie unter Replikationsslots.

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

12. 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 ist ein Beispiel für eine Konfigurationsdatei für den Cloud SQL for PostgreSQL-Quellconnector:

   connector.class: io.debezium.connector.postgresql.PostgresConnector
   database.dbname: DATABASE_NAME
   driver.cloudSqlInstance: INSTANCE_ID
   driver.enableIamAuth: "true"
   driver.ipTypes: IP_TYPES
   driver.sslmode: disable
   key.converter: org.apache.kafka.connect.json.JsonConverter
   key.converter.schemas.enable: "false"
   plugin.name: pgoutput
   slot.name: SLOT_NAME
   table.include.list: TABLE_LIST
   topic.prefix: TOPIC_PREFIX
   value.converter: org.apache.kafka.connect.json.JsonConverter
   value.converter.schemas.enable: "true"
   

   Ersetzen Sie Folgendes:

   • INSTANCE_ID: Die ID der Cloud SQL-Instanz, die die Datenbank enthält, im folgenden Format:

     PROJECT_ID:REGION:INSTANCE_NAME
     

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

   • IP_TYPES: Eine durch Kommas getrennte Liste von IP-Adresstypen

   • SLOT_NAME: Der Name des zu erstellenden Replikations-Slots.

   • TABLE_LIST: Eine durch Kommas getrennte Liste von Tabellen, aus denen Änderungsdaten gelesen werden sollen, im Format "schema_name"."table_name".

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

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.

Weitere Informationen finden Sie unter Netzwerk konfigurieren.

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

Warnung: Wenn ein Connector pausiert wird, fehlschlägt oder gelöscht wird, wird sein Replikationsslot nicht automatisch gelöscht. Die Datenbank speichert Write-Ahead-Log-Dateien (WAL) auf unbestimmte Zeit. Dadurch kann sich das Laufwerk des Datenbankservers schnell füllen und es kann zu einem Datenbankausfall kommen.

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

• Fehlerbehebung für einen PostgreSQL-Quellconnector
• Fehlerbehebung für Connect-Cluster und -Connectors
• Allgemeinen PostgreSQL-Quellconnector erstellen

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