PolarDB:Flink CDC untuk PolarDB-O

Prasyarat

• Tentukan versi Flink-CDC

  Jika Anda menggunakan Realtime Compute for Apache Flink Alibaba Cloud, Anda harus menentukan versi Flink-CDC komunitas yang kompatibel dengan versi Ververica Runtime (VVR) Anda. Untuk informasi lebih lanjut, lihat Pemetaan Versi CDC dan VVR.

  Catatan

  Repositori kode Flink-CDC tersedia di Flink-CDC.

• Tentukan versi Debezium

  Dalam file pom.xml dari versi Flink-CDC yang sesuai, temukan properti debezium.version untuk menentukan versi Debezium.

  Catatan

  Repositori kode Debezium tersedia di Debezium.

• Tentukan versi PgJDBC

  Dalam file pom.xml dari versi Postgres-CDC yang sesuai, temukan dependensi org.postgresql untuk menentukan versi PgJDBC.

  Catatan

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

  • Untuk release-3.0 dan seterusnya, jalur file-nya adalah flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/pom.xml.

  • Repositori kode PgJDBC tersedia di PgJDBC.

Prosedur

Paket JAR berikut untuk konektor PolarDB-O Flink CDC dibangun menggunakan JDK 8 sesuai langkah-langkah di atas: flink-sql-connector-postgres-cdc-2.3-SNAPSHOT.jar.

Konektor sumber

Contoh ini menunjukkan cara menggunakan konektor PolarDB-O Flink CDC untuk menyinkronkan tabel shipments dari database flink_source ke tabel shipments_sink di database flink_sink pada kluster PolarDB for PostgreSQL (Compatible with Oracle) 2.0.

1. Prasyarat

   • Siapkan PolarDB for PostgreSQL (Compatible with Oracle)

     1. Di halaman pembelian kluster PolarDB, beli kluster PolarDB for PostgreSQL (Compatible with Oracle) 2.0.

     2. Buat akun istimewa.

     3. Lihat titik akhir utama kluster. Jika kluster PolarDB dan ruang kerja Realtime Compute for Apache Flink Anda berada di virtual private cloud (VPC) yang sama, Anda dapat menggunakan titik akhir pribadi. Jika tidak, Anda harus mengajukan dan menggunakan titik akhir publik.

     4. Konfigurasikan daftar putih alamat IP untuk kluster. Tambahkan alamat IP instans Flink ke daftar putih kluster PolarDB.

     5. Di Konsol, buat database sumber flink_source dan database tujuan flink_sink. Untuk detailnya, lihat Buat database.

     6. Jalankan pernyataan berikut untuk membuat tabel shipments di database flink_source dan memasukkan 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 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)
         );

   • Siapkan Realtime Compute for Apache Flink

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

        Catatan

        Kami merekomendasikan membuat ruang kerja Realtime Compute for Apache Flink di Region dan VPC yang sama dengan kluster PolarDB. Hal ini memungkinkan Anda menggunakan titik akhir utama pribadi kluster PolarDB untuk koneksi.

     2. Buat konektor kustom dan unggah paket PolarDB-O Flink CDC yang telah Anda bangun. Pilih debezium-json untuk Formats. Untuk detailnya, lihat Buat konektor kustom.

2. Buat pekerjaan Flink

   1. Login ke Konsol Realtime Compute for Apache Flink dan buat draft SQL. Untuk detailnya, lihat Kembangkan draft SQL. Gunakan kode Flink SQL berikut dan ganti placeholder untuk titik akhir utama, port, username, dan password kluster PolarDB Anda.

      Catatan

      Tipe DATE di PolarDB for PostgreSQL (Compatible with Oracle) berupa 64-bit, sedangkan tipe DATE di Flink SQL dan kebanyakan database berupa 32-bit. Oleh karena itu, Anda harus memetakan kolom DATE di tabel sumber ke tipe TIMESTAMP di tabel sumber dan sink 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.

      Di bilah alat atas editor pekerjaan, klik Deploy.

      Di panel navigasi sebelah kiri, pilih Operations Center > Deployments. Di daftar pekerjaan, temukan pekerjaan target dan klik Start di kolom Actions.

   3. Uji dan verifikasi hasilnya.

      • Setelah pekerjaan dideploy dan dalam status berjalan, data dari tabel shipments disinkronkan ke tabel shipments_sink di database flink_sink.

        SELECT * FROM public.shipments_sink;

        Hasil berikut dikembalikan:

         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 flink_source. Perubahan tersebut 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 disinkronkan ke tabel shipments_sink di database flink_sink.

        SELECT * FROM public.shipments_sink;

        Hasil berikut dikembalikan:

         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)

