Realtime Compute for Apache Flink:Paimon connector

SQL

Gunakan Paimon connector dalam job SQL sebagai tabel sumber atau tabel sink.

Freshness dan konsistensi data

Tabel sink Paimon menggunakan protokol two-phase commit untuk meng-commit data selama setiap checkpoint job Flink. Oleh karena itu, freshness data ditentukan oleh interval checkpoint job Flink. Setiap commit menghasilkan hingga dua snapshot.

Saat dua job Flink menulis ke tabel Paimon yang sama secara konkuren, jika job menulis ke bucket yang berbeda, mereka mencapai konsistensi serializable. Jika job menulis ke bucket yang sama, mereka hanya mencapai isolasi snapshot. Artinya, data tabel mungkin merupakan campuran hasil dari kedua job, tetapi tidak terjadi kehilangan data.

Mesin Penggabungan

Saat tabel sink Paimon menerima beberapa record dengan kunci primer yang sama, tabel tersebut menggabungkannya menjadi satu record untuk menjaga keunikan. Anda dapat mengontrol perilaku ini dengan mengatur parameter merge-engine. Tabel berikut menjelaskan mesin penggabungan yang tersedia.

CREATE TABLE MyTable (
  product_id BIGINT,
  price DOUBLE,
  sales BIGINT,
  PRIMARY KEY (product_id) NOT ENFORCED
) WITH (
  'merge-engine' = 'aggregation',
  'fields.price.aggregate-function' = 'max',
  'fields.sales.aggregate-function' = 'sum'
);

Produsen changelog

Atur parameter changelog-producer untuk mengonfigurasi Paimon agar menghasilkan changelog lengkap (di mana setiap record update_after memiliki record update_before yang sesuai) untuk aliran input apa pun. Tabel berikut menjelaskan produsen changelog yang tersedia. Untuk detail lebih lanjut, lihat dokumentasi resmi Apache Paimon.

Mode penulisan

Tabel Paimon mendukung mode penulisan berikut.

Target untuk ingesti data Flink CDC

Tabel Paimon mendukung sinkronisasi real-time data dari tabel tunggal atau seluruh database. Perubahan skema upstream disinkronkan ke tabel Paimon secara real-time. Untuk informasi lebih lanjut, lihat bagian Ingesti Data.

Pemangkasan baca Variant

Hanya membaca bidang Variant yang direferensikan oleh kueri, mengurangi overhead I/O dan memori.

Cara mengaktifkan

Prasyarat

• Data Variant hanya dapat ditulis oleh VVR 11.6 atau yang lebih baru. Data Variant yang ditulis oleh versi sebelumnya atau mesin lain tidak mendukung pemangkasan baca.

• Agar data upstream berlaku, Anda harus mengaktifkan opsi variant.inferShreddingSchema untuk mengaktifkan inferensi shredding Variant otomatis, atau secara eksplisit mengonfigurasi variant.shreddingSchema dan opsi terkait untuk menentukan anggota Variant mana yang cocok untuk shredding. Untuk informasi lebih lanjut, lihat Coreoptions.

Skenario yang didukung

Pemangkasan baca berlaku saat SQL mengakses bidang Variant dengan kunci string, misalnya:

SELECT v['a']            FROM t;   -- Satu level
SELECT v['a']['b']       FROM t;   -- Bersarang
SELECT v['a'], v['b']    FROM t;   -- Beberapa bidang
SELECT id, v['a']        FROM t;   -- Dicampur dengan kolom biasa
SELECT v['a'] + 1        FROM t;   -- Direferensikan dalam ekspresi

Skenario yang tidak didukung

• Mereferensikan seluruh Variant secara langsung, seperti SELECT v FROM t.

• Menggabungkan referensi langsung dengan akses bidang, seperti SELECT v, v['a'] FROM t.

• Akses indeks array, seperti v[0].

• Skema tabel berisi bidang Row bersarang. Dalam kasus ini, pemangkasan baca tidak berlaku untuk bidang Variant apa pun di tabel.

