AnalyticDB:Use Flink CDC to subscribe to full and incremental data in real time (invitational preview)

Adım 1: Test tablosu ve test verilerini hazırlama

1. AnalyticDB for PostgreSQL konsoluna giriş yapın ve hedef örneği bulun. Örnek kimliğine tıklayın.

2. Basic Information sayfasının sağ alt köşesinde Log On to Database seçeneğine tıklayın.

3. Bir test veritabanı ve adbpg_source_table adlı bir kaynak tablo oluşturun. Ardından kaynak tabloya 50 satır veri ekleyin.

   -- Create a test database.
   CREATE DATABASE testdb;
   -- Switch to the testdb database and create a schema.
   CREATE SCHEMA testschema;
   -- Create a source table named adbpg_source_table.
   CREATE TABLE testschema.adbpg_source_table(
     id int,
     username text,
     PRIMARY KEY(id)
   );
   -- Insert 50 rows of data into the adbpg_source_table table.
   INSERT INTO testschema.adbpg_source_table(id, username)
   SELECT i, 'username'||i::text
   FROM generate_series(1, 50) AS t(i);

4. Flink'in sonuç verilerini yazması için adbpg_sink_table adlı bir havuz tablosu oluşturun.

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

Adım 2: Flink işi oluşturma

1. Realtime Compute konsoluna giriş yapın. Fully Managed Flink sekmesinde hedef çalışma alanını bulun ve Actions sütunundaki Console seçeneğine tıklayın.

2. Sol gezinme bölmesinde Development > ETL yolunu izleyin.

3. Üst menü çubuğunda simgesine tıklayın. New Blank Stream Draft seçeneğini belirleyin ve aşağıdaki parametreleri ayarlayın.

   İş parametresi	Açıklama	Örnek
   Name	İşin adı. Not İş adı mevcut proje içinde benzersiz olmalıdır.	adbpg-test
   Location	İşin kod dosyasının depolandığı klasör. Mevcut bir klasörün yanındaki simgesine tıklayarak da alt klasör oluşturabilirsiniz.	Job Drafts
   Engine Version	İş tarafından kullanılan Flink motor sürümü. Motor sürüm numaraları, sürüm eşlemeleri ve yaşam döngüsü kilometre taşları hakkında daha fazla bilgi için bkz. Flink motor sürümleri.	vvr-6.0.7-flink-1.15

4. Create seçeneğine tıklayın.

Adım 3: İş kodunu yazma ve işi dağıtma

