Cloud SQL for PostgreSQL-Quellconnector erstellen

In diesem Dokument wird beschrieben, wie Sie einen Cloud SQL for PostgreSQL-Quell-Connector für Kafka Connect erstellen.

Ein Cloud SQL for PostgreSQL-Quellconnector ist eine Instanz eines Debezium PostgreSQL-Connectors. Er liest Änderungen auf Zeilenebene aus einer Cloud SQL for PostgreSQL-Datenbank und schreibt sie in Themen in einem Managed Service for Apache Kafka-Cluster.

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 Quell-Connector für Cloud SQL for PostgreSQL erstellen, müssen Sie Folgendes haben:

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

Datenbank konfigurieren

Bevor Sie den Connector erstellen, müssen Sie die Datenbankreplikation konfigurieren und dem Connector die Authentifizierung bei der Datenbank ermöglichen. Diese Schritte werden in den folgenden Abschnitten beschrieben.

Logische Decodierung aktivieren

Ein Cloud SQL for PostgreSQL-Quellconnector basiert auf der Funktion Logische Dekodierung 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.

Weitere Informationen finden Sie unter Logische Replikation und Decodierung einrichten.

Change Data Capture (CDC) konfigurieren

Nachdem Sie die logische Dekodierung 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 Veröffentlichung erstellt, die eine Gruppe von Tabellen definiert, die repliziert werden sollen.

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

    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 Namen der Publikation. Wenn Sie eine Publikation mit einem anderen Namen verwenden möchten, lesen Sie den Abschnitt Name der Publikation.

Nutzerkonto für das Dienstkonto von Managed Kafka erstellen

Der Cloud SQL for PostgreSQL-Quellconnector verwendet die IAM-Datenbankauthentifizierung, um eine Verbindung zur Datenbank herzustellen. So aktivieren Sie die IAM-Datenbankauthentifizierung: Fügen Sie das Dienstkonto für Managed Kafka 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:

    service-PROJECT_NUMBER@gcp-sa-managedkafka.iam.gserviceaccount.com

    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 service-PROJECT_NUMBER@gcp-sa-managedkafka.iam \
  --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. Der Nutzername lautet also service-PROJECT_NUMBER@gcp-sa-managedkafka.iam. 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. als Standardnutzer postgres) eine Verbindung zur Datenbank her und führen Sie die folgenden SQL-Abfragen aus.

Console

  1. Dem Nutzer erlauben, das Write-Ahead-Log zu lesen.

    ALTER USER "service-PROJECT_NUMBER@gcp-sa-managedkafka.iam" WITH REPLICATION;

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

    GRANT SELECT ON ALL TABLES IN SCHEMA "SCHEMA_NAME"
TO "service-PROJECT_NUMBER@gcp-sa-managedkafka.iam";

    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 "service-PROJECT_NUMBER@gcp-sa-managedkafka.iam";

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

    GRANT USAGE ON SCHEMA SCHEMA_NAME
TO "service-PROJECT_NUMBER@gcp-sa-managedkafka.iam";

Netzwerk konfigurieren

Ein Cloud SQL for PostgreSQL-Quell-Connector kann auf folgende Arten 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
  1. 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.
  2. 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
  1. Konfigurieren Sie eine öffentliche IP-Adresse für Ihre Instanz. Weitere Informationen finden Sie unter Öffentliche IP-Adresse konfigurieren.
  2. Richten Sie Public NAT ein, damit die Connect-Cluster-Worker mit dem Internet kommunizieren können. Weitere Informationen finden Sie unter Public NAT einrichten. Wenn Sie das Cloud NAT-Gateway erstellen, geben Sie das VPC-Netzwerk an, das das primäre Subnetz des Connect-Clusters enthält.

Cloud SQL for PostgreSQL-Quellconnector erstellen

Führen Sie die folgenden Schritte aus, um einen Cloud SQL for PostgreSQL-Quell-Connector zu erstellen.

Wenn der Connector initialisiert wird, werden die folgenden Aktionen ausgeführt:

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

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

    Zu „Cluster verbinden“

  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 Connectornamen einen String ein.

    Tipps zum Benennen von Connectors finden Sie unter 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 Namen der Kafka-Themen 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 Konfigurationseigenschaften hinzu oder bearbeiten Sie die Standardeigenschaften. Weitere Informationen finden Sie unter Connector konfigurieren.

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

    • driver.ipTypes: Diese Eigenschaft 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 Replikations-Slots.

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

    • LOCATION: Der Ort, 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, in folgendem 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-Adressentypen

    • 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 Namen von Kafka-Themen verwendet werden soll.

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

IP-Adresstypen

Mit dem Attribut driver.ipTypes wird der Typ der IP-Adresse angegeben, die vom Connector verwendet wird, um eine Verbindung zur Datenbank herzustellen:

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

Die Eigenschaft driver.ipTypes enthält eine durch Kommas getrennte Liste von IP-Typen in bevorzugter 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 Replikations-Slots, um Änderungen an Datenbanktabellen zu streamen. Standardmäßig erstellt der Connector einen Replikations-Slot 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 wird für die Property slot.drop.on.stop der Wert false festgelegt, um Datenverlust zu verhindern. Wenn Sie einen Connector endgültig löschen, müssen Sie den Replikations-Slot, den der Connector verwendet hat, manuell löschen. Der Name des Replikationsslots ist standardmäßig debezium, sofern er nicht mit der Property slot.name anders konfiguriert wird.

Wir empfehlen, Benachrichtigungen einzurichten, um die WAL-Festplattennutzung auf Ihrem PostgreSQL-Quelldatenbankserver zu überwachen und alle nicht verwendeten Replikations-Slots 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 einzuschließenden Schemas.
  • 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 Konfiguration topic.prefix ist.

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

Nächste Schritte