URL yang ditandatangani Blob

Paimon dapat menghasilkan URL HTTPS GET publik berdurasi pendek yang ditandatangani untuk data kolom Blob yang disimpan di OSS. Layanan eksternal, seperti layanan model multimodal, dapat menggunakan URL tersebut untuk mengambil konten file secara langsung, tanpa mentransfer byte file di dalam job Flink.

Batasan

• Hanya mesin komputasi Flink VVR 11.9-preview1 dan yang lebih baru yang mendukung fitur ini.

• Hanya tabel Paimon yang disimpan di OSS yang didukung. Jenis metastore tidak menjadi masalah. Jenis metastore Filesystem, DLF, dan lainnya semuanya didukung.

• URL yang dihasilkan adalah URL HTTPS publik dan harus dikonsumsi oleh layanan eksternal yang memiliki akses jaringan publik. Jika catalog menggunakan titik akhir internal, kami menyarankan Anda mengubahnya ke titik akhir HTTPS publik agar layanan eksternal dapat mengakses URL yang dihasilkan.

• Untuk membaca kolom Blob biasa, Anda harus terlebih dahulu mendapatkan byte BlobDescriptor dengan menggunakan opsi blob-as-descriptor, lalu meneruskan byte tersebut ke fungsi.

Sintaks

descriptor_to_presigned_url(source_table, descriptor, validity)
try_descriptor_to_presigned_url(source_table, descriptor, validity)

Parameter

Nilai kembalian

Contoh

-- Gunakan petunjuk SQL untuk membaca kolom Blob sebagai descriptor dan menghasilkan URL yang ditandatangani
SELECT sys.descriptor_to_presigned_url(
    'default.image_table',
    image,
    INTERVAL '5' MINUTE)
FROM image_table /*+ OPTIONS('blob-as-descriptor'='true') */;

-- Mode toleransi kesalahan: mengembalikan NULL pada kesalahan tingkat baris, dengan perilaku lain tidak berubah
SELECT sys.try_descriptor_to_presigned_url(
    'default.image_table',
    image,
    INTERVAL '5' MINUTE)
FROM image_table /*+ OPTIONS('blob-as-descriptor'='true') */;

Tabel berantai

Tabel rantai merekonstruksi set data lengkap dari cabang snapshot dan perubahan yang mengikutinya di cabang delta. Tulis data lengkap periodik ke cabang snapshot dan perubahan antara dua muatan lengkap ke cabang delta. Pembacaan menggunakan partisi snapshot terbaru sebagai jangkar dan menggabungkan partisi delta berikutnya untuk membangun kembali data lengkap, sehingga Anda tidak perlu menulis ulang seluruh tabel. Ini berguna ketika hanya sebagian kecil baris yang berubah antar eksekusi, seperti dump binlog ODS. Untuk informasi lebih lanjut, lihat Chain Table dalam dokumentasi Apache Paimon. Tabel rantai hanya didukung di VVR 11.9 dan yang lebih baru.

Aktifkan tabel rantai

Atur chain-table.enabled ke true saat Anda membuat tabel, buat cabang snapshot dan delta, lalu konfigurasikan opsi fallback cabang pada tabel utama dan kedua tabel cabang.

CREATE TABLE `default`.`t` (
  `t1` STRING,
  `t2` STRING,
  `t3` STRING,
  `date` STRING,
  PRIMARY KEY (`date`, `t1`) NOT ENFORCED
) PARTITIONED BY (`date`) WITH (
  'chain-table.enabled' = 'true',
  'sequence.field' = 't2',
  'bucket-key' = 't1',
  'bucket' = '2',
  'partition.timestamp-pattern' = '$date',
  'partition.timestamp-formatter' = 'yyyyMMdd'
);

-- mycat adalah nama catalog Paimon.
CALL `mycat`.sys.create_branch('default.t', 'snapshot');
CALL `mycat`.sys.create_branch('default.t', 'delta');

ALTER TABLE `default`.`t` SET (
  'scan.fallback-snapshot-branch' = 'snapshot',
  'scan.fallback-delta-branch' = 'delta');

