PolarDB:Flink CDC kompatibel dengan PolarDB untuk PostgreSQL (Kompatibel dengan Oracle)

Prasyarat

• Tentukan versi Flink-CDC.

  Jika Anda menggunakan Realtime Compute for Apache Flink Alibaba Cloud, tentukan versi Flink-CDC komunitas yang kompatibel dengan versi Ververica Runtime (VVR) yang sesuai. Untuk detailnya, lihat Pemetaan versi CDC dan VVR.

  Catatan

  Untuk repositori kode Flink-CDC, lihat Flink-CDC.

• Tentukan versi Debezium.

  Dalam file pom.xml dari versi Flink-CDC yang sesuai, cari kata kunci debezium.version untuk menentukan versi Debezium.

  Catatan

  Untuk repositori kode Debezium, lihat Debezium.

• Tentukan versi PgJDBC.

  Dalam file pom.xml dari versi Postgres-CDC yang sesuai, cari kata kunci org.postgresql untuk menentukan versi PgJDBC.

  Catatan

  • Untuk versi sebelum release-3.0, jalur file adalah: flink-connector-postgres-cdc/pom.xml.

  • Untuk versi release-3.0 dan yang lebih baru, jalur file adalah: flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/pom.xml.

  • Untuk repositori kode PgJDBC, lihat PgJDBC.

Prosedur

Ikuti langkah-langkah di atas untuk mengemas JAR konektor PolarDB for PostgreSQL Flink CDC menggunakan JDK 8: flink-sql-connector-postgres-cdc-2.3-SNAPSHOT.jar.

Konektor sumber

Contoh berikut mengilustrasikan cara menyinkronkan tabel `shipments` dari database `flink_source` di kluster 2.0 ke tabel `shipments_sink` di database `flink_sink` menggunakan PolarDBO Flink CDC.

1. Prasyarat

   • Persiapan

     1. Beli kluster 2.0 di halaman pembelian kluster PolarDB.

     2. Buat akun istimewa.

     3. Lihat titik akhir utama kluster. Jika kluster PolarDB dan Realtime Compute for Apache Flink berada dalam virtual private cloud (VPC) yang sama, Anda dapat langsung menggunakan titik akhir pribadi. Jika tidak, mintalah titik akhir publik.

     4. Atur daftar putih kluster: Tambahkan alamat instans Flink ke daftar putih kluster PolarDB.

     5. Buat database sumber `flink_source` dan database target `flink_sink` di Konsol. Untuk langkah-langkah detail, lihat Buat database.

     6. Jalankan pernyataan berikut untuk membuat tabel `shipments` di database sumber `flink_source` dan masukkan data.

        CREATE TABLE public.shipments (
          shipment_id INT,
          order_id INT,
          origin TEXT,
          destination TEXT,
          is_arrived BOOLEAN,
          order_time DATE,
          PRIMARY KEY (shipment_id) 
        );
        ALTER TABLE public.shipments REPLICA IDENTITY FULL;
        INSERT INTO public.shipments SELECT 1, 1, 'test1', 'test1', false, now();

     7. Jalankan pernyataan berikut untuk membuat tabel `shipments_sink` di database target `flink_sink`.

        CREATE TABLE public.shipments_sink (
           shipment_id INT,
           order_id INT,
           origin TEXT,
           destination TEXT,
           is_arrived BOOLEAN,
           order_time TIMESTAMP,
           PRIMARY KEY (shipment_id)
         );

   • Realtime Compute for Apache Flink Persiapan

     1. Masuk ke Konsol Realtime Compute dan beli instans Realtime Compute for Apache Flink. Untuk informasi lebih lanjut, lihat Beli Realtime Compute for Apache Flink.

        Catatan

        Kami merekomendasikan agar Region dan Virtual Private Cloud (VPC) Realtime Compute for Apache Flink konsisten dengan kluster PolarDB. Anda dapat langsung menggunakan titik akhir pribadi dari titik akhir utama kluster PolarDB sebagai alamat koneksi.

     2. Buat konektor kustom dan unggah PolarDBO Flink CDC yang telah dikemas. Atur Formats ke debezium-json. Untuk langkah-langkah detail, lihat Buat konektor kustom.

        

