AnalyticDB:Gunakan Flink CDC untuk berlangganan data penuh dan inkremental secara real time (pratinjau undangan)

Langkah 1: Siapkan tabel uji dan data uji

1. Masuk ke konsol AnalyticDB for PostgreSQL, temukan instans target, lalu klik ID instans.

2. Di pojok kanan bawah halaman Basic Information, klik Log On to Database.

3. Buat database uji dan tabel sumber bernama adbpg_source_table, lalu masukkan 50 baris data ke dalam tabel sumber.

   -- Buat database uji.
   CREATE DATABASE testdb;
   -- Beralih ke database testdb dan buat skema.
   CREATE SCHEMA testschema;
   -- Buat tabel sumber bernama adbpg_source_table.
   CREATE TABLE testschema.adbpg_source_table(
     id int,
     username text,
     PRIMARY KEY(id)
   );
   -- Masukkan 50 baris data ke dalam tabel adbpg_source_table.
   INSERT INTO testschema.adbpg_source_table(id, username)
   SELECT i, 'username'||i::text
   FROM generate_series(1, 50) AS t(i);

4. Buat tabel sink bernama adbpg_sink_table agar Flink dapat menulis data hasilnya.

   CREATE TABLE testschema.adbpg_sink_table(
     id int,
     username text,
     score int
   );

Langkah 2: Buat pekerjaan Flink

1. Masuk ke konsol Realtime Compute. Pada tab Fully Managed Flink, temukan ruang kerja target dan klik Console di kolom Tindakan.

2. Di panel navigasi sebelah kiri, pilih Development > ETL.

3. Di bilah menu atas, klik . Pilih New Blank Stream Draft, lalu atur parameter berikut.

   Parameter pekerjaan	Deskripsi	Contoh
   Name	Nama pekerjaan. Catatan Nama pekerjaan harus unik dalam proyek saat ini.	adbpg-test
   Location	Folder tempat file kode pekerjaan disimpan. Anda juga dapat mengklik ikon di samping folder yang sudah ada untuk membuat subfolder.	Job Drafts
   Engine Version	Versi mesin Flink yang digunakan oleh pekerjaan. Untuk informasi selengkapnya tentang nomor versi mesin, pemetaan versi, dan tonggak siklus hidup, lihat Flink engine versions.	vvr-6.0.7-flink-1.15

4. Klik Create.

Langkah 3: Tulis kode pekerjaan dan sebarkan pekerjaan

