Crea un connettore di origine Cloud SQL per PostgreSQL

Un connettore Origine Cloud SQL per PostgreSQL consente di acquisire le modifiche a livello di riga da un database Cloud SQL per PostgreSQL e di scriverle negli argomenti di un cluster Managed Service per Apache Kafka.

I casi d'uso tipici di 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 su eventi.
  • Rispondi agli eventi del database, ad esempio l'inserimento o l'eliminazione 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.

Concedi le autorizzazioni per leggere 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 al account di servizio:

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

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

Se il cluster di connessione si trova in un progetto diverso da quello del 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 abilitare il connettore per l'autenticazione con il database. Le sezioni seguenti descrivono questi passaggi.

Abilita la decodifica logica

Un connettore di origine Cloud SQL per PostgreSQL si basa sulla funzionalità di 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ù, consulta Configura la replica e la decodifica logiche.

Configura Change Data Capture (CDC)

Dopo aver attivato la decodifica logica nell'istanza, attiva 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 replica 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". Racchiudere i nomi di schema e tabella tra virgolette doppie, come mostrato, evita 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 Kafka gestito

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

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.iam.gserviceaccount.com

    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 di progetto del cluster Connect.

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

A causa del limite di lunghezza del nome utente di un database, il suffisso .gserviceaccount.com viene rimosso dal nome utente, che diventa 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. Consenti 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 su 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 Scegli come connetterti a Cloud SQL. Come best practice per la sicurezza, è consigliabile utilizzare un 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 saperne di più, consulta Configura l'IP privato.
Private Service Connect
  1. Configura Private Service Connect per l'istanza e ottieni il nome DNS dell'endpoint Private Service Connect. Per saperne di più, consulta Connettiti a un'istanza utilizzando Private Service Connect.
  2. Aggiungi il nome DNS dell'endpoint ai domini DNS risolvibili del cluster di connessione. Per saperne di più, consulta Aggiorna un cluster connesso.
IP pubblico
  1. Configura l'IP pubblico per l'istanza. Per saperne di più, consulta Configura 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 principale del cluster connesso.

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 contenente 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 ulteriori informazioni sullo snapshot iniziale, consulta Snapshot nella documentazione di Debezium.

Console

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

    Vai a Cluster connessi

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

  3. Fai clic su Crea connettore.

  4. Inserisci una stringa per 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.

  5. Per Plug-in 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 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 nel 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à:

  11. (Facoltativo) Seleziona la Policy di riavvio attività. Per saperne di più, consulta Policy di 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 non è modificabile.

    • 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 del 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 come segue:

      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 di 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 in ordine di preferenza, 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 di 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 utilizzato dal connettore. Il nome dello slot di replica è debezium per impostazione predefinita, a meno che non sia configurato diversamente 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 eventuali slot di replica inutilizzati.

Filtro delle tabelle

Per impostazione predefinita, il connettore acquisisce i dati delle modifiche da ogni tabella non di sistema nel 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 saperne di più, consulta Nomi degli argomenti nella documentazione di Debezium.

Passaggi successivi

Apache Kafka® è un marchio registrato di Apache Software Foundation o delle sue società affiliate negli Stati Uniti e/o in altri paesi.