1. Analog veri oluşturmak için datagen_source adlı bir kaynak ve AnalyticDB for PostgreSQL veritabanındaki gerçek zamanlı veri değişikliklerini yakalamak için source_adbpg adlı bir kaynak oluşturun. Ardından bu iki kaynağı birleştirin ve sonuçları sink_adbpg adlı bir havuz tablosuna yazın. İşlenen veriler AnalyticDB for PostgreSQL üzerine yazılır.

   Aşağıdaki iş kodunu düzenleyiciye kopyalayın.

   ---Create a Datagen source table to generate streaming data using the Datagen connector.
   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'
   );
   
   --Create an adbpg source table to capture data changes of the adbpg_source_table table based on slot.name and pgoutput using the adbpg-cdc connector.
   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'
   );
   
   --Create an adbpg sink table to write the processed results to the destination table adbpg_sink_table in the 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'
   );
   
   -- Write the join results of the datagen_source and source_adbpg tables to the adbpg sink table.
   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;

   Parametreler

   Parametre	Gerekli	Veri türü	Açıklama
   connector	Evet	STRING	Bağlayıcı türü. Kaynak tablo için değeri adbpg-cdc, havuz tablo için adbpg olarak ayarlayın.
   hostname	Evet	STRING	AnalyticDB for PostgreSQL örneğinin iç uç noktası. Örneğin Basic Information sayfasından iç uç noktayı edinebilirsiniz.
   username	Evet	STRING	AnalyticDB for PostgreSQL örneğinin veritabanı hesabı ve parolası.
   password	Evet	STRING
   database-name	Evet	STRING	Veritabanının adı.
   schema-name	Evet	STRING	Şemanın adı. Bu parametre normal ifadeleri destekler. Aynı anda birden çok şemaya abone olabilirsiniz.
   table-name	Evet	STRING	Tablonun adı. Bu parametre normal ifadeleri destekler. Aynı anda birden çok tabloya abone olabilirsiniz.
   port	Evet	INTEGER	AnalyticDB for PostgreSQL bağlantı noktası. Değer 5432 olarak sabittir.
   decoding.plugin.name	Evet	STRING	PostgreSQL mantıksal kod çözme eklentisinin adı. Değer pgoutput olarak sabittir.
   slot.name	Evet	STRING	Mantıksal kod çözme yuvasının adı. Aynı Flink işindeki kaynak tablolar için slot.name değerini aynı tutun. Farklı Flink işleri aynı tabloyu içeriyorsa her iş için benzersiz bir slot.name ayarlayın. Bu durum aşağıdaki hatayı önler: PSQLException: ERROR: replication slot "debezium" is active for PID 974.
   debezium.*	Hayır	STRING	Debezium istemcisini daha ince granülerlikte kontrol eder. Örneğin 'debezium.snapshot.mode' = 'never' ayarı anlık görüntü özelliğini devre dışı bırakır. Daha fazla bilgi için bkz. yapılandırma özellikleri.
   scan.incremental.snapshot.enabled	Hayır	BOOLEAN	Artımlı anlık görüntülerin etkinleştirilip etkinleştirilmeyeceğini belirtir. Geçerli değerler: false (varsayılan): Artımlı anlık görüntüler devre dışı bırakılır. true: Artımlı anlık görüntüler etkinleştirilir.
   scan.startup.mode	Hayır	STRING	Veri tüketimi için başlangıç modu. Geçerli değerler: initial (varsayılan): İş ilk kez başladığında tüm geçmiş verileri tarar ve ardından en son write-ahead logging (WAL) verilerini okur. Bu sayede tam ve artımlı veriler arasında kesintisiz bir geçiş sağlanır. latest-offset: İş ilk kez başladığında geçmiş verileri taramaz. WAL'ın sonundan, yani en son günlük konumundan okumaya başlar. Yalnızca bağlayıcı başladıktan sonra oluşan veri değişikliklerini yakalar. snapshot: Tüm geçmiş verileri tarar ve tam tarama sırasında oluşturulan yeni WAL girdilerini okur. Tam tarama tamamlandığında iş durur.
   changelog-mode	Hayır	STRING	Akış değişikliklerini kodlamak için değişiklik günlüğü modu. Geçerli değerler: ALL (varsayılan): INSERT, DELETE, UPDATE_BEFORE ve UPDATE_AFTER dahil olmak üzere tüm işlem türlerini destekler. UPSERT: Yalnızca UPSERT işlemlerini destekler. Bunlar INSERT, DELETE ve UPDATE_AFTER işlemlerini içerir.
   heartbeat.interval.ms	Hayır	DURATION	Heartbeat paketlerinin gönderilme aralığı. Varsayılan değer 30 saniyedir. Birim milisaniyedir. AnalyticDB for PostgreSQL CDC bağlayıcısı, yuva ofsetinin sürekli ilerlemesini sağlamak için veritabanına heartbeat paketleri gönderir. Tablo verileri sık değişmiyorsa WAL günlüklerini zamanında temizlemek ve disk alanı israfını önlemek için bu parametreyi makul bir değere ayarlayın.
   scan.incremental.snapshot.chunk.key-column	Hayır	STRING	Anlık görüntü aşamasında parçalama için kullanılacak bir sütun belirtir. Varsayılan olarak birincil anahtarın ilk sütunu seçilir.
   url	Evet	STRING	Format jdbc:postgresql://<Address>:<PortId>/<DatabaseName> şeklindedir.

2. İş geliştirme sayfasının üst kısmında sözdizimi denetimi gerçekleştirmek için Deep Check seçeneğine tıklayın.

3. Deploy seçeneğine tıklayın ve ardından Confirm seçeneğine tıklayın.

4. Sağ üst köşede Operations seçeneğine tıklayın. Deployments sayfasında Start seçeneğine tıklayın.

Adım 4: Flink tarafından yazılan verileri görüntüleme

