Crea un connettore di origine PostgreSQL generico

Questo documento descrive come creare un connettore di origine PostgreSQL generico.

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

I casi d'uso per questo connettore includono:

  • Monitorare le modifiche a livello di riga del database in tempo reale.
  • Integrare gli eventi di modifica del database in un'architettura basata sugli eventi.
  • Rispondere agli eventi del database, ad esempio inserimenti o eliminazioni di righe.
  • Copiare le modifiche del database in altri sistemi.
  • Eseguire la replica o il ripristino delle tabelle PostgreSQL.

Prima di iniziare

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

  • Un database PostgreSQL.

  • Un cluster di connessione associato al cluster Kafka.

  • Crea un secret di Secret Manager che memorizzi la password del database. Se la configurazione utilizza SSL del database, crea anche un secret per la password SSL del database. Configura il cluster di connessione con i secret. Per ulteriori informazioni, 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 Editor connettore Kafka gestito (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 avere l'autorizzazione per visualizzare e accedere ai secret archiviati in Secret Manager. Concedi i seguenti ruoli IAM al account di servizio:

  • Visualizzatore Secret Manager (roles/secretmanager.viewer)
  • Accessore secret Secret Manager (roles/secretmanager.secretAccessor)

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 di connessione.

Se il cluster di connessione si trova in un progetto diverso dal cluster Managed Service per Apache Kafka, consulta Creare 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 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 l'autenticazione del connettore a PostgreSQL. L'utente del database deve essere 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 di cui vuoi acquisire i dati. Il connettore si abbona 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 replichi 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 lasciare che il connettore la crei automaticamente. Per ulteriori informazioni, 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'autorizzazione di selezione 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 Google Cloud console, vai alla pagina Cluster di connessione.

    Vai a Cluster di connessione

  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 le linee guida su come assegnare un nome a un connettore, consulta Linee guida per assegnare un nome a una risorsa Managed Service per Apache Kafka.

  5. Per Plug-in connettore, seleziona Origine PostgreSQL generica.

  6. Nel campo Nome host 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 ulteriori informazioni, consulta Configurare il connettore.

  12. (Facoltativo) Seleziona Norme di riavvio delle attività. Per ulteriori informazioni, consulta Norme di riavvio delle attività.

  13. Fai clic su Crea.

gcloud

  1. Nella Google Cloud console, attiva Cloud Shell.

    Attiva Cloud Shell

    Nella parte inferiore della Google Cloud console, 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à inclusa e installata e con valori già impostati per il progetto corrente. L'inizializzazione della sessione può richiedere alcuni secondi.

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

    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 le linee guida su come assegnare un nome a un connettore, consulta 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 di connessione in cui viene creato il connettore.

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

    Di seguito è riportato 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 al database.

    • CREDENTIALS: un percorso del 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

Nei file di configurazione 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 di origine PostgreSQL generico trasmette gli eventi di modifica da una pubblicazione nel database. Puoi creare la pubblicazione manualmente o lasciare che il connettore la crei automaticamente.

L'publication.autocreate.mode impostazione specifica come e se 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 in modo da 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 superutente.

    I ruoli di superutente 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 delle 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 di 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 slot.name proprietà.

Ti consigliamo di configurare gli avvisi per monitorare l'utilizzo del disco WAL sul server di database PostgreSQL di origine ed eliminare gli 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 ulteriori informazioni, consulta Nomi degli argomenti nella documentazione di Debezium.

Passaggi successivi