ALTER TABLE `default`.`t$branch_snapshot` SET (
  'scan.fallback-snapshot-branch' = 'snapshot',
  'scan.fallback-delta-branch' = 'delta');

ALTER TABLE `default`.`t$branch_delta` SET (
  'scan.fallback-snapshot-branch' = 'snapshot',
  'scan.fallback-delta-branch' = 'delta');

Menulis data

Tulis data lengkap ke t$branch_snapshot. Setiap partisi cabang ini adalah tampilan lengkap pada titik waktu tertentu. Tulis data inkremental ke t$branch_delta. Setiap partisi cabang ini hanya berisi perubahan periode tersebut.

-- Penulisan lengkap
INSERT OVERWRITE `default`.`t$branch_snapshot` PARTITION (date = '20250810') VALUES ('1', '1', '1');

-- Penulisan inkremental
INSERT INTO `default`.`t$branch_delta` PARTITION (date = '20250811') VALUES ('2', '1', '1');

Membaca data

-- Kueri lengkap
SELECT t1, t2, t3 FROM `default`.`t` WHERE date = '20250811';

-- Kueri inkremental
SELECT t1, t2, t3 FROM `default`.`t$branch_delta` WHERE date = '20250811';

-- Kueri hibrid
SELECT t1, t2, t3 FROM `default`.`t` WHERE date = '20250811'
UNION ALL
SELECT t1, t2, t3 FROM `default`.`t$branch_delta` WHERE date = '20250811';

Baca streaming

Job baca streaming berjalan dalam dua fase:

• Fase pemuatan awal: secara default, job membaca partisi snapshot terbaru dari setiap grup dan partisi delta yang mengikutinya. Partisi snapshot yang lebih lama dikecualikan. Untuk menggabungkan data snapshot dan delta berdasarkan jangkar dalam fase ini dan menyelesaikan record DELETE lintas cabang, atur chain-table.streaming.merge-snapshot ke true. Pertukarannya adalah pemindaian yang lebih berat saat startup job.

• Fase inkremental: job terus-menerus membaca commit baru dari cabang delta.

SET 'execution.runtime-mode' = 'streaming';

INSERT INTO downstream_sink SELECT * FROM `default`.`t`;

Kunci partisi ganda

Jika tabel memiliki lebih dari satu kunci partisi, gunakan chain-table.chain-partition-keys untuk memisahkan dimensi rantai dari dimensi grup. Nilainya harus berupa akhiran berurutan dari kunci partisi. Bidang akhiran membentuk rantai yang diurutkan berdasarkan waktu, dan bidang partisi sebelumnya menjadi dimensi grup. Setiap grup nilai partisi grup mempertahankan rantai dan jangkarnya sendiri. Jika opsi ini tidak diatur, semua partisi termasuk dalam satu grup implisit.

CREATE TABLE `default`.`t` (
  `t1` STRING,
  `t2` STRING,
  `t3` STRING,
  `region` STRING,
  `date` STRING,
  PRIMARY KEY (`region`, `date`, `t1`) NOT ENFORCED
) PARTITIONED BY (`region`, `date`) WITH (
  'chain-table.enabled' = 'true',
  'sequence.field' = 't2',
  'bucket-key' = 't1',
  'bucket' = '2',
  'partition.timestamp-pattern' = '$date',
  'partition.timestamp-formatter' = 'yyyyMMdd',
  -- Hanya date yang membentuk rantai. Setiap region memiliki rantainya sendiri.
  'chain-table.chain-partition-keys' = 'date'
);

Batasan

• Tabel rantai hanya didukung untuk tabel kunci primer. Anda harus menentukan bucket dan bucket-key.

• Skema setiap cabang harus konsisten.

• Tabel rantai memerlukan sequence.field. Mesin penggabungan first-row dan aggregation tidak didukung.

• Baca streaming dan join lookup memerlukan cabang delta untuk menggunakan mesin penggabungan deduplicate (nilai default). Baca batch tidak terpengaruh.