Konektor pipeline

Contoh ini menunjukkan cara menggunakan konektor pipeline PolarDB-O Flink CDC untuk menyinkronkan tabel shipments1 dan shipments2 dari kluster PolarDB for PostgreSQL (Compatible with Oracle) 2.0. Untuk debugging, sink menggunakan konektor Print. Di lingkungan produksi, pilih konektor sink yang sesuai berdasarkan kebutuhan bisnis Anda.

1. Prasyarat

   • Siapkan PolarDB for PostgreSQL (Compatible with Oracle)

     1. Di halaman pembelian kluster PolarDB, beli kluster PolarDB for PostgreSQL (Compatible with Oracle) 2.0.

     2. Buat akun istimewa.

     3. Lihat titik akhir utama kluster. Jika kluster PolarDB dan ruang kerja Realtime Compute for Apache Flink Anda berada di VPC yang sama, Anda dapat menggunakan titik akhir pribadi. Jika tidak, Anda harus mengajukan dan menggunakan titik akhir publik.

     4. Konfigurasikan daftar putih alamat IP untuk kluster. Tambahkan alamat IP instans Flink ke daftar putih kluster PolarDB.

     5. Di Konsol, buat database sumber flink_source. Untuk detailnya, lihat Buat database.

     6. Jalankan pernyataan berikut untuk membuat tabel shipments1 dan shipments2 di database flink_source dan memasukkan 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();

   • Siapkan Realtime Compute for Apache Flink

     Login ke Konsol Realtime Compute dan beli instans Realtime Compute for Apache Flink. Untuk informasi lebih lanjut, lihat Aktifkan Realtime Compute for Apache Flink.

     Catatan

     Kami merekomendasikan membuat ruang kerja Realtime Compute for Apache Flink di Region dan VPC yang sama dengan kluster PolarDB. Hal ini memungkinkan Anda menggunakan titik akhir utama pribadi kluster PolarDB untuk koneksi.

2. Buat pekerjaan Flink

   1. Login ke Konsol Realtime Compute for Apache Flink dan buat draft ingesti data. Untuk detailnya, lihat Ingesti data Flink CDC. Gunakan konfigurasi ingesti data berikut dan ganti placeholder untuk titik akhir utama, port, username, dan password kluster PolarDB Anda.

      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. Di bagian More di sebelah kiri, tambahkan konektor pipeline yang telah Anda bangun. Di panel More Configurations di sebelah kanan, pastikan versi engine adalah vvr-11.5-jdk11-flink-1.20 dan tambahkan file dependensi yang diperlukan di area Additional Dependency Files.

   3. Deploy dan mulai pekerjaan.

      1. Klik Deploy di pojok kanan atas.

         source:
           type: polardbo
           name: PolarDB Oracle Source
           hostname: xxx
           port: xxx
           username: xxx
           password: xxx
           tables: flink_source.public.shipments[12]
           decoding.plugin.name: pgoutput
           slot.name: pgtest
         sink:
           type: values
           name: values Sink
           print.enabled: true

      2. Buka halaman Deployments dan klik Enable.

   4. Uji dan verifikasi hasilnya.

      • Setelah pekerjaan deployment berhasil dijalankan, statusnya menjadi Running. Anda dapat menemukan CreateTableEvent dan DataChangeEvent dari fase data lengkap di log Job Log > Running Task Managers > Stdout. Di halaman detail deployment, klik tab Job Log, pilih Running Task Managers, lalu klik tab Stdout untuk melihat output log. Pastikan output berisi event pembuatan tabel dan event perubahan data berikut:

        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 pernyataan DML pada tabel shipments1 dan shipments2 di database flink_source. Perubahan tersebut 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 menemukan DataChangeEvent untuk fase 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=()}