2. Buat pekerjaan Flink

   1. Masuk ke Konsol Realtime Compute. Buat draf pekerjaan SQL baru. Untuk panduan, lihat peta pengembangan pekerjaan. Gunakan pernyataan Flink SQL berikut. Ganti titik akhir utama kluster PolarDB, port, username, dan password dengan nilai Anda.

      Catatan

      Tipe DATE di berukuran 64-bit. Sebaliknya, tipe DATE di Flink SQL dan sebagian besar database berukuran 32-bit. Oleh karena itu, kolom bertipe DATE di tabel sumber harus didefinisikan sebagai TIMESTAMP di kedua tabel sumber dan sink dalam Flink SQL. Jika tidak, pekerjaan akan gagal dengan error ketidakcocokan tipe, seperti: “java.time.DateTimeException: Invalid value for EpochDay (valid values -365243219162 - 365241780471): 1720891573000”.

      CREATE TEMPORARY TABLE shipments (
         shipment_id INT,
         order_id INT,
         origin STRING,
         destination STRING,
         is_arrived BOOLEAN,
         order_time TIMESTAMP,
         PRIMARY KEY (shipment_id) NOT ENFORCED
       ) WITH (
         'connector' = 'polardbo-cdc',
         'hostname' = '<yourHostname>',
         'port' = '<yourPort>',
         'username' = '<yourUserName>',
         'password' = '<yourPassWord>',
         'database-name' = 'flink_source',
         'schema-name' = 'public',
         'table-name' = 'shipments',
         'decoding.plugin.name' = 'pgoutput',
         'slot.name' = 'flink'
       );
      
      CREATE TEMPORARY TABLE shipments_sink (
         shipment_id INT,
         order_id INT,
         origin STRING,
         destination STRING,
         is_arrived BOOLEAN,
         order_time TIMESTAMP,
         PRIMARY KEY (shipment_id) NOT ENFORCED
       ) WITH (
        'connector' = 'jdbc',
        'url' = 'jdbc:postgresql://<yourHostname>:<yourPort>/flink_sink',
        'table-name' = 'shipments_sink',
        'username' = '<yourUserName>',
        'password' = '<yourPassWord>'
       );
      
      INSERT INTO shipments_sink SELECT * FROM shipments;

   2. Deploy dan mulai pekerjaan.

      

      

   3. Uji dan validasi.

      • Setelah pekerjaan berhasil dideploy dan statusnya berubah menjadi running, data dari tabel shipments disinkronkan ke tabel shipments_sink di database tujuan flink_sink.

        SELECT * FROM public.shipments_sink;

        Hasilnya adalah sebagai berikut:

         shipment_id | order_id | origin | destination | is_arrived |     order_time      
        -------------+----------+--------+-------------+------------+---------------------
                   1 |        1 | test1  | test1       | f          | 2024-09-18 05:45:08
        (1 row)

      • Jalankan pernyataan DML pada tabel shipments di database sumber flink_source. Insert dan update baru disinkronkan secara real time.

        INSERT INTO public.shipments SELECT 2, 2, 'test2', 'test2', false, now();
        UPDATE public.shipments SET is_arrived = true WHERE shipment_id = 1;
        DELETE FROM public.shipments WHERE shipment_id = 2;
        INSERT INTO public.shipments SELECT 3, 3, 'test3', 'test3', false, now();
        UPDATE public.shipments SET is_arrived = true WHERE shipment_id = 3;

        Data di tabel shipments kini disinkronkan ke tabel shipments_sink di database tujuan flink_sink.

        SELECT * FROM public.shipments_sink;

        Hasilnya adalah sebagai berikut:

         shipment_id | order_id | origin | destination | is_arrived |     order_time      
        -------------+----------+--------+-------------+------------+---------------------
                   1 |        1 | test1  | test1       | t          | 2024-09-18 05:45:08
                   3 |        3 | test3  | test3       | t          | 2024-09-18 07:33:23
        (2 rows)

