Crea un connettore di origine Cloud SQL per PostgreSQL

Questo documento descrive come creare un connettore di origine Cloud SQL per PostgreSQL per Kafka Connect.

Un connettore di origine Cloud SQL per PostgreSQL è un'istanza di un connettore Debezium PostgreSQL. Legge le modifiche a livello di riga da un database Cloud SQL per PostgreSQL e le scrive negli argomenti di un cluster Managed Service per Apache Kafka.

I casi d'uso per questo connettore includono:

  • Monitora le modifiche al database a livello di riga in tempo reale.
  • Integra gli eventi di modifica del database in un'architettura basata sugli eventi.
  • Rispondere a eventi di database come inserimenti o eliminazioni di righe.
  • Copia le modifiche al database in altri sistemi.

Prima di iniziare

Prima di creare un connettore di origine Cloud SQL per PostgreSQL, assicurati di disporre di quanto segue:

Ruoli e autorizzazioni richiesti

Per ottenere le autorizzazioni necessarie per creare un connettore, chiedi all'amministratore di concederti il ruolo IAM Managed Kafka Connector Editor (roles/managedkafka.connectorEditor) nel progetto. Per saperne di più sulla concessione dei ruoli, consulta Gestisci l'accesso a progetti, cartelle e organizzazioni.

Questo ruolo predefinito contiene le autorizzazioni necessarie per creare un connettore. Per vedere quali sono esattamente le autorizzazioni richieste, espandi la sezione Autorizzazioni obbligatorie:

Autorizzazioni obbligatorie

Per creare un connettore sono necessarie le seguenti autorizzazioni:

  • Crea un connettore: managedkafka.connectors.create

Potresti anche ottenere queste autorizzazioni con ruoli personalizzati o altri ruoli predefiniti.

Concedere le autorizzazioni per la lettura da Cloud SQL

Il account di servizio Kafka gestito deve disporre dell'autorizzazione per accedere a Cloud SQL per PostgreSQL. Concedi i seguenti ruoli IAM alaccount di serviziot:

  • Client Cloud SQL (roles/cloudsql.client)
  • Utente dell'istanza Cloud SQL (roles/cloudsql.instanceUser)

Il account di servizio Kafka gestito ha il seguente formato: service-PROJECT_NUMBER@gcp-sa-managedkafka., dove PROJECT_NUMBER è il numero di progetto del cluster Connect.

Se il cluster di connessione si trova in un progetto diverso dal cluster Managed Service per Apache Kafka, consulta Crea un cluster di connessione in un progetto diverso.

Configura il database

Prima di creare il connettore, devi configurare la replica del database e attivare l'autenticazione del connettore con il database. Le sezioni seguenti descrivono questi passaggi.

Abilita decodifica logica

Un connettore Origine Cloud SQL per PostgreSQL si basa sulla decodifica logica di PostgreSQL. Per abilitare la decodifica logica nell'istanza Cloud SQL per PostgreSQL, segui questi passaggi.

Console

  1. Vai a Cloud SQL > Istanze.

    Vai a Istanze

  2. Fai clic sul nome dell'istanza.

  3. Fai clic su Modifica.

  4. Espandi Flag e parametri.

  5. Fai clic su Aggiungi un flag di database.

  6. Nell'elenco Scegli un flag, seleziona cloudsql.logical_decoding.

  7. In Valore, seleziona On.

  8. Fai clic su Fine.

  9. Fai clic su Salva.

Per saperne di più, vedi Configurazione di replica e decodifica logiche.

Configura Change Data Capture (CDC)

Dopo aver abilitato la decodifica logica nell'istanza, abilita Change Data Capture (CDC) per le tabelle che vuoi replicare.

Per abilitare CDC per una tabella, esegui l'istruzione SQL CREATE PUBLICATION. Questa istruzione crea una pubblicazione, che definisce un gruppo di tabelle da replicare.

  • Opzione 1. Crea una pubblicazione che replichi le modifiche per tutte le tabelle del database.

    CREATE PUBLICATION dbz_publication FOR ALL TABLES;

  • Opzione 2. Crea una pubblicazione per un insieme specifico di tabelle.

    CREATE PUBLICATION dbz_publication FOR TABLE TABLE_LIST;

    Sostituisci TABLE_LIST con un elenco separato da virgole di tabelle, nel formato "schema_name"."table_name". L'inserimento dei nomi di schema e tabella tra virgolette doppie, come mostrato, impedisce errori di sintassi se i nomi contengono caratteri speciali o lettere maiuscole.