1. Buat sumber bernama datagen_source untuk menghasilkan data analog dan sumber bernama source_adbpg untuk menangkap perubahan data real-time dari database AnalyticDB for PostgreSQL. Gabungkan kedua sumber tersebut dan tulis hasilnya ke tabel sink bernama sink_adbpg. Data yang diproses akan ditulis ke AnalyticDB for PostgreSQL.

   Salin kode pekerjaan berikut ke editor.

   ---Buat tabel sumber Datagen untuk menghasilkan data streaming menggunakan konektor Datagen.
   CREATE TEMPORARY TABLE datagen_source (
    id INT,
    score INT
   ) WITH (
    'connector' = 'datagen',
    'fields.id.kind'='sequence',
    'fields.id.start'='1',
    'fields.id.end'='100',
    'fields.score.kind'='random',
    'fields.score.min'='70',
    'fields.score.max'='100'
   );
   
   --Buat tabel sumber adbpg untuk menangkap perubahan data tabel adbpg_source_table berdasarkan slot.name dan pgoutput menggunakan konektor adbpg-cdc.
   CREATE TEMPORARY TABLE source_adbpg(
    id int,
    username varchar,
    PRIMARY KEY(id) NOT ENFORCED
   ) WITH(
     'connector' = 'adbpg-cdc', 
     'hostname' = 'gp-bp16v8cgx46ns****-master.gpdb.rds.aliyuncs.com',
     'port' = '5432',
     'username' = 'account****',
     'password' = 'password****',
     'database-name' = 'testdb',
     'schema-name' = 'testschema',
     'table-name' = 'adbpg_source_table',
     'slot.name' = 'flink',
     'decoding.plugin.name' = 'pgoutput'
   );
   
   --Buat tabel sink adbpg untuk menulis hasil yang diproses ke tabel tujuan adbpg_sink_table di database.
   CREATE TEMPORARY TABLE sink_adbpg (
     id int,
     username varchar,
     score int
   ) WITH (
     'connector' = 'adbpg', 
     'url' = 'jdbc:postgresql://gp-bp16v8cgx46ns****-master.gpdb.rds.aliyuncs.com:5432/testdb',
     'tablename' = 'testschema.adbpg_sink_table',  
     'username' = 'account****',
     'password' = 'password****',
     'maxRetryTimes' = '2',
     'batchsize' = '5000',
     'conflictMode' = 'ignore',
     'writeMode' = 'insert',
     'retryWaitTime' = '200'
   );
   
   -- Tulis hasil penggabungan tabel datagen_source dan source_adbpg ke tabel sink adbpg.
   INSERT INTO sink_adbpg
   SELECT ts.id,ts.username,ds.score FROM datagen_source AS ds
   JOIN source_adbpg AS ts ON ds.id = ts.id;

   Parameter

   Parameter	Wajib	Tipe data	Deskripsi
   connector	Ya	STRING	Jenis konektor. Atur nilai ke adbpg-cdc untuk tabel sumber dan adbpg untuk tabel sink.
   hostname	Ya	STRING	Titik akhir internal instans AnalyticDB for PostgreSQL. Anda dapat memperoleh titik akhir internal pada halaman Basic Information instans.
   username	Ya	STRING	Akun database dan kata sandi instans AnalyticDB for PostgreSQL.
   password	Ya	STRING
   database-name	Ya	STRING	Nama database.
   schema-name	Ya	STRING	Nama skema. Parameter ini mendukung ekspresi reguler. Anda dapat berlangganan beberapa skema sekaligus.
   table-name	Ya	STRING	Nama tabel. Parameter ini mendukung ekspresi reguler. Anda dapat berlangganan beberapa tabel sekaligus.
   port	Ya	INTEGER	Port AnalyticDB for PostgreSQL. Nilainya tetap 5432.
   decoding.plugin.name	Ya	STRING	Nama plug-in decoding logis PostgreSQL. Nilainya tetap pgoutput.
   slot.name	Ya	STRING	Nama slot decoding logis. Untuk tabel sumber dalam pekerjaan Flink yang sama, gunakan nilai yang sama untuk slot.name. Jika pekerjaan Flink yang berbeda melibatkan tabel yang sama, atur slot.name yang unik untuk setiap pekerjaan. Hal ini mencegah error berikut: PSQLException: ERROR: replication slot "debezium" is active for PID 974.
   debezium.*	Tidak	STRING	Mengontrol perilaku klien Debezium dengan granularitas lebih halus. Misalnya, mengatur 'debezium.snapshot.mode' = 'never' menonaktifkan fitur snapshot. Untuk informasi selengkapnya, lihat properti konfigurasi.
   scan.incremental.snapshot.enabled	Tidak	BOOLEAN	Menentukan apakah akan mengaktifkan snapshot inkremental. Nilai yang valid: false (default): Snapshot inkremental dinonaktifkan. true: Snapshot inkremental diaktifkan.
   scan.startup.mode	Tidak	STRING	Mode startup untuk konsumsi data. Nilai yang valid: initial (default): Saat pekerjaan pertama kali dijalankan, semua data historis dipindai lalu membaca data write-ahead logging (WAL) terbaru. Hal ini memberikan transisi mulus antara data penuh dan inkremental. latest-offset: Saat pekerjaan pertama kali dijalankan, data historis tidak dipindai. Pembacaan dimulai dari akhir WAL, yaitu posisi log terbaru. Hanya perubahan data yang terjadi setelah konektor dimulai yang ditangkap. snapshot: Memindai semua data historis dan membaca entri WAL baru yang dihasilkan selama pemindaian penuh. Pekerjaan berhenti setelah pemindaian penuh selesai.
   changelog-mode	Tidak	STRING	Mode changelog untuk mengenkode perubahan aliran. Nilai yang valid: ALL (default): Mendukung semua jenis operasi, termasuk INSERT, DELETE, UPDATE_BEFORE, dan UPDATE_AFTER. UPSERT: Hanya mendukung operasi UPSERT, termasuk INSERT, DELETE, dan UPDATE_AFTER.
   heartbeat.interval.ms	Tidak	DURATION	Interval pengiriman paket heartbeat. Nilai default adalah 30 detik. Satuan dalam milidetik. Konektor CDC AnalyticDB for PostgreSQL mengirim paket heartbeat ke database untuk memastikan offset slot terus maju. Jika data tabel tidak sering berubah, atur parameter ini ke nilai yang wajar agar log WAL segera dibersihkan dan menghindari pemborosan disk space.
   scan.incremental.snapshot.chunk.key-column	Tidak	STRING	Menentukan kolom yang digunakan untuk chunking selama fase snapshot. Secara default, kolom pertama dari primary key dipilih.
   url	Ya	STRING	Formatnya adalah jdbc:postgresql://<Address>:<PortId>/<DatabaseName>.

2. Di bagian atas halaman pengembangan pekerjaan, klik Deep Check untuk melakukan pemeriksaan sintaksis.

3. Klik Deploy, lalu klik Confirm.