Pipeline connector

Contoh berikut mengilustrasikan cara menyinkronkan tabel `shipments1` dan `shipments2` dari database `flink_source` di kluster 2.0 menggunakan konektor Pipeline PolarDBO Flink CDC. Saat debugging, gunakan konektor Print. Di lingkungan produksi, pilih konektor yang sesuai sesuai kebutuhan.

1. Prasyarat

   • Persiapan

     1. Beli kluster 2.0 di halaman pembelian kluster PolarDB.

     2. Buat akun istimewa.

     3. Lihat titik akhir utama kluster. Jika kluster PolarDB dan Realtime Compute for Apache Flink berada dalam virtual private cloud (VPC) yang sama, Anda dapat langsung menggunakan titik akhir pribadi. Jika tidak, mintalah titik akhir publik.

     4. Atur daftar putih kluster: Tambahkan alamat instans Flink ke daftar putih kluster PolarDB.

     5. Buat database sumber `flink_source` di Konsol. Untuk langkah-langkah detail, lihat Buat database.

     6. Jalankan pernyataan berikut untuk membuat tabel `shipments1` dan `shipments2` di database sumber `flink_source` dan masukkan data.

        CREATE TABLE public.shipments1 (
          shipment_id INT,
          order_id INT,
          origin TEXT,
          destination TEXT,
          is_arrived BOOLEAN,
          order_time DATE,
          PRIMARY KEY (shipment_id) 
        );
        ALTER TABLE public.shipments1 REPLICA IDENTITY FULL;
        INSERT INTO public.shipments1 SELECT 1, 1, 'test1', 'test1', false, now();
        
        CREATE TABLE public.shipments2 (
          shipment_id INT,
          order_id INT,
          origin TEXT,
          destination TEXT,
          is_arrived BOOLEAN,
          order_time DATE,
          PRIMARY KEY (shipment_id) 
        );
        ALTER TABLE public.shipments2 REPLICA IDENTITY FULL;
        INSERT INTO public.shipments2 SELECT 1, 1, 'test1', 'test1', false, now();

   • Realtime Compute for Apache Flink persiapan

     Masuk ke Konsol Realtime Compute dan beli instans Realtime Compute for Apache Flink. Untuk operasi detail, lihat Aktifkan Realtime Compute for Apache Flink.

     Catatan

     Kami merekomendasikan agar region dan virtual private cloud (VPC) Realtime Compute for Apache Flink sesuai dengan kluster PolarDB. Alamat koneksi dapat langsung menggunakan titik akhir pribadi dari titik akhir utama kluster PolarDB.