Per impostazione predefinita, il connettore utilizza dbz_publication per il nome della pubblicazione. Per utilizzare una pubblicazione con un nome diverso, consulta Nome pubblicazione.

Crea un account utente per il account di servizio Managed Kafka

Il connettore di origine Cloud SQL per PostgreSQL utilizza l'autenticazione IAM dei database per connettersi al database. Per abilitare l'autenticazione del database IAM, aggiungi il account di servizio Kafka gestito all'istanza Cloud SQL nel seguente modo:

Console

  1. Vai a Cloud SQL > Istanze

    Vai a Istanze

  2. Fai clic sul nome dell'istanza.

  3. Nel riquadro di navigazione, fai clic su Utenti.

  4. Fai clic su Aggiungi account utente.

  5. Nel riquadro Aggiungi un account utente, seleziona Cloud IAM.

  6. Nel campo Entità IAM, inserisci quanto segue:

    service-PROJECT_NUMBER@gcp-sa-managedkafka.

    Sostituisci PROJECT_NUMBER con il numero di progetto del cluster Connect.

  7. Fai clic su Aggiungi.

gcloud

Esegui il comando gcloud sql users create:

gcloud sql users create service-PROJECT_NUMBER@gcp-sa-managedkafka.iam \
  --instance=INSTANCE_NAME \
  --type=cloud_iam_service_account

Sostituisci quanto segue:

  • PROJECT_NUMBER: il numero del progetto del cluster Connect.

  • INSTANCE_NAME: il nome dell'istanza Cloud SQL per PostgreSQL.

A causa del limite di lunghezza di un nome utente del database, il suffisso . viene eliminato dal nome utente, quindi il nome utente è service-PROJECT_NUMBER@gcp-sa-managedkafka.iam. Quando esegui query SQL che fanno riferimento all'account utente IAM, specifica il nome troncato.

Configura l'account utente

Dopo aver creato l'account utente IAM, connettiti al database come utente con il ruolo cloudsqlsuperuser (ad esempio l'utente postgres predefinito) ed esegui le seguenti query SQL.

Console

  1. Consente all'utente di leggere il log write-ahead.

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

  2. Concedi all'utente l'autorizzazione SELECT sulle tabelle.

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

    In alternativa, puoi concedere l'autorizzazione SELECT per singole tabelle. Se scegli questa opzione, devi anche impostare la proprietà di configurazione table.include.list del connettore sull'elenco delle tabelle consentite. La seguente query SQL concede l'autorizzazione SELECT su una singola tabella:

    GRANT SELECT ON TABLE SCHEMA_NAME.TABLE_NAME
TO "service-PROJECT_NUMBER@gcp-sa-managedkafka.iam";

  3. Per ogni tabella, concedi all'utente l'accesso allo schema della tabella. Puoi saltare questo passaggio se la tabella si trova nello schema public predefinito.

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

Configura il networking

Un connettore di origine Cloud SQL per PostgreSQL può connettersi all'istanza Cloud SQL nei seguenti modi:

  • IP privato
  • Private Service Connect
  • IP pubblico

Per saperne di più su queste opzioni, consulta la pagina Scegliere come connettersi a Cloud SQL. Come best practice per la sicurezza, è consigliabile utilizzare l'IP privato o Private Service Connect, perché queste opzioni non richiedono la connessione a un indirizzo IP esterno.

La tabella seguente mostra i requisiti di rete per ogni opzione:

Tipo di indirizzo IP Requisiti
IP privato Configura l'IP privato per l'istanza. Per ulteriori informazioni, consulta Configura l'IP privato.
Private Service Connect
  1. Configura Private Service Connect per la tua istanza e recupera il nome DNS dell'endpoint Private Service Connect. Per ulteriori informazioni, consulta Connettersi a un'istanza utilizzando Private Service Connect.
  2. Aggiungi il nome DNS dell'endpoint a Domini DNS risolvibili del cluster di connessione. Per saperne di più, vedi Aggiornare un cluster Connect.