• Vektor penghapusan hanya didukung untuk tabel rantai yang menggunakan mesin penggabungan deduplicate.

• Hanya none (nilai default) dan input yang didukung untuk changelog-producer. Mode lookup dan full-compaction tidak didukung. Jika nilainya none, record dinormalisasi oleh kunci primer lengkap yang mencakup partisi rantai, dan record -D dan -U yang record aslinya berada di partisi lain dibuang. Atur nilainya ke input jika konsumen downstream harus menerima record changelog lintas partisi.

• Baca streaming dan join lookup hanya mendukung mode startup default (latest-full). Error dilempar jika Anda menentukan posisi awal seperti scan.snapshot-id, scan.timestamp-millis, scan.mode diatur ke latest, atau consumer-id. Untuk melakukan baca streaming standar tanpa logika tabel rantai, baca tabel cabang tertentu seperti t$branch_delta.

• Filter partisi tidak didukung dalam baca streaming dan join lookup tabel rantai. Ini termasuk kolom partisi dalam klausa WHERE dan opsi scan.partitions, karena filter mengganggu logika penggabungan lintas cabang. Gunakan mode batch untuk membaca partisi tertentu.

• Kunci join dari join lookup tidak boleh berisi kunci partisi. Jika tidak, error dikembalikan.

• partition.timestamp-formatter harus menentukan tanggal lengkap (tahun-bulan-hari atau tahun-hari-dalam-tahun), hanya menggunakan bidang tanggal dan waktu yang dapat dibalik (y/u, M/L, d, D, H, k, m, dan s), dan memiliki langkah minimum setidaknya satu detik.

Format file Mosaic

Mosaic adalah format hibrida kolom-bucket yang dioptimalkan untuk tabel lebar yang berisi ribuan atau puluhan ribu kolom. Kolom diurutkan berdasarkan nama dan didistribusikan ke bucket berdasarkan rentang, disimpan dalam format berorientasi kolom di setiap bucket, dan dikompresi secara independen menggunakan ZSTD. Projection pushdown bekerja pada granularitas bucket, sehingga kueri hanya membaca bucket yang berisi kolom yang diminta. Kolom dengan awalan nama serupa berada di bucket yang sama, yang meningkatkan rasio kompresi dan lokalitas proyeksi. Untuk informasi lebih lanjut, lihat Mosaic. Format Mosaic hanya didukung di VVR 11.9 dan yang lebih baru.

Mengaktifkan format

Atur file.format ke mosaic saat Anda membuat tabel. Dalam job ingesti data, opsi yang sesuai adalah table.properties.file.format.

CREATE TABLE `default`.`wide_table` (
  `id` BIGINT,
  `col_a` STRING,
  `col_b` DOUBLE
) WITH (
  'file.format' = 'mosaic'
);

Opsi format

Batasan

Mosaic tidak mendukung tipe kompleks: ARRAY, MAP, MULTISET, ROW, VARIANT, BLOB, dan VECTOR.

Menggunakan Paimon sebagai tabel dimensi

Tabel Paimon dapat digunakan sebagai tabel dimensi. Untuk sintaks JOIN, lihat Pernyataan JOIN tabel dimensi.

Secara default, lookup memuat semua data di setiap instans paralel. Pendekatan ini hanya cocok untuk tabel dimensi kecil. Untuk tabel dimensi besar, gunakan solusi Shuffle Lookup yang dijelaskan di bawah.

Tabel dimensi terpartisi

Jika tabel dimensi Anda terpartisi dan Anda hanya memerlukan data dari satu atau dua partisi terbaru, Anda dapat menggunakan fitur pemuatan partisi dinamis:

SELECT * FROM T
JOIN DIM /*+ OPTIONS('lookup.dynamic-partition'='max_pt()', 'lookup.dynamic-partition.refresh-interval'='1 h') */
FOR SYSTEM_TIME AS OF T.proc_time AS D
ON T.col = D.col;

Tabel dimensi besar: tabel bucket tetap

