Change Data Capture (CDC) della tabella di PostgreSQL utilizzando Apache Flink®
Importante
Azure HDInsight su Azure Kubernetes Service è stato ritirato il 31 gennaio 2025. Scopri di più con questo annuncio.
È necessario eseguire la migrazione dei carichi di lavoro a Microsoft Fabric o a un prodotto Azure equivalente per evitare la chiusura brusca dei carichi di lavoro.
Importante
Questa funzionalità è attualmente in anteprima. Le condizioni supplementari per l'utilizzo per le anteprime di Microsoft Azure includono termini legali più validi applicabili alle funzionalità di Azure in versione beta, in anteprima o altrimenti non ancora rilasciate nella disponibilità generale. Per informazioni su questa anteprima specifica, vedere informazioni sull'anteprima di Azure HDInsight su AKS. Per domande o suggerimenti sulle funzionalità, inviare una richiesta in AskHDInsight con i dettagli e seguirci per altri aggiornamenti sulla Community di Azure HDInsight.
Change Data Capture (CDC) è una tecnica che consente di tenere traccia delle modifiche a livello di riga nelle tabelle di database in risposta a operazioni di creazione, aggiornamento ed eliminazione. In questo articolo vengono usati CONNETTORI CDC per Apache Flink®, che offrono un set di connettori di origine per Apache Flink. I connettori integrano Debezium® come motore per catturare le modifiche ai dati.
Flink supporta l'interpretazione dei messaggi JSON Debezium e Avro come messaggi INSERT/UPDATE/DELETE nel sistema SQL Apache Flink.
Questo supporto è utile in molti casi per:
- Sincronizzare i dati incrementali dai database ad altri sistemi
- Log di controllo
- Creare viste materializzate in tempo reale nei database
- Visualizzare la cronologia delle modifiche del join temporale di una tabella di database
Si apprenderà ora come monitorare le modifiche nella tabella PostgreSQL usando Flink-SQL CDC. Il connettore PostgreSQL CDC consente di leggere i dati degli snapshot e i dati incrementali dal database PostgreSQL.
Prerequisiti
- server flessibile di Azure PostgreSQL versione 14.7
- cluster Apache Flink su HDInsight su AKS
- Macchina virtuale Linux per usare il client PostgreSQL
- Aggiungere la regola del gruppo di sicurezza di rete che consente connessioni in ingresso e in uscita sulla porta 5432 nella subnet del pool di HDInsight su Azure Kubernetes Service (AKS).
Preparare la tabella PostgreSQL & Client
Usando una macchina virtuale Linux, installare il client PostgreSQL usando i comandi seguenti
sudo apt-get update sudo apt-get install postgresql-clientInstallare il certificato per connettersi al server PostgreSQL tramite SSL
wget --no-check-certificate https://dl.cacerts.digicert.com/DigiCertGlobalRootCA.crt.pemConnettersi al server (sostituire host, nome utente e database di conseguenza)
psql --host=flinkpostgres.postgres.database.azure.com --port=5432 --username=admin --dbname=postgres --set=sslmode=require --set=sslrootcert=DigiCertGlobalRootCA.crt.pemDopo la connessione al database, creare una tabella di esempio
CREATE TABLE shipments ( shipment_id SERIAL NOT NULL PRIMARY KEY, order_id SERIAL NOT NULL, origin VARCHAR(255) NOT NULL, destination VARCHAR(255) NOT NULL, is_arrived BOOLEAN NOT NULL ); ALTER SEQUENCE public.shipments_shipment_id_seq RESTART WITH 1001; ALTER TABLE public.shipments REPLICA IDENTITY FULL; INSERT INTO shipments VALUES (default,10001,'Beijing','Shanghai',false), (default,10002,'Hangzhou','Shanghai',false), (default,10003,'Shanghai','Hangzhou',false);Per abilitare CDC nel database PostgreSQL, è necessario apportare le modifiche seguenti.
Il livello WAL deve essere modificato in logico. Questo valore può essere modificato nella sezione parametri del server nel portale di Azure.
L'utente che accede alla tabella deve avere il ruolo 'REPLICATION' aggiunto
ALTER USER
<username>WITH REPLICATION;
Creare una tabella Apache Flink PostgreSQL CDC
Per creare la tabella Flink PostgreSQL CDC, scaricare tutti i file JAR dipendenti. Usare il file
pom.xmlcon il contenuto seguente.<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>com.dep.download</groupId> <artifactId>dep-download</artifactId> <version>1.0-SNAPSHOT</version> <!-- https://mvnrepository.com/artifact/com.ververica/flink-sql-connector-sqlserver-cdc --> <dependencies> <dependency> <groupId>com.ververica</groupId> <artifactId>flink-sql-connector-postgres-cdc</artifactId> <version>2.4.2</version> </dependency> </dependencies> </project>Usare il comando maven per scaricare tutti i file JAR dipendenti
mvn -DoutputDirectory=target -f pom.xml dependency:copy-dependencies -XNota
- Se il pod SSH Web non contiene maven, seguire i collegamenti per scaricarlo e installarlo.
- Per scaricare il file jar jsr, usare il comando seguente
wget https://repo1.maven.org/maven2/net/java/loci/jsr308-all/1.1.2/jsr308-all-1.1.2.jar
Dopo aver scaricato i file JAR dipendenti, avviare il client SQL Flink , con questi file JAR da importare nella sessione. Completare il comando come indicato di seguito.
/opt/flink-webssh/bin/sql-client.sh -j /opt/flink-webssh/target/flink-sql-connector-postgres-cdc-2.4.2.jar -j /opt/flink-webssh/target/slf4j-api-1.7.15.jar -j /opt/flink-webssh/target/hamcrest-2.1.jar -j /opt/flink-webssh/target/flink-shaded-guava-31.1-jre-17.0.jar-j /opt/flink-webssh/target/awaitility-4.0.1.jar -j /opt/flink-webssh/target/jsr308-all-1.1.2.jarQuesti comandi avviano il client sql con le dipendenze come,
user@sshnode-0 [ ~ ]$ bin/sql-client.sh -j flink-sql-connector-postgres-cdc-2.4.2.jar -j slf4j-api-1.7.15.jar -j hamcrest-2.1.jar -j flink-shaded-guava-31.1-jre-17.0.jar -j awaitility-4.0.1.jar -j jsr308-all-1.1.2.jar ???????? ???????????????? ??????? ??????? ? ???? ????????? ????? ??? ??????? ????? ??? ??? ????? ?? ??????????????? ?? ? ??? ?????? ????? ????? ???? ????? ????? ??????? ??? ??????? ??? ????????? ?? ?? ?????????? ???????? ?? ? ?? ??????? ???? ??? ? ?? ???????? ????? ???? ? ?? ? ?? ???????? ???? ?? ???? ???? ?????????? ??? ?? ???? ???? ?? ??? ??????????? ???? ? ? ??? ??? ?? ??? ????????? ???? ??? ?? ? ??????? ???????? ??? ?? ??? ??? ???????????????????? ???? ? ????? ??? ?????? ???????? ???? ?? ???????? ??????????????? ?? ?? ???? ??????? ??? ?????? ?? ??? ??? ??? ??? ??????? ???? ????????????? ??? ????? ???? ?? ?? ???? ??? ?? ??? ? ?? ?? ?? ?? ?? ?? ?? ???????? ?? ????? ?? ??????????? ?? ?? ???? ? ??????? ?? ??? ????? ?? ??????????? ???? ???? ??????? ???????? ????? ?? ???? ????? ????????????????????????????????? ????? ______ _ _ _ _____ ____ _ _____ _ _ _ BETA | ____| (_) | | / ____|/ __ \| | / ____| (_) | | | |__ | |_ _ __ | | __ | (___ | | | | | | | | |_ ___ _ __ | |_ | __| | | | '_ \| |/ / \___ \| | | | | | | | | |/ _ \ '_ \| __| | | | | | | | | < ____) | |__| | |____ | |____| | | __/ | | | |_ |_| |_|_|_| |_|_|\_\ |_____/ \___\_\______| \_____|_|_|\___|_| |_|\__| Welcome! Enter 'HELP;' to list all available commands. 'QUIT;' to exit. Command history file path: /home/xcao/.flink-sql-history Flink SQL>Creare una tabella Flink PostgreSQL CDC usando il connettore CDC
CREATE TABLE shipments ( shipment_id INT, order_id INT, origin STRING, destination STRING, is_arrived BOOLEAN, PRIMARY KEY (shipment_id) NOT ENFORCED ) WITH ( 'connector' = 'postgres-cdc', 'hostname' = 'flinkpostgres.postgres.database.azure.com', 'port' = '5432', 'username' = 'username', 'password' = 'password', 'database-name' = 'postgres', 'schema-name' = 'public', 'table-name' = 'shipments', 'decoding.plugin.name' = 'pgoutput', 'slot.name' = 'flink' );
Validazione
Eseguire il comando 'select *' per monitorare le modifiche.
select * from shipments;
Riferimento
- sito Web Apache Flink
- Il connettore PostgreSQL CDC è concesso in licenza sotto la Licenza Apache 2.0
- Apache, Apache Flink, Flink e i nomi dei progetti open source associati sono marchi della Apache Software Foundation (ASF).