IP pubblico
  1. Configura l'IP pubblico per l'istanza. Per saperne di più, consulta Configurare l'IP pubblico.
  2. Configura Public NAT per consentire ai worker del cluster Connect di comunicare con internet. Per saperne di più, consulta Configura Public NAT. Quando crei il gateway Cloud NAT, specifica la rete VPC che contiene la subnet primaria del cluster Connect.

Crea un connettore di origine Cloud SQL per PostgreSQL

Per creare un connettore di origine Cloud SQL per PostgreSQL, segui questi passaggi.

Quando il connettore viene inizializzato, esegue le seguenti azioni:

  1. Crea uno snapshot iniziale del database.
  2. Crea un argomento Kafka per ogni tabella con righe.
  3. Per ogni riga del database, invia un evento di modifica all'argomento corrispondente.

Mentre il connettore è in esecuzione, continua a inviare eventi di modifica agli argomenti. Per saperne di più sullo snapshot iniziale, consulta la sezione Snapshot nella documentazione di Debezium.

Console

  1. Nella console Google Cloud , vai alla pagina Connetti cluster.

    Vai a Connetti cluster

  2. Fai clic sul cluster di connessione in cui vuoi creare il connettore.

  3. Fai clic su Crea connettore.

  4. Per il nome del connettore, inserisci una stringa.

    Per maggiori informazioni su come assegnare un nome a un connettore, consulta le linee guida per assegnare un nome a una risorsa Managed Service per Apache Kafka.

  5. Per Plug-in del connettore, seleziona Origine Cloud SQL per PostgreSQL.

  6. Nell'elenco Istanza, seleziona l'istanza Cloud SQL.

  7. Nell'elenco Database, seleziona il database Cloud SQL.

  8. Nel campo Prefisso argomento, inserisci un prefisso da utilizzare per i nomi degli argomenti Kafka. Scegli un prefisso univoco per ogni connettore di origine Cloud SQL per PostgreSQL.

  9. (Facoltativo) Nel campo Nomi delle tabelle, inserisci un elenco separato da virgole delle tabelle da cui leggere i dati delle modifiche, nel formato "schema_name"."table_name". Se lasci vuoto questo campo, il connettore legge i dati delle modifiche da tutte le tabelle non di sistema del database.

  10. (Facoltativo) Nella casella Configurazioni, aggiungi le proprietà di configurazione o modifica le proprietà predefinite. Per saperne di più, consulta Configura il connettore.

    Potresti dover eseguire l'override dei valori predefiniti per le seguenti proprietà:

    • driver.ipTypes: questa proprietà deve corrispondere alla configurazione di rete della tua istanza Cloud SQL. Consulta la sezione Tipi di indirizzi IP.

    • slot.name: se crei più istanze del connettore per lo stesso database, specifica un valore univoco per ogni connettore. Vedi Slot di replica.

  11. (Facoltativo) Seleziona la policy di riavvio attività. Per saperne di più, consulta le norme sul riavvio delle attività.

  12. Fai clic su Crea.

