Crea un connettore di origine PostgreSQL generico

Un connettore Origine PostgreSQL generica ti consente di acquisire le modifiche a livello di riga da un database PostgreSQL e di trasmetterle in flussi di dati agli 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.
  • Replica o ripristina le tabelle PostgreSQL.

Per leggere le modifiche da un database Cloud SQL per PostgreSQL, utilizza il connettore di origine Cloud SQL per PostgreSQL.

Prima di iniziare

Prima di creare un connettore di origine PostgreSQL generico, assicurati di disporre di quanto segue:

  • Un database PostgreSQL.

  • Un cluster di connessione associato al cluster Kafka.

  • Crea un secret di Secret Manager che memorizza la password del database. Se la tua configurazione utilizza SSL per il database, crea anche un secret per la password SSL del database. Configura il cluster di connessione con i secret. Per saperne di più, consulta Risorse di Secret Manager.

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 accedere ai secret di Secret Manager

Il account di servizio Kafka gestito deve disporre dell'autorizzazione per visualizzare e accedere ai secret archiviati in Secret Manager. Concedi i seguenti ruoli IAM al account di servizio:

  • Secret Manager Viewer (roles/secretmanager.viewer)
  • Secret Manager Secret Accessor (roles/secretmanager.secretAccessor)

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 PostgreSQL

Per consentire al connettore di leggere gli eventi di modifica dei dati dal tuo database, configura le seguenti impostazioni.

  1. Imposta wal_level del server su logical.

    ALTER SYSTEM SET wal_level = logical;

    Riavvia il server per applicare l'impostazione.

  2. Crea un utente del database per consentire al connettore di autenticarsi su PostgreSQL. L'utente del database deve avere un ruolo di replica, che gli consenta di connettersi al server in modalità di replica.

    CREATE ROLE ROLE_NAME WITH REPLICATION LOGIN PASSWORD 'ROLE_PASSWORD';

    Sostituisci quanto segue:

    • ROLE_NAME: il nome dell'utente, ad esempio debezium_user.
    • ROLE_PASSWORD: la password dell'utente.

  3. Crea una pubblicazione per le tabelle che vuoi acquisire. Il connettore si iscrive alla pubblicazione per ricevere gli eventi di modifica dei dati.

    CREATE PUBLICATION dbz_publication FOR TABLE "SCHEMA_NAME"."TABLE_NAME";

    Sostituisci quanto segue:

    • SCHEMA_NAME: lo schema della tabella.

    • TABLE_NAME: il nome della tabella.

    Ti consigliamo di racchiudere i nomi di schema e tabella tra virgolette doppie, come mostrato, per evitare errori di sintassi se i nomi contengono caratteri speciali o lettere maiuscole.

    In alternativa, puoi creare una pubblicazione che replica le modifiche per tutte le tabelle del database:

    CREATE PUBLICATION dbz_publication FOR ALL TABLES;

    A seconda dell'impostazione publication.autocreate.mode del connettore, puoi creare la pubblicazione manualmente o consentire al connettore di crearla automaticamente. Per saperne di più, consulta Modalità di pubblicazione.

  4. Per ogni tabella, concedi i privilegi SELECT sulla tabella all'utente del database.

    GRANT SELECT ON TABLE "SCHEMA_NAME"."TABLE_NAME" TO ROLE_NAME;

    In alternativa, puoi concedere l'accesso SELECT su tutte le tabelle di uno schema:

    GRANT SELECT ON ALL TABLES IN SCHEMA "SCHEMA_NAME" TO ROLE_NAME;

  5. Per ogni tabella, concedi i privilegi USAGE sullo schema della tabella all'utente del database. Puoi saltare questo passaggio se la tabella si trova nello schema public predefinito.

    GRANT USAGE ON SCHEMA "SCHEMA_NAME" TO ROLE_NAME;

Crea un connettore di origine PostgreSQL generico

Per creare un connettore di origine PostgreSQL generico, 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. In Plug-in connettore, seleziona Origine PostgreSQL generica.

  6. Nel campo Nome host del database, inserisci il nome host o l'indirizzo IP del server PostgreSQL.

  7. Nel campo Nome database, inserisci il nome del database.

  8. Nel campo Utente database, inserisci il nome del ruolo di replica. Il connettore esegue l'autenticazione al server PostgreSQL utilizzando questo ruolo.

  9. Nel campo Prefisso argomento, inserisci un prefisso da utilizzare per i nomi degli argomenti Kafka.

  10. Nell'elenco Secret, seleziona il secret che contiene la password del database.

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

  12. (Facoltativo) Seleziona la Policy di riavvio attività. Per saperne di più, consulta Policy di riavvio delle attività.

  13. 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 PostgreSQL generico:

    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"

    Sostituisci quanto segue:

    • HOSTNAME: il nome host del database PostgreSQL da cui leggere.

    • DATABASE_NAME: il nome del database PostgreSQL da cui leggere.

    • DATABASE_USER: l'utente del database PostgreSQL da utilizzare per l'autenticazione nel database.

    • CREDENTIALS: un percorso al secret di Secret Manager che contiene la password del database. Specifica il secret utilizzando il seguente formato:

      ${directory:/var/secrets:PROJECT_ID-SECRET_NAME-SECRET_VERSION}

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

Configurazioni di password e password SSL

Nelle configurazioni database.password e database.sslpassword sono supportati solo i percorsi dei secret. Il backend prevede che queste configurazioni utilizzino il seguente formato: ${directory:/var/secrets:PROJECT_ID-SECRET_NAME-SECRET_VERSION}.

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.

Modalità di pubblicazione

Un connettore Origine PostgreSQL generica trasmette gli eventi di modifica da una pubblicazione nel database. Puoi creare la pubblicazione manualmente o lasciare che il connettore la crei automaticamente.

L'impostazione publication.autocreate.mode specifica se e come il connettore deve creare una pubblicazione.

  • filtered. Se la pubblicazione non esiste, il connettore ne crea una nuova che include solo le tabelle acquisite. L'utente del database deve disporre delle autorizzazioni CREATE sul database ed essere il proprietario delle tabelle incluse.

    Se la pubblicazione esiste già, il connettore la modifica per includere le tabelle acquisite. Per modificare una pubblicazione esistente, l'utente del database deve essere il proprietario della pubblicazione e delle tabelle incluse.

  • all_tables. Se la pubblicazione non esiste, il connettore ne crea una nuova utilizzando il parametro FOR ALL TABLES. L'utente del database deve essere un super utente.

    I ruoli di super utente ignorano tutti i controlli delle autorizzazioni in un database, pertanto non è consigliabile concedere SUPERUSER all'utente del database. In alternativa, crea la pubblicazione manualmente o imposta publication.autocreate.mode=filtered.

  • disabled. Se la pubblicazione non esiste, si verifica un errore. Il connettore non crea una nuova pubblicazione.

Il valore predefinito è all_tables.

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.