Hanya didukung di VVR 8.0.8 dan yang lebih baru. Untuk tabel bucket tetap (bucket > 0), Anda dapat menggunakan Shuffle Lookup untuk mendistribusikan data berdasarkan kunci bucket di instans paralel, sehingga setiap instans hanya memuat data di bucket yang ditugaskan:

SELECT /*+ LOOKUP('table'='D', 'shuffle'='true') */ T.col1, D.col2
FROM T
JOIN DIM FOR SYSTEM_TIME AS OF T.proc_time AS D
ON T.col1 = D.col1;

Tabel dimensi besar: tabel bucket tidak tetap

Hanya didukung di VVR 8.0.10 dan yang lebih baru. Untuk tabel bucket dinamis atau tabel append, Anda dapat menggunakan SHUFFLE_HASH atau REPLICATED_SHUFFLE_HASH sehingga setiap instans paralel membaca semua data tetapi hanya menyimpan bagian yang dibutuhkan:

-- Shuffle Hash
SELECT /*+ SHUFFLE_HASH(D) */ T.col1, D.col2
FROM T
JOIN DIM FOR SYSTEM_TIME AS OF T.proc_time AS D
ON T.col1 = D.col1;

-- Replicated Shuffle Hash
SELECT /*+ REPLICATED_SHUFFLE_HASH(D) */ T.col1, D.col2
FROM T
JOIN DIM FOR SYSTEM_TIME AS OF T.proc_time AS D
ON T.col1 = D.col1;

Untuk informasi lebih lanjut tentang SHUFFLE_HASH dan REPLICATED_SHUFFLE_HASH, lihat Pernyataan JOIN tabel dimensi.

Ingesti Data

Anda dapat menggunakan Paimon connector sebagai sink dalam job YAML ingesti data.

Sintaks

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: filesystem
  catalog.properties.warehouse: /path/warehouse

Parameter

Menggunakan kembali catalog yang ada

Mulai dari Realtime Compute for Apache Flink 11.5, Anda dapat langsung mereferensikan catalog Paimon bawaan dari halaman Data Management dalam job ingesti data Flink CDC. Hal ini mengurangi konfigurasi manual.

sink:
  type: paimon
  using.built-in-catalog: paimon_dlf_catalog
  catalog.properties.fs.oss.endpoint: oss-cn-beijing-internal.aliyuncs.com

Job ingesti data dapat secara otomatis menggunakan kembali semua parameter di catalog Paimon. Hal ini setara dengan mengonfigurasi secara manual parameter yang diawali dengan catalog.properties. dalam job YAML.

Untuk mengganti parameter yang digunakan kembali secara otomatis, atur secara eksplisit di job YAML. Konfigurasi YAML eksplisit memiliki prioritas lebih tinggi. Sebagai contoh, dalam sampel di atas, parameter fs.oss.endpoint menggunakan nilai dari job YAML, mengganti nilai di paimon_dlf_catalog.

Perubahan Skema

Job pipeline YAML CDC dapat menggunakan strategi berbeda untuk menangani perubahan skema. Untuk mengonfigurasi strategi, atur parameter tingkat pipeline schema.change.behavior seperti yang dijelaskan di Konfigurasi evolusi skema. schema.change.behavior dapat diatur ke IGNORE, LENIENT, TRY_EVOLVE, EVOLVE, atau EXCEPTION. Bagian berikut menjelaskan bagaimana mode LENIENT dan EVOLVE, yang melibatkan perubahan skema, menangani berbagai event perubahan skema.

Contoh

Contoh berikut menunjukkan konfigurasi dalam skenario khas.