2. Buat pekerjaan Flink.

   1. Masuk ke Konsol Realtime Compute dan buat draf ingesti data baru. Lihat Panduan Cepat Pekerjaan Ingesti Data Flink CDC. Gunakan konfigurasi ingesti data berikut, ubah titik akhir utama kluster PolarDB, port, akun, dan password.

      source:
         type: polardbo
         name: PolarDB Oracle Source
         hostname: '<yourHostname>'
         port: '<yourPort>'
         username: '<yourUserName>'
         password: '<yourPassWord>'
         tables: flink_source.public.shipments[12]
         decoding.plugin.name:  pgoutput
         slot.name: pgtest
      
      sink:
        type: values
        name: values Sink
        print.enabled: true

   2. Tambahkan konektor Pipeline yang telah berhasil dikemas di bagian More Configurations di sebelah kiri.

   3. Deploy dan mulai pekerjaan.

      1. Klik Deploy di pojok kanan atas.

      2. Buka halaman O&M pekerjaan dan klik Start.

         

   4. Uji dan verifikasi.

      • Setelah pekerjaan berhasil dideploy dan berjalan, Anda dapat melihat `CreateTableEvent` dan `DataChangeEvent` untuk fase data lengkap di log Job logs > Running Task Managers > Stdout.

        CreateTableEvent{tableId=public.shipments2, schema=columns={`shipment_id` INT NOT NULL,`order_id` INT,`origin` STRING,`destination` STRING,`is_arrived` BOOLEAN,`order_time` TIMESTAMP(6)}, primaryKeys=shipment_id, options=()}
        CreateTableEvent{tableId=public.shipments1, schema=columns={`shipment_id` INT NOT NULL,`order_id` INT,`origin` STRING,`destination` STRING,`is_arrived` BOOLEAN,`order_time` TIMESTAMP(6)}, primaryKeys=shipment_id, options=()}
        DataChangeEvent{tableId=public.shipments2, before=[], after=[1, 1, test1, test1, false, 2026-01-07T16:30:44], op=INSERT, meta=()}
        DataChangeEvent{tableId=public.shipments1, before=[], after=[1, 1, test1, test1, false, 2026-01-07T16:30:44], op=INSERT, meta=()}

      • Jalankan DML pada tabel `shipments1` dan `shipments2` di database sumber `flink_source`. Penambahan dan modifikasi baru juga akan disinkronkan secara real time.

        INSERT INTO public.shipments1 SELECT 2, 2, 'test2', 'test2', false, now();
        UPDATE public.shipments1 SET is_arrived = true WHERE shipment_id = 1;
        DELETE FROM public.shipments1 WHERE shipment_id = 2;
        INSERT INTO public.shipments1 SELECT 3, 3, 'test3', 'test3', false, now();
        UPDATE public.shipments1 SET is_arrived = true WHERE shipment_id = 3;
        
        INSERT INTO public.shipments2 SELECT 2, 2, 'test2', 'test2', false, now();
        UPDATE public.shipments2 SET is_arrived = true WHERE shipment_id = 1;
        DELETE FROM public.shipments2 WHERE shipment_id = 2;
        INSERT INTO public.shipments2 SELECT 3, 3, 'test3', 'test3', false, now();
        UPDATE public.shipments2 SET is_arrived = true WHERE shipment_id = 3;

      • Anda dapat melihat `DataChangeEvent` untuk fase data inkremental di log Job logs > Running Task Managers > Stdout:

        DataChangeEvent{tableId=public.shipments1, before=[], after=[2, 2, test2, test2, false, 2026-01-07T16:44:50], op=INSERT, meta=()}
        DataChangeEvent{tableId=public.shipments1, before=[1, 1, test1, test1, false, 2026-01-07T16:30:44], after=[1, 1, test1, test1, true, 2026-01-07T16:30:44], op=UPDATE, meta=()}
        DataChangeEvent{tableId=public.shipments1, before=[2, 2, test2, test2, false, 2026-01-07T16:44:50], after=[], op=DELETE, meta=()}
        DataChangeEvent{tableId=public.shipments1, before=[], after=[3, 3, test3, test3, false, 2026-01-07T16:44:50], op=INSERT, meta=()}
        DataChangeEvent{tableId=public.shipments1, before=[3, 3, test3, test3, false, 2026-01-07T16:44:50], after=[3, 3, test3, test3, true, 2026-01-07T16:44:50], op=UPDATE, meta=()}
        DataChangeEvent{tableId=public.shipments2, before=[], after=[2, 2, test2, test2, false, 2026-01-07T16:44:50], op=INSERT, meta=()}
        DataChangeEvent{tableId=public.shipments2, before=[1, 1, test1, test1, false, 2026-01-07T16:30:44], after=[1, 1, test1, test1, true, 2026-01-07T16:30:44], op=UPDATE, meta=()}
        DataChangeEvent{tableId=public.shipments2, before=[2, 2, test2, test2, false, 2026-01-07T16:44:50], after=[], op=DELETE, meta=()}
        DataChangeEvent{tableId=public.shipments2, before=[], after=[3, 3, test3, test3, false, 2026-01-07T16:44:50], op=INSERT, meta=()}
        DataChangeEvent{tableId=public.shipments2, before=[3, 3, test3, test3, false, 2026-01-07T16:44:50], after=[3, 3, test3, test3, true, 2026-01-07T16:44:50], op=UPDATE, meta=()}