1. Flink tarafından yazılan verileri görüntülemek için test veritabanında aşağıdaki deyimleri yürütün.

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

2. Kaynak tabloya 50 satır daha veri ekleyin. Ardından Flink'in havuz tablosuna yazdığı artımlı veri satırlarının toplam sayısını kontrol edin.

   -- Insert 50 rows of incremental data into the source table.
   INSERT INTO testschema.adbpg_source_table(id, username)
   SELECT i, 'username'||i::text
   FROM generate_series(51, 100) AS t(i);
   
   -- Check the new data in the destination table.
   SELECT COUNT(*) FROM testschema.adbpg_sink_table where id > 50;

   Sonuç aşağıda gösterilmektedir.

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

Adım 1: Test tablolarını hazırlama

AnalyticDB for PostgreSQL örneğinde iki kaynak tablo oluşturun.

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[]
);

Adım 2: Kafka kaynaklarını hazırlama

1. Bir Kafka örneği satın alın ve dağıtın.

2. Flink çalışma alanının CIDR bloğunu Kafka örneğinin izin listesine ekleyin.

3. Kafka örneğinde kaynaklar oluşturun.

Adım 3: Flink işi oluşturma

1. Realtime Compute konsoluna giriş yapın. Fully Managed Flink sekmesinde hedef çalışma alanını bulun ve Actions sütunundaki Console seçeneğine tıklayın.

2. Sol gezinme bölmesinde Development > ETL yolunu izleyin.

3. Üst menü çubuğunda simgesine tıklayın. New Blank Stream Draft seçeneğini belirleyin ve iş parametrelerini ayarlayın.

4. Create seçeneğine tıklayın.

Adım 4: İş kodunu yazma ve işi dağıtma

1. Flink çalışma alanında bir SQL işi yazın. Aşağıdaki iş kodunu düzenleyiciye kopyalayın ve yapılandırmaları gerçek değerlerinizle değiştirin.

   -- Use one source to capture data from multiple tables
   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'
   );
   
   -- Use a STATEMENT SET to wrap multiple statements
   BEGIN STATEMENT SET;
   -- Use the table_name METADATA to route data to the destination table
   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;

   Bu SQL işiyle ilgili aşağıdaki hususlara dikkat edin:

   • Çoklu tablo senkronizasyon görevleri için bu SQL örneğinde gösterildiği gibi birden çok tablodan veri yakalamak üzere tek bir kaynak tablo kullanmanızı öneririz. Bu kaynak tabloda tüm kaynak tabloların tüm sütunlarını tanımlamanız gerekir. Sütun adları yineleniyorsa yalnızca birini tutun. Hedef tabloya yazarken verileri belirtilen tabloya yönlendirmek için METADATA table_name kullanın. Bu yaklaşım AnalyticDB for PostgreSQL üzerinde yalnızca bir Çoğaltma Yuvası oluşturmayı gerektirir. Bu yöntem kaynak veritabanının kaynak kullanımını azaltır, senkronizasyon performansını iyileştirir ve gelecekteki bakımı kolaylaştırır.

   • Birden çok kaynak tablo belirtmek için table-name parametresini kullanın. Tablo adlarını parantez içine alın ve dikey çizgilerle (|) ayırın; örneğin (table1|table2|table3).

   • debezium.snapshot.mode ayarını never olarak yapmak kaynak tablodan yalnızca artımlı verilerin senkronize edilmesini sağlar. Hem tam hem de artımlı verileri senkronize etmek için ayarı initial olarak değiştirin.

2. İş geliştirme sayfasının üst kısmında sözdizimi denetimi gerçekleştirmek için Deep Check seçeneğine tıklayın.

3. Deploy seçeneğine tıklayın ve ardından OK seçeneğine tıklayın.

4. Sağ üst köşede Operations seçeneğine tıklayın. Deployments sayfasında Start seçeneğine tıklayın.

Adım 5: Test verisi ekleme

AnalyticDB for PostgreSQL örneğinde iki kaynak tablodaki verileri güncelleyin ve Kafka konusunda mesaj değişikliklerini gözlemleyin.

Test verisi eklemek için aşağıdaki SQL deyimini kullanabilirsiniz:

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', 'A piece of test product data', 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"}'
);