source:
  type: mysql
  name: MySQL Source
  hostname: ${secret_values.mysql.hostname}
  port: ${mysql.port}
  username: ${secret_values.mysql.username}
  password: ${secret_values.mysql.password}
  tables: ${mysql.source.table}
  server-id: 8601-8604
  #(Opsional) Sinkronkan data dari tabel yang baru ditambahkan selama fase inkremental.
  scan.binlog.newly-added-table.enabled: true
  #(Opsional) Sinkronkan komentar tabel dan bidang.
  include-comments.enabled: true
  #(Opsional) Utamakan distribusi shard tak terbatas untuk mencegah potensi masalah Out Of Memory (OOM) pada TaskManager.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #(Opsional) Aktifkan parsing dan filtering untuk mempercepat pembacaan data.
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  #(Opsional) Tentukan pengguna commit. Kami menyarankan Anda menggunakan pengguna commit berbeda untuk setiap job untuk menghindari konflik.
  commit.user: your_job_name
  #(Opsional) Aktifkan vektor penghapusan untuk meningkatkan performa baca.
  table.properties.deletion-vectors.enabled: true

source:
  type: mysql
  name: MySQL Source
  hostname: ${secret_values.mysql.hostname}
  port: ${mysql.port}
  username: ${secret_values.mysql.username}
  password: ${secret_values.mysql.password}
  tables: ${mysql.source.table}
  server-id: 8601-8604
  #(Opsional) Sinkronkan data dari tabel yang baru ditambahkan selama fase inkremental.
  scan.binlog.newly-added-table.enabled: true
  #(Opsional) Sinkronkan komentar tabel dan bidang.
  include-comments.enabled: true
  #(Opsional) Utamakan distribusi shard tak terbatas untuk mencegah potensi masalah Out Of Memory (OOM) pada TaskManager.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #(Opsional) Aktifkan parsing dan filtering untuk mempercepat pembacaan data.
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: filesystem
  catalog.properties.warehouse: oss://default/test
  catalog.properties.fs.oss.endpoint: oss-cn-beijing-internal.aliyuncs.com
  catalog.properties.fs.oss.accessKeyId: xxxxxxxx
  catalog.properties.fs.oss.accessKeySecret: xxxxxxxx
  #(Opsional) Tentukan pengguna commit. Kami menyarankan Anda menggunakan pengguna commit berbeda untuk setiap job untuk menghindari konflik.
  commit.user: your_job_name
  #(Opsional) Aktifkan vektor penghapusan untuk meningkatkan performa baca.
  table.properties.deletion-vectors.enabled: true

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.test_source_table
  server-id: 5401-5499
  #(Opsional) Sinkronkan data dari tabel yang baru ditambahkan selama fase inkremental.
  scan.binlog.newly-added-table.enabled: true
  #(Opsional) Sinkronkan komentar tabel dan bidang.
  include-comments.enabled: true
  #(Opsional) Utamakan distribusi shard tak terbatas untuk mencegah potensi masalah Out Of Memory (OOM) pada TaskManager.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #(Opsional) Aktifkan parsing dan filtering untuk mempercepat pembacaan data.
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #(Opsional) Tentukan pengguna commit. Kami menyarankan Anda menggunakan pengguna commit berbeda untuk setiap job untuk menghindari konflik.
  commit.user: your_job_name
  #(Opsional) Aktifkan vektor penghapusan untuk meningkatkan performa baca.
  table.properties.deletion-vectors.enabled: true
 
transform:
  - source-table: test_db.test_source_table
    projection: \*, DATE_FORMAT(CAST(create_time AS TIMESTAMP), 'yyyy-MM-dd') as partition_key
    primary-keys: id, create_time, partition_key
    partition-keys: partition_key
    description: add partition key 

pipeline:
  name: MySQL to Paimon Pipeline

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.test_source_table
  server-id: 5401-5499
  #(Opsional) Sinkronkan komentar tabel dan bidang.
  include-comments.enabled: true
  #(Opsional) Utamakan distribusi shard tak terbatas untuk mencegah potensi masalah Out Of Memory (OOM) pada TaskManager.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #(Opsional) Aktifkan parsing dan filtering untuk mempercepat pembacaan data.
  scan.only.deserialize.captured.tables.changelog.enabled: true


sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #(Opsional) Tentukan pengguna commit. Kami menyarankan Anda menggunakan pengguna commit berbeda untuk setiap job untuk menghindari konflik.
  commit.user: your_job_name
  #(Opsional) Aktifkan vektor penghapusan untuk meningkatkan performa baca.
  table.properties.deletion-vectors.enabled: true