gcloud

  1. Nella console Google Cloud , attiva Cloud Shell.

    Attiva Cloud Shell

    Nella parte inferiore della console Google Cloud viene avviata una sessione di Cloud Shell e viene visualizzato un prompt della riga di comando. Cloud Shell è un ambiente shell con Google Cloud CLI già installata e con valori già impostati per il progetto corrente. L'inizializzazione della sessione può richiedere alcuni secondi.

  2. Esegui il comando gcloud managed-kafka connectors create:

    gcloud managed-kafka connectors create CONNECTOR_ID \
    --location=LOCATION \
    --connect-cluster=CONNECT_CLUSTER_ID \
    --config-file=CONFIG_FILE

    Sostituisci quanto segue:

    • CONNECTOR_ID: l'ID o il nome del connettore. Per maggiori informazioni su come assegnare un nome a un connettore, consulta le linee guida per assegnare un nome a una risorsa Managed Service per Apache Kafka. Il nome di un connettore è immutabile.

    • LOCATION: la località in cui crei il connettore. Deve essere la stessa località in cui hai creato il cluster di connessione.

    • CONNECT_CLUSTER_ID: l'ID del cluster Connect in cui viene creato il connettore.

    • CONFIG_FILE: il percorso del file di configurazione YAML per il connettore.

    Ecco un esempio di file di configurazione per il connettore di origine Cloud SQL per PostgreSQL:

    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"

    Sostituisci quanto segue:

    • INSTANCE_ID: l'ID dell'istanza Cloud SQL che contiene il database, formattato nel seguente modo:

      PROJECT_ID:REGION:INSTANCE_NAME

    • DATABASE_NAME: il nome del database Cloud SQL da cui leggere.

    • IP_TYPES: un elenco separato da virgole di tipi di indirizzi IP

    • SLOT_NAME: il nome dello slot di replica da creare.

    • TABLE_LIST: un elenco separato da virgole delle tabelle da cui leggere i dati delle modifiche, nel formato "schema_name"."table_name".

    • TOPIC_PREFIX: un prefisso da utilizzare per i nomi degli argomenti Kafka.

Configura il connettore

Questa sezione descrive alcune proprietà di configurazione che puoi impostare sul connettore. Per un elenco completo, consulta Connettore Debezium per PostgreSQL nella documentazione di Debezium.

Tipi di indirizzi IP

La proprietà driver.ipTypes specifica il tipo di indirizzo IP utilizzato dal connettore per connettersi al database:

  • PRIVATE: IP privato
  • PSC: Private Service Connect
  • PUBLIC: IP pubblico

La proprietà driver.ipTypes contiene un elenco separato da virgole di tipi di IP nell'ordine preferito; ad esempio, driver.ipTypes=PRIVATE,PUBLIC.

Per saperne di più, consulta Configura il networking.

Nome pubblicazione

Per impostazione predefinita, il connettore tenta di eseguire lo streaming da una pubblicazione denominata dbz_publication. Per specificare una pubblicazione diversa, aggiungi publication.name=PUBLICATION_NAME alla configurazione, dove PUBLICATION_NAME è il nome della pubblicazione. Esempio: publication.name=my_publication.

Slot di replica

PostgreSQL utilizza gli slot di replica per trasmettere in streaming le modifiche alle tabelle del database. Per impostazione predefinita, il connettore crea uno slot di replica denominato debezium. Per utilizzare un nome dello slot diverso, imposta la proprietà slot.name.

Se crei due istanze del connettore per lo stesso database, devi specificare un nome slot univoco per ogni connettore.

Per impostazione predefinita, il connettore imposta la proprietà slot.drop.on.stop su false per evitare la perdita di dati. Quando elimini definitivamente un connettore, devi eliminare manualmente lo slot di replica che il connettore utilizzava. Il nome dello slot di replica è impostato per impostazione predefinita su debezium, a meno che non sia configurato in modo diverso utilizzando la proprietà slot.name.

Ti consigliamo di configurare avvisi per monitorare l'utilizzo del disco WAL sul server di database PostgreSQL di origine e di eliminare gli slot di replica inutilizzati.

Filtro tabella

Per impostazione predefinita, il connettore acquisisce i dati delle modifiche da ogni tabella non di sistema del database. Per filtrare le tabelle acquisite, specifica una o più delle seguenti impostazioni:

  • schema.include.list. Un elenco di schemi da includere.
  • schema.exclude.list. Un elenco di schemi da escludere. Non può essere utilizzato con schema.include.list.
  • table.include.list. Un elenco di tabelle da includere.
  • table.exclude.list. Un elenco di tabelle da escludere. Non può essere utilizzato con table.include.list.

Nomi degli argomenti

Per impostazione predefinita, il connettore crea argomenti Kafka con la seguente convenzione di denominazione: topic_prefix.schema.table_name, dove topic.prefix è il valore della configurazione topic.prefix.

Per ulteriori informazioni, consulta la sezione Nomi degli argomenti nella documentazione di Debezium.

Passaggi successivi