4. Di pojok kanan atas, klik Operations. Pada halaman Deployments, klik Start.

Langkah 4: Lihat data yang ditulis oleh Flink

1. Jalankan pernyataan berikut di database uji untuk melihat data yang ditulis oleh Flink.

   SELECT * FROM testschema.adbpg_sink_table;
   SELECT COUNT(*) FROM testschema.adbpg_sink_table; 

2. Masukkan 50 baris data tambahan ke dalam tabel sumber, lalu periksa jumlah total baris data inkremental yang ditulis Flink ke tabel sink.

   -- Masukkan 50 baris data inkremental ke dalam tabel sumber.
   INSERT INTO testschema.adbpg_source_table(id, username)
   SELECT i, 'username'||i::text
   FROM generate_series(51, 100) AS t(i);
   
   -- Periksa data baru di tabel tujuan.
   SELECT COUNT(*) FROM testschema.adbpg_sink_table where id > 50;

   Hasilnya ditampilkan di bawah ini.

    count 
   -------
       50
   (1 row)

Langkah 1: Siapkan tabel uji

Di instans AnalyticDB for PostgreSQL, buat dua tabel sumber.

CREATE TABLE products (
    product_id SERIAL PRIMARY KEY,
    product_name VARCHAR(200) NOT NULL,
    sku CHAR(12) NOT NULL,
    description TEXT,
    price NUMERIC(10,2) NOT NULL,
    discount_price DECIMAL(10,2),
    stock_quantity INTEGER DEFAULT 0,
    weight REAL,
    volume DOUBLE PRECISION,
    dimensions BOX,
    release_date DATE,
    is_featured BOOLEAN DEFAULT FALSE,
    rating FLOAT,
    warranty_period INTERVAL,
    metadata JSON,
    tags TEXT[]
);

CREATE TABLE documents (
    document_id UUID PRIMARY KEY,
    title VARCHAR(200) NOT NULL,
    content TEXT,
    summary TEXT,
    publication_date TIMESTAMP WITHOUT TIME ZONE,
    last_updated TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
    author_id BIGINT,
    file_data BYTEA,
    xml_content XML,
    json_metadata JSON,
    reading_time INTERVAL,
    is_public BOOLEAN DEFAULT TRUE,
    views_count INTEGER DEFAULT 0,
    category VARCHAR(50),
    tags TEXT[]
);

Langkah 2: Siapkan sumber daya Kafka

1. Beli dan sebarkan instans Kafka.

2. Tambahkan blok CIDR ruang kerja Flink ke daftar putih instans Kafka.

3. Buat sumber daya di instans Kafka.

Langkah 3: Buat pekerjaan Flink

1. Masuk ke konsol Realtime Compute. Pada tab Fully Managed Flink, temukan ruang kerja target dan klik Console di kolom Tindakan.

2. Di panel navigasi sebelah kiri, pilih Development > ETL.

3. Di bilah menu atas, klik . Pilih New Blank Stream Draft dan atur parameter pekerjaan.

4. Klik Create.

Langkah 4: Tulis kode pekerjaan dan sebarkan pekerjaan