pipeline:
  name: MySQL to Paimon Pipeline

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  server-id: 5401-5499
  #(Opsional) Sinkronkan data dari tabel yang baru ditambahkan selama fase inkremental.
  scan.binlog.newly-added-table.enabled: true
  #(Opsional) Sinkronkan komentar tabel dan bidang.
  include-comments.enabled: true
  #(Opsional) Utamakan distribusi shard tak terbatas untuk mencegah potensi masalah Out Of Memory (OOM) pada TaskManager.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #(Opsional) Aktifkan parsing dan filtering untuk mempercepat pembacaan data.
  scan.only.deserialize.captured.tables.changelog.enabled: true


sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #(Opsional) Tentukan pengguna commit. Kami menyarankan Anda menggunakan pengguna commit berbeda untuk setiap job untuk menghindari konflik.
  commit.user: your_job_name
  #(Opsional) Aktifkan vektor penghapusan untuk meningkatkan performa baca.
  table.properties.deletion-vectors.enabled: true

pipeline:
  name: MySQL to Paimon Pipeline

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  server-id: 5401-5499
  #(Opsional) Sinkronkan data dari tabel yang baru ditambahkan selama fase inkremental.
  scan.binlog.newly-added-table.enabled: true
  #(Opsional) Sinkronkan komentar tabel dan bidang.
  include-comments.enabled: true
  #(Opsional) Utamakan distribusi shard tak terbatas untuk mencegah potensi masalah Out Of Memory (OOM) pada TaskManager.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #(Opsional) Aktifkan parsing dan filtering untuk mempercepat pembacaan data.
  scan.only.deserialize.captured.tables.changelog.enabled: true


sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #(Opsional) Tentukan pengguna commit. Kami menyarankan Anda menggunakan pengguna commit berbeda untuk setiap job untuk menghindari konflik.
  commit.user: your_job_name
  #(Opsional) Aktifkan vektor penghapusan untuk meningkatkan performa baca.
  table.properties.deletion-vectors.enabled: true
  
route:
  # Sinkronkan semua tabel di database test_db MySQL ke database test_db2 Paimon dan pertahankan nama tabelnya.
  - source-table: test_db.\.*
    sink-table: test_db2.<>
    replace-symbol: <>

pipeline:
  name: MySQL to Paimon Pipeline

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.user\.*
  server-id: 5401-5499
  #(Opsional) Sinkronkan data dari tabel yang baru ditambahkan selama fase inkremental.
  scan.binlog.newly-added-table.enabled: true
  #(Opsional) Sinkronkan komentar tabel dan bidang.
  include-comments.enabled: true
  #(Opsional) Utamakan distribusi shard tak terbatas untuk mencegah potensi masalah Out Of Memory (OOM) pada TaskManager.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #(Opsional) Aktifkan parsing dan filtering untuk mempercepat pembacaan data.
  scan.only.deserialize.captured.tables.changelog.enabled: true


sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #(Opsional) Tentukan pengguna commit. Kami menyarankan Anda menggunakan pengguna commit berbeda untuk setiap job untuk menghindari konflik.
  commit.user: your_job_name
  #(Opsional) Aktifkan vektor penghapusan untuk meningkatkan performa baca.
  table.properties.deletion-vectors.enabled: true
  
route:
  # Semua tabel ter-shard di database test_db MySQL digabungkan menjadi satu tabel Paimon test_db.user.
  - source-table: test_db.user\.*
    sink-table: test_db.user

pipeline:
  name: MySQL to Paimon Pipeline

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  server-id: 5401-5499
  scan.startup.mode: initial
  # Saat restart job, periksa tabel baru yang ditangkap oleh parameter tables dan ambil snapshot.
  # Catatan: Harus digunakan bersama dengan scan.startup.mode: initial.
  scan.newly-added-table.enabled: true
  #(Opsional) Sinkronkan data dari tabel yang baru ditambahkan selama fase inkremental.
  scan.binlog.newly-added-table.enabled: true
  #(Opsional) Sinkronkan komentar tabel dan bidang.
  include-comments.enabled: true
  #(Opsional) Utamakan distribusi shard tak terbatas untuk mencegah potensi masalah Out Of Memory (OOM) pada TaskManager.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #(Opsional) Aktifkan parsing dan filtering untuk mempercepat pembacaan data.
  scan.only.deserialize.captured.tables.changelog.enabled: true


sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #(Opsional) Tentukan pengguna commit. Kami menyarankan Anda menggunakan pengguna commit berbeda untuk setiap job untuk menghindari konflik.
  commit.user: your_job_name
  #(Opsional) Aktifkan vektor penghapusan untuk meningkatkan performa baca.
  table.properties.deletion-vectors.enabled: true

pipeline:
  name: MySQL to Paimon Pipeline

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  # Tabel yang cocok dengan ekspresi reguler ini tidak disinkronkan.
  tables.exclude: test_db.table1
  server-id: 5401-5499
  #(Opsional) Sinkronkan data dari tabel yang baru ditambahkan selama fase inkremental.
  scan.binlog.newly-added-table.enabled: true
  #(Opsional) Sinkronkan komentar tabel dan bidang.
  include-comments.enabled: true
  #(Opsional) Utamakan distribusi shard tak terbatas untuk mencegah potensi masalah Out Of Memory (OOM) pada TaskManager.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #(Opsional) Aktifkan parsing dan filtering untuk mempercepat pembacaan data.
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #(Opsional) Tentukan pengguna commit. Kami menyarankan Anda menggunakan pengguna commit berbeda untuk setiap job untuk menghindari konflik.
  commit.user: your_job_name
  #(Opsional) Aktifkan vektor penghapusan untuk meningkatkan performa baca.
  table.properties.deletion-vectors.enabled: true

pipeline:
  name: MySQL to Paimon Pipeline

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  server-id: 5401-5499
  #(Opsional) Sinkronkan data dari tabel yang baru ditambahkan selama fase inkremental.
  scan.binlog.newly-added-table.enabled: true
  #(Opsional) Sinkronkan komentar tabel dan bidang.
  include-comments.enabled: true
  #(Opsional) Utamakan distribusi shard tak terbatas untuk mencegah potensi masalah Out Of Memory (OOM) pada TaskManager.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #(Opsional) Aktifkan parsing dan filtering untuk mempercepat pembacaan data.
  scan.only.deserialize.captured.tables.changelog.enabled: true


sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #(Opsional) Tentukan pengguna commit. Kami menyarankan Anda menggunakan pengguna commit berbeda untuk setiap job untuk menghindari konflik.
  commit.user: your_job_name
  #(Opsional) Aktifkan vektor penghapusan untuk meningkatkan performa baca.
  table.properties.deletion-vectors.enabled: true
  # Tentukan format file file output sebagai lance 
  table.properties.file.format: lance
pipeline:
  name: MySQL to Paimon Pipeline

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.test_source_table
  server-id: 5401-5499
  #(Opsional) Sinkronkan data dari tabel yang baru ditambahkan selama fase inkremental.
  scan.binlog.newly-added-table.enabled: true
  #(Opsional) Sinkronkan komentar tabel dan bidang.
  include-comments.enabled: true
  #(Opsional) Utamakan distribusi shard tak terbatas untuk mencegah potensi masalah Out Of Memory (OOM) pada TaskManager.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #(Opsional) Aktifkan parsing dan filtering untuk mempercepat pembacaan data.
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #(Opsional) Tentukan pengguna commit. Kami menyarankan Anda menggunakan pengguna commit berbeda untuk setiap job untuk menghindari konflik.
  commit.user: your_job_name
  #(Opsional) Aktifkan vektor penghapusan untuk meningkatkan performa baca.
  table.properties.deletion-vectors.enabled: true

pipeline:
  name: MySQL to Paimon Pipeline
  schema.change.behavior: evolve