1. Tulis pekerjaan SQL di ruang kerja Flink. Salin kode pekerjaan berikut ke editor dan ganti konfigurasinya dengan nilai aktual Anda.

   -- Gunakan satu sumber untuk menangkap data dari beberapa tabel
   CREATE TEMPORARY TABLE ADBPGSource(
       table_name STRING METADATA FROM 'table_name' VIRTUAL,
       row_kind STRING METADATA FROM 'row_kind' VIRTUAL,
       product_id BIGINT,
       product_name STRING,
       sku STRING,
       description STRING,
       price STRING,
       discount_price STRING,
       stock_quantity INT,
       weight STRING,
       volume STRING,
       dimensions STRING,
       release_date STRING,
       is_featured BOOLEAN,
       rating FLOAT,
       warranty_period STRING,
       metadata STRING,
       tags STRING,
       document_id STRING,
       title STRING,
       content STRING,
       summary STRING,
       publication_date STRING,
       last_updated STRING,
       author_id BIGINT,
       file_data STRING,
       xml_content STRING,
       json_metadata STRING,
       reading_time STRING,
       is_public BOOLEAN,
       views_count INT,
       category STRING
   ) WITH (
     'connector' = 'adbpg-cdc',
     'hostname' = 'gp-2zev887z58390***-master.gpdb.rds.aliyuncs.com',
     'port' = '5432',
     'username' = 'account****',
     'password' = 'password****',
     'database-name' = 'testdb',
     'schema-name' = 'public',
     'table-name' = '(products|documents)',
     'slot.name' = 'flink',
     'decoding.plugin.name' = 'pgoutput',
     'debezium.snapshot.mode' = 'never'
   );
   
   CREATE TEMPORARY TABLE KafkaProducts (
       product_id BIGINT,
       product_name STRING,
       sku STRING,
       description STRING,
       price STRING,
       discount_price STRING,
       stock_quantity INT,
       weight STRING,
       volume STRING,
       dimensions STRING,
       release_date STRING,
       is_featured BOOLEAN,
       rating FLOAT,
       warranty_period STRING,
       metadata STRING,
       tags STRING,
       PRIMARY KEY(product_id) NOT ENFORCED
   ) WITH (
     'connector' = 'upsert-kafka',
     'topic' = '****',
     'properties.bootstrap.servers' = 'alikafka-post-cn-****-1-vpc.alikafka.aliyuncs.com:9092',
     'key.format'='avro',
     'value.format'='avro'
   );
   
   CREATE TEMPORARY TABLE KafkaDocuments (
       document_id STRING,
       title STRING,
       content STRING,
       summary STRING,
       publication_date STRING,
       last_updated STRING,
       author_id BIGINT,
       file_data STRING,
       xml_content STRING,
       json_metadata STRING,
       reading_time STRING,
       is_public BOOLEAN,
       views_count INT,
       category STRING,
       tags STRING,
       PRIMARY KEY(document_id) NOT ENFORCED
   ) WITH (
     'connector' = 'upsert-kafka',
     'topic' = '****',
     'properties.bootstrap.servers' = 'alikafka-post-cn-****-1-vpc.alikafka.aliyuncs.com:9092',
     'key.format'='avro',
     'value.format'='avro'
   );
   
   -- Gunakan STATEMENT SET untuk membungkus beberapa pernyataan
   BEGIN STATEMENT SET;
   -- Gunakan METADATA table_name untuk mengarahkan data ke tabel tujuan
   INSERT INTO KafkaProducts
   SELECT product_id,product_name,sku,description,price,discount_price,stock_quantity,weight,volume,dimensions,release_date,is_featured,rating,warranty_period,metadata,tags
   FROM ADBPGSource
   WHERE table_name = 'products';
   
   INSERT INTO KafkaDocuments
   SELECT document_id,title,content,summary,publication_date,last_updated,author_id,file_data,xml_content,json_metadata,reading_time,is_public,views_count,category,tags
   FROM ADBPGSource
   WHERE table_name = 'documents';
   
   END;

   Perhatikan poin-poin berikut mengenai pekerjaan SQL ini:

   • Untuk tugas sinkronisasi multi-tabel, kami merekomendasikan agar Anda menggunakan satu tabel sumber untuk menangkap data dari beberapa tabel, seperti yang ditunjukkan dalam contoh SQL ini. Anda harus mendefinisikan semua kolom dari semua tabel sumber dalam tabel sumber ini. Jika nama kolom duplikat, simpan hanya satu. Saat menulis ke tabel tujuan, gunakan METADATA table_name untuk mengarahkan data ke tabel tertentu. Pendekatan ini hanya memerlukan pembuatan satu Replication Slot di AnalyticDB for PostgreSQL. Hal ini mengurangi penggunaan sumber daya database sumber, meningkatkan performa sinkronisasi, dan menyederhanakan pemeliharaan di masa depan.

   • Gunakan parameter table-name untuk menentukan beberapa tabel sumber. Sertakan nama tabel dalam tanda kurung dan pisahkan dengan garis vertikal (|), misalnya, (table1|table2|table3).

   • Mengatur debezium.snapshot.mode ke never berarti hanya data inkremental dari tabel sumber yang disinkronkan. Untuk menyinkronkan data penuh dan inkremental, ubah pengaturannya ke initial.

2. Di bagian atas halaman pengembangan pekerjaan, klik Deep Check untuk melakukan pemeriksaan sintaksis.

3. Klik Deploy, lalu klik OK.

4. Di pojok kanan atas, klik Operations. Pada halaman Deployments, klik Start.

Langkah 5: Masukkan data uji

Di instans AnalyticDB for PostgreSQL, perbarui data di kedua tabel sumber dan amati perubahan pesan di topik Kafka.

Anda dapat menggunakan pernyataan SQL berikut untuk memasukkan data uji:

INSERT INTO products (
    product_name, sku, description, price, discount_price, stock_quantity, weight, volume, dimensions, release_date, is_featured, rating, warranty_period, metadata, tags
) VALUES (
    'Test Product', 'Test-2025', 'Sebuah data produk uji', 299.99, 279.99, 150, 50.5, 120.75, '(10,20),(30,40)', '2023-05-01', TRUE, 4.8, INTERVAL '1 year', '{"brand": "TechCo", "model": "X1"}', '{"Test1", "Test2"}'
);