Parameter konektor - Realtime Compute for Apache Flink - Alibaba Cloud Documentation Center

Topik ini menjelaskan parameter untuk konektor Fluss.

Cara mengonfigurasi parameter

Konfigurasikan parameter saat pembuatan tabel

Fluss menyimpan parameter yang ditentukan saat pembuatan tabel sebagai metadata tabel. Parameter tersebut berlaku untuk semua operasi baca dan tulis.

# Atur table.log.ttl menjadi 7 hari dan nonaktifkan pemeriksaan CRC saat membaca
CREATE TABLE `my-catalog`.`my_db`.`my_log_table` (
  order_id BIGINT,
  item_id BIGINT,
  amount INT,
  address STRING
) WITH (
  'table.log.ttl' = '7d',
  'client.scanner.log.check-crc' = 'false'
);

Penting

Untuk tabel yang telah disimpan dalam metadata Fluss—yaitu tabel terdaftar yang dibuat melalui pernyataan DDL CREATE TABLE—Anda tidak dapat mengubah parameter konektor terdaftar seperti client.writer.batch-size dan client.writer.buffer.memory-size. Perubahan tersebut tidak akan berlaku meskipun Anda mengklik Edit di Konsol.

Konfigurasi dinamis dengan SQL Hints

Fluss juga memungkinkan Anda mengonfigurasi parameter secara dinamis menggunakan SQL Hints. Parameter ini tidak disimpan dalam metadata tabel dan hanya berlaku untuk kueri saat ini.

Penting

Parameter yang dikonfigurasi secara dinamis memiliki prioritas lebih tinggi daripada parameter yang disimpan dalam metadata tabel.
Parameter penyimpanan tidak mendukung konfigurasi dinamis menggunakan SQL Hints.

-- Atur SQL hints untuk operasi baca
SELECT * FROM `my-catalog`.`my_db`.`my_log_table` /*+ OPTIONS('client.scanner.log.check-crc' = 'false') */;

-- Atur SQL hints untuk operasi tulis
INSERT INTO `my-catalog`.`my_db`.`my_pk_table_2` /*+ OPTIONS('sink.ignore-delete'='true') */ select * from `my-catalog`.`my_db`.`my_pk_table_1`;

Parameter penyimpanan

Parameter	Tipe	Bawaan	Deskripsi
bucket.num	int	1	Jumlah bucket dalam kluster Fluss.
bucket.key	String	None	Data akan didistribusikan ke dalam bucket berdasarkan nilai hash dari `bucket.key`. `bucket.key` harus merupakan subset dari primary key dan tidak boleh mencakup partition key. Jika Anda menentukan beberapa bidang sebagai bucket.key, pisahkan bidang-bidang tersebut dengan koma `,` . Jika tabel memiliki primary key dan `bucket.key` tidak ditentukan secara eksplisit, primary key (tidak termasuk partition key) digunakan sebagai `bucket.key` secara bawaan. Jika primary key tidak didefinisikan untuk tabel dan `bucket.key` tidak ditentukan, data akan didistribusikan secara acak ke dalam bucket.
table.log.ttl	Duration	7d	Waktu hidup (TTL) untuk data log, yang mengontrol periode retensi maksimum. Setelah periode ini berakhir, Fluss menghapus data log lama untuk membebaskan ruang. Nilai -1 menonaktifkan kedaluwarsa, sehingga log tidak pernah dihapus.
table.auto-partition.enabled	Boolean	false	Jika diaktifkan, Fluss secara otomatis membuat partisi untuk tabel.
table.auto-partition.time-unit	ENUM	DAY	Granularitas waktu untuk pembuatan partisi otomatis. Nilai yang valid adalah `YEAR`, `QUARTER`, `MONTH`, `DAY`, atau `HOUR`. Contoh: Jika nilainya `HOUR`, format partisi yang dibuat secara otomatis adalah `yyyyMMddHH`. Jika nilainya `DAY`, formatnya adalah `yyyyMMdd`.
table.auto-partition.num-precreate	Integer	2	Jumlah partisi masa depan yang akan dibuat sebelumnya. Misalnya, jika tanggal saat ini adalah 2024-11-11 dan nilai ini diatur ke 3, Fluss membuat partisi untuk `20241111`, `20241112`, dan `20241113`, melewati partisi yang sudah ada. Nilai bawaannya adalah 2. Dengan `time-unit` `DAY`, nilai bawaan ini membuat partisi untuk hari ini dan besok.
table.auto-partition.num-retention	Integer	7	Jumlah partisi historis yang dipertahankan. Misalnya, jika tanggal saat ini adalah 2024-11-11, satuan waktu adalah `DAY`, dan nilai ini adalah 3, Fluss mempertahankan partisi untuk `20241108`, `20241109`, dan `20241110`. Fluss menghapus partisi yang lebih lama dari `20241108`. Nilai bawaannya adalah 7, artinya Fluss mempertahankan 7 partisi terbaru.
table.auto-partition.time-zone	String	Zona waktu sistem	Zona waktu untuk partisi otomatis. Secara bawaan, menggunakan zona waktu sistem.
table.replication.factor	Integer	(None)	Faktor replikasi untuk tabel log. Jika tidak diatur, Fluss menggunakan faktor replikasi bawaan kluster, `default.replication.factor`. Nilainya harus berupa bilangan bulat positif dan tidak boleh lebih besar dari jumlah TabletServer dalam kluster Fluss. Jika nilainya melebihi jumlah TabletServer, pembuatan tabel akan gagal.
table.log.format	Enum	ARROW	Format penyimpanan untuk log. Format yang didukung adalah `ARROW` (bawaan) dan `INDEXED`.
table.log.arrow.compression.type	Enum	ZSTD	Jika format log diatur ke ARROW, parameter ini menentukan tipe kompresi untuk data log. Tipe yang didukung adalah NONE, LZ4_FRAME, dan ZSTD. Nilai bawaannya adalah ZSTD.
table.log.arrow.compression.zstd.level	Integer	3	Jika format log diatur ke ARROW dan tipe kompresi adalah ZSTD, parameter ini menentukan tingkat kompresi. Rentang nilai yang valid adalah 1 hingga 22. Nilai bawaannya adalah 3.
table.statistics.columns	String	(None)	Menentukan daftar nama kolom yang dipisahkan koma untuk pengumpulan statistik, seperti `'temperature,location'`. Atur parameter ini ke `*` untuk mengumpulkan statistik untuk semua tipe kolom yang didukung. Statistik kolom digunakan untuk optimasi predicate pushdown dan hanya berlaku untuk tabel log yang menggunakan format log Arrow. Tipe data yang didukung meliputi: BOOLEAN, TINYINT, SMALLINT, INT, BIGINT, FLOAT, DOUBLE, DECIMAL, STRING, CHAR, DATE, TIME, TIMESTAMP, dan TIMESTAMP_LTZ. Tipe yang tidak didukung (BYTES, BINARY, ARRAY, MAP, ROW) secara otomatis dikecualikan. Untuk informasi selengkapnya, lihat Predicate Pushdown.
table.kv.format	Enum	COMPACTED	Format penyimpanan untuk data KV. Format yang didukung adalah `COMPACTED` (bawaan) dan `INDEXED`.
table.log.tiered.local-segments	Integer	2	Saat penyimpanan log tiered diaktifkan, parameter ini menentukan jumlah segmen log yang dipertahankan secara lokal untuk setiap tabel. Nilainya harus lebih besar dari 0. Nilai bawaannya adalah 2.
table.datalake.enabled	Boolean	false	Menentukan apakah penyimpanan lakehouse diaktifkan untuk tabel. Secara bawaan dinonaktifkan. Jika diaktifkan dan layanan tiering lakehouse aktif, Fluss secara otomatis melakukan tiering dan kompresi data tabel ke penyimpanan lakehouse.
table.datalake.format	Enum	(None)	Menentukan format data lake yang digunakan tabel untuk penyimpanan lakehouse berlapis. Saat ini, hanya `paimon` yang didukung. Fungsi utama parameter ini adalah: Memastikan konsistensi dalam strategi encoding kunci dan bucketing antara Fluss dan sistem lakehouse eksternal. Memungkinkan kueri terpadu (Union Read) lintas tabel Fluss dan data lake. Perilaku Konfigurasi Prioritas tertinggi: Jika `table.datalake.format` diatur secara eksplisit saat pembuatan tabel, nilai tersebut yang digunakan. Nilai bawaan: Jika tidak diatur, konfigurasi `datalake.format` tingkat kluster di Fluss yang digunakan. Aktivasi dinamis: Anda dapat mengonfigurasi `table.datalake.format` sebelumnya saat `table.datalake.enabled` bernilai `false`. Anda kemudian dapat mengaktifkan fungsionalitas data lake dengan mengaktifkan `table.datalake.enabled` tanpa perlu membuat ulang tabel.
table.merge-engine	Enum	(None)	Menentukan mesin merge untuk tabel primary key. Strategi bawaannya adalah `last_row`, yang menyimpan baris terbaru. Strategi lain yang didukung adalah: `first_row`: Menyimpan baris pertama untuk primary key tertentu. `versioned`: Menyimpan baris dengan nomor versi tertinggi untuk primary key tertentu.
table.merge-engine.versioned.ver-column	String	(None)	Jika mesin merge diatur ke `versioned`, parameter ini wajib diisi dan harus menentukan nama kolom versi.
table.log.local-ttl	Duration	(None)	Masa hidup untuk segmen log lokal. Opsi ini mengontrol waktu maksimum segmen log lokal dipertahankan sebelum memenuhi syarat untuk dihapus. Saat tiering log remote diaktifkan, segmen lokal yang kedaluwarsa hanya dihapus setelah disalin ke penyimpanan remote. Atur nilainya ke `0ms` untuk menonaktifkan penghapusan berbasis TTL. Jika tidak dikonfigurasi, nilai ini mewarisi `table.log.ttl`. Saat kedua nilai positif, nilai ini harus kurang dari atau sama dengan `table.log.ttl`.
table.kv.format-version	Integer	(None)	Versi format KV. Jika tidak dikonfigurasi, koordinator mengaturnya secara otomatis saat pembuatan tabel. Strategi encoding dan bucketing datalake yang dijelaskan di bawah hanya berlaku ketika `datalake.format` dikonfigurasi di tingkat kluster. Versi 1: Tabel yang dibuat sebelum pengenalan `table.kv.format-version` dianggap sebagai versi 1. Baik primary key maupun bucket key menggunakan encoder datalake, seperti Paimon atau Iceberg. Beberapa encoder datalake tidak menjamin bahwa byte bucket key yang diencode merupakan awalan dari byte primary key yang diencode, sehingga pencarian berbasis awalan mungkin tidak berfungsi dengan baik. Versi 2 (saat ini): Untuk tabel baru yang bucket key-nya berbeda dari primary key, primary key menggunakan encoder bawaan Fluss, yang menjamin dukungan pencarian berbasis awalan yang tepat. Saat bucket key sama dengan primary key, encoder datalake tetap digunakan untuk mengoptimalkan penulisan ke lake. Encoding bucket key selalu menggunakan encoder datalake agar selaras dengan perhitungan bucket di datalake.
table.changelog.image	Enum	FULL	Mode gambar changelog untuk tabel primary key. Semantiknya mengikuti pengaturan serupa di sistem database seperti `binlog_row_image` di MySQL dan `replica identity` di PostgreSQL. Opsi ini hanya berlaku untuk tabel primary key. `FULL` (bawaan): Menghasilkan catatan UPDATE_BEFORE dan UPDATE_AFTER untuk operasi pembaruan. Ini menangkap informasi lengkap tentang pembaruan dan memungkinkan Anda melacak nilai sebelumnya. `WAL`: Tidak menghasilkan catatan UPDATE_BEFORE. Hanya catatan INSERT, UPDATE_AFTER, dan DELETE (jika diizinkan) yang dikirim. Saat mode WAL diaktifkan, mesin merge bawaan digunakan, pembaruan bersifat full row update bukan partial update, dan tidak ada kolom auto-increment, sehingga optimasi melewati pencarian nilai lama dan mengonversi operasi INSERT menjadi event UPDATE_AFTER. Mode ini mengurangi biaya penyimpanan dan transmisi tetapi kehilangan kemampuan melacak nilai sebelumnya.
table.delete.behavior	Enum	ALLOW	Mengontrol perilaku operasi penghapusan pada tabel primary key. Opsi ini memberikan jaminan tingkat sistem untuk pipeline downstream, seperti Flink Delta Join, yang tidak boleh menerima event penghapusan dalam changelog tabel. Opsi ini hanya berlaku untuk tabel primary key. `ALLOW` (bawaan untuk mesin merge bawaan): Mengizinkan operasi penghapusan normal. `IGNORE`: Mengabaikan permintaan penghapusan tanpa menampilkan error. `DISABLE`: Menolak permintaan penghapusan dan melemparkan error eksplisit. Untuk tabel dengan mesin merge `first_row`, `versioned`, atau `aggregation`, opsi ini secara otomatis diatur ke `IGNORE` dan tidak dapat di-override. Untuk mesin merge `aggregation`, saat diatur ke `ALLOW`, operasi penghapusan menghapus seluruh catatan.
table.auto-inc.batch-size	Long	100000	Ukuran batch ID auto-increment yang diambil dari counter terdistribusi setiap kali. Nilai ini menentukan panjang segmen ID yang di-cache secara lokal. Ukuran batch yang lebih besar dapat menyebabkan celah signifikan pada ID auto-increment, terutama ketika segmen ID yang di-cache tidak digunakan dibuang akibat restart TabletServer atau terminasi abnormal. Ukuran batch yang lebih kecil meningkatkan frekuensi permintaan pengambilan ID ke counter terdistribusi, yang menimbulkan overhead jaringan tambahan dan mengurangi throughput tulis.
table.datalake.auto-compaction	Boolean	false	Apakah kompaksi dipicu secara otomatis saat layanan tiering menulis ke datalake. Secara bawaan dinonaktifkan.
table.datalake.auto-expire-snapshot	Boolean	false	Apakah kedaluwarsa snapshot dipicu secara otomatis saat layanan tiering melakukan commit ke datalake. Secara bawaan dinonaktifkan.
table.datalake.historical-partition.enabled	Boolean	false	Apakah pencarian partisi historis diaktifkan untuk tabel. Saat diaktifkan, koordinator membuat dan mempertahankan partisi sistem untuk mengarahkan pencarian partisi yang kedaluwarsa ke penyimpanan lake. Saat ini, opsi ini hanya mendukung tabel primary key Paimon berpartisi otomatis dengan satu partition key. Setelah mengubah opsi ini, restart job lookup yang ada yang perlu mencari data partisi historis agar klien mereka memuat konfigurasi tabel yang diperbarui.

Parameter baca

Parameter	Tipe	Bawaan	Deskripsi
scan.startup.mode	Enum	full	Mode startup konektor, yang menentukan dari mana mulai mengonsumsi data. Mode yang didukung adalah `full` (bawaan), `earliest`, `latest`, dan `timestamp`. Untuk detailnya, lihat Baca Data.
scan.startup.timestamp	Long	(None)	Timestamp untuk mulai membaca. Opsi ini hanya berlaku saat `scan.startup.mode` adalah `timestamp`. Tentukan nilainya sebagai milidetik sejak epoch (misalnya, '1678883047356') atau string 'yyyy-MM-dd HH:mm:ss' (misalnya, '2023-12-09 23:09:12'). Catatan Parameter ini tidak mendukung fungsi waktu bawaan Flink, seperti CURRENT_TIMESTAMP, untuk menghasilkan nilai waktu secara dinamis. Parameter konektor dalam klausa WITH hanya menerima literal tetap dan tidak dapat menyelesaikan ekspresi fungsi SQL. Untuk meneruskan waktu dinamis, masukkan timestamp tetap atau nilai milidetik dari luar job menggunakan variabel.
scan.partition.discovery.interval	Duration	10s	Interval bagi sumber Fluss untuk memeriksa partisi baru. Sumber secara dinamis menambahkan partisi baru ke rencana baca dan berhenti berlangganan dari partisi yang dihapus. Nilai negatif menonaktifkan penemuan partisi otomatis.
client.scanner.log.check-crc	Boolean	true	Menentukan apakah akan melakukan pemeriksaan CRC32 pada pesan yang dibaca. Pemeriksaan ini memverifikasi bahwa pesan tidak rusak saat transit atau di disk, meningkatkan keandalan data. Meskipun ini menambahkan sedikit overhead komputasi, kami merekomendasikan mengaktifkannya di lingkungan produksi untuk memastikan integritas data.
client.scanner.log.max-poll-records	Integer	500	Jumlah maksimum catatan yang dikembalikan oleh satu panggilan `poll()` dari LogScanner. Perhatikan bahwa pengaturan ini tidak memengaruhi perilaku fetch dasar. LogScanner menyimpan catatan dari setiap permintaan fetch dan mengembalikannya secara bertahap pada setiap poll.
client.scanner.log.fetch.max-bytes	MemorySize	16mb	Ukuran data maksimum per permintaan fetch. Jika batch catatan pertama dalam bucket pertama yang tidak kosong melebihi batas ini, klien tetap mengembalikannya untuk memastikan kemajuan. Oleh karena itu, nilai ini merupakan batas lunak, bukan batas mutlak.
client.scanner.log.fetch.max-bytes-for-bucket	MemorySize	1mb	Jumlah maksimum data yang dikembalikan per bucket tabel dalam satu permintaan fetch. Catatan diambil dalam batch, dan ukuran byte maksimum dikonfigurasi oleh opsi ini.
client.scanner.log.fetch.min-bytes	MemorySize	1b	Jumlah minimum data dalam byte yang diharapkan klien dari server sebagai respons terhadap setiap permintaan fetch log. Jika ukuran data tidak mencukupi, klien menunggu maksimal selama `client.scanner.log.fetch-wait-max-time` sebelum mengembalikan respons.
client.scanner.log.fetch.wait-max-time	Duration	500ms	Waktu maksimum klien menunggu respons dari server untuk setiap permintaan fetch log.
client.scanner.io.tmpdir	String	System.getProperty("java.io.tmpdir") + "/fluss"	Direktori lokal tempat klien menyimpan file data sementara, seperti snapshot KV dan file segmen log, untuk dibaca.
client.scanner.remote-log.prefetch-num	Integer	4	Jumlah segmen log remote yang diunduh dari penyimpanan remote oleh LogScanner. Segmen log ini disimpan dalam file sementara lokal. Nilai bawaannya adalah 4.
client.remote-file.download-thread-num	Integer	3	Jumlah thread yang digunakan klien untuk mengunduh file remote.
scan.kv.snapshot.lease.id	String	UUID	ID lease yang digunakan untuk melindungi snapshot KV yang telah diambil dari penghapusan. Jika ditentukan, snapshot dipertahankan hingga konsumen selesai memproses semuanya atau durasi lease berakhir. Jika tidak ditentukan secara eksplisit, nilai ini adalah string UUID yang dihasilkan secara acak.
scan.kv.snapshot.lease.duration	Duration	1day	Periode waktu sebelum lease snapshot KV kedaluwarsa, yang mencegah snapshot menghalangi penghapusan.

Parameter tulis

Parameter	Tipe	Bawaan	Deskripsi
sink.ignore-delete	Boolean	false	Saat `true`, sink mengabaikan event changelog `DELETE` dan `UPDATE_BEFORE`.
sink.bucket-shuffle	Boolean	true	Menentukan apakah akan melakukan shuffle data berdasarkan ID bucket sebelum menulis ke sink. Shuffle untuk mengelompokkan catatan dengan ID bucket yang sama ke dalam task yang sama meningkatkan efisiensi sisi klien dan mengurangi penggunaan resource. Untuk tabel log, ini memerlukan `bucket.key`. Opsi ini diaktifkan secara bawaan untuk tabel primary key.
client.writer.buffer.memory-size	MemorySize	64mb	Jumlah total byte yang dapat digunakan writer untuk menyangga baris internal.
client.writer.batch-size	MemorySize	2mb	Ukuran batch untuk satu bucket. Writer mengelompokkan catatan untuk bucket yang sama ke dalam batch, yang meningkatkan kinerja klien dan server.
client.writer.buffer.wait-timeout	Duration	2^(63)-1ns	Menentukan berapa lama writer memblokir saat menunggu segmen tersedia.
client.writer.batch-timeout	Duration	100ms	Untuk meningkatkan throughput, writer dapat menambahkan penundaan buatan kecil untuk menunggu lebih banyak catatan tiba sebelum mengirim batch. Ini mirip dengan algoritma Nagle di TCP dan menetapkan batas atas pada latensi pengiriman. Batch dikirim segera saat mencapai `client.writer.batch-size`, tetapi jika lebih kecil, writer menunggu hingga timeout ini untuk lebih banyak catatan.
client.writer.bucket.no-key-assigner	Enum	STICKY	Menentukan strategi penugasan bucket untuk tabel tanpa kunci. Untuk tabel dengan bucket key atau primary key, sistem memilih bucket berdasarkan nilai hash kunci. Untuk tabel tanpa kunci, Anda dapat memilih assigner bucket dengan parameter ini. Strategi berikut didukung: `STICKY`: Menetapkan ID bucket baru hanya saat batch catatan berubah. Jika tidak, menggunakan ID bucket yang sama seperti catatan sebelumnya. `ROUND_ROBIN`: Menetapkan ID bucket ke baris input secara bergiliran.
client.writer.acks	String	all	Nilai yang valid: `-1` (atau `all`): Leader menunggu semua replica sinkron (ISRs) mengakui catatan. Ini menjamin tidak ada kehilangan data selama setidaknya satu replica sinkron tetap aktif. `1`: Leader mengakui segera setelah menulis catatan ke log lokal, tanpa menunggu replica. Jika leader gagal sebelum replikasi selesai, data mungkin hilang. `0`: Writer tidak menunggu konfirmasi apa pun dari server. Writer tidak dapat menjamin server menerima catatan. Kesalahan jaringan atau kegagalan server dapat menyebabkan kehilangan data. Ini cocok untuk skenario throughput tinggi di mana kehilangan data sebagian dapat diterima. Catatan Di lingkungan produksi, gunakan `all` untuk memastikan ketersediaan tinggi dan mencegah kehilangan data.
client.writer.request-max-size	MemorySize	10mb	Ukuran maksimum permintaan dalam byte. Pengaturan ini membatasi jumlah batch catatan yang dapat dikirim writer dalam satu permintaan untuk menghindari permintaan yang terlalu besar.
client.writer.retries	Integer	Integer.MAX_VALUE	Mengatur nilai lebih dari nol menyebabkan klien mengirim ulang catatan apa pun yang gagal dengan error yang berpotensi sementara.
client.writer.enable-idempotence	Boolean	true	Menentukan apakah akan mengaktifkan idempotensi, yang diaktifkan secara bawaan. Saat diaktifkan, writer memastikan semantik tepat-sekali, yang mencegah duplikasi dan mempertahankan urutan. Jika dinonaktifkan, retry klien akibat kegagalan server dapat menghasilkan pesan duplikat.
client.writer.max-inflight-requests-per-bucket	Integer	5	Jumlah maksimum permintaan yang belum dikonfirmasi per bucket. Pengaturan ini hanya berlaku saat `client.writer.enable-idempotence` diatur ke `true`. Jika jumlah permintaan yang sedang berjalan untuk suatu bucket mencapai batas ini, writer memblokir hingga permintaan tertunda dikonfirmasi sebelum mengirim yang lain.
client.writer.buffer.page-size	MemorySize	128kb	Ukuran setiap halaman dalam buffer memori yang ditentukan oleh `client.writer.buffer.memory-size`.
client.writer.buffer.per-request-memory-size	MemorySize	16mb	Jumlah minimum byte yang dialokasikan oleh writer, dibulatkan ke bawah ke kelipatan terdekat dari `client.writer.buffer.page-size`. Nilainya harus lebih besar dari atau sama dengan `client.writer.buffer.page-size`. Alokasi memori dalam batch meningkatkan keramahan cache CPU karena segmen-segmennya bersambungan.
client.writer.dynamic-batch-size.enabled	Boolean	true	Mengontrol apakah writer klien secara dinamis menyesuaikan ukuran batch berdasarkan throughput tulis aktual. Diaktifkan secara bawaan. Dengan ukuran batch dinamis diaktifkan, writer menyesuaikan alokasi memori per batch sesuai dengan ukuran tulis historis untuk tabel atau partisi target, yang meningkatkan pemanfaatan memori dan kinerja dalam kondisi throughput yang bervariasi. Ukuran batch dinamis dibatasi: tidak melebihi `client.writer.batch-size` atau turun di bawah `client.writer.buffer.page-size`. Saat dinonaktifkan, writer menggunakan ukuran batch tetap `client.writer.batch-size` untuk semua batch. Hal ini dapat menyebabkan sering terjadi penundaan memori dan kinerja tulis suboptimal jika laju data masuk tidak konsisten di berbagai partisi.
sink.distribution-mode	Enum	AUTO	Mode distribusi untuk melakukan shuffle data ke sink. `AUTO` (bawaan): Secara otomatis memilih mode terbaik berdasarkan tipe tabel. Menggunakan BUCKET untuk tabel primary key dan tabel log dengan bucket key untuk memaksimalkan throughput, dan menggunakan NONE untuk tabel log tanpa bucket key. `NONE`: Menggunakan strategi shuffle Flink bawaan. Biasanya menggunakan FORWARD saat paralelisme sink cocok dengan paralelisme upstream, dan ROUND_ROBIN saat paralelisme berbeda. `BUCKET`: Melakukan shuffle data berdasarkan ID bucket sebelum menulis ke sink, sehingga data dengan ID bucket yang sama diproses oleh task yang sama. Ini meningkatkan efisiensi pemrosesan klien dan mengurangi konsumsi resource. Mode ini sangat direkomendasikan untuk tabel primary key karena dapat meningkatkan throughput secara signifikan. Untuk tabel log, shuffle bucket hanya berlaku saat `bucket.key` didefinisikan. Saat paralelisme sink melebihi jumlah bucket, beberapa task sink mungkin tetap menganggur. `PARTITION_DYNAMIC`: Secara dinamis menyesuaikan strategi shuffle berdasarkan pola lalu lintas partition key. Mode ini memantau distribusi data dan menyesuaikan perilaku shuffle untuk menyeimbangkan beban. Hanya didukung untuk tabel log berpartisi, bukan untuk tabel primary key. Gunakan mode ini saat data sangat miring di berbagai partisi atau saat terdapat banyak partisi. Mode ini memiliki biaya overhead, termasuk pengumpulan statistik data dan operasi shuffle tambahan.

Parameter tabel dimensi

Parameter	Tipe	Bawaan	Deskripsi
lookup.async	Boolean	true	Menentukan apakah akan menggunakan lookup asinkron. Lookup asinkron memberikan throughput yang lebih baik daripada lookup sinkron.
`lookup.insert-if-not-exists`	Boolean	false	Saat `true`, secara otomatis menyisipkan baris baru jika kunci lookup tidak ditemukan. Fluss menyisipkan baris baru menggunakan kunci lookup sebagai primary key lalu mengembalikan baris tersebut. Fitur ini hanya berlaku untuk lookup primary key dan tidak didukung untuk lookup berbasis awalan.
lookup.cache	Enum	NONE	Kebijakan caching untuk tabel lookup ini. Nilai yang didukung adalah `NONE` dan `PARTIAL`.
lookup.max-retries	Integer	3	Jumlah maksimum retry yang diizinkan saat operasi lookup gagal.
lookup.partial-cache.expire-after-access	Duration	(None)	Waktu kedaluwarsa entri cache setelah diakses.
lookup.partial-cache.expire-after-write	Duration	(None)	Waktu kedaluwarsa entri cache setelah ditulis.
lookup.partial-cache.cache-missing-key	Boolean	true	Jika `true`, menyimpan fakta bahwa kunci tidak ditemukan (dengan menyimpan nilai null) untuk menghindari lookup berulang untuk kunci yang tidak ada.
lookup.partial-cache.max-rows	Long	(None)	Jumlah maksimum baris yang disimpan dalam cache.
client.lookup.queue-size	Integer	25600	Jumlah maksimum operasi lookup yang tertunda.
client.lookup.max-batch-size	Integer	128	Jumlah maksimum lookup yang dibatch ke dalam satu permintaan.
client.lookup.max-inflight-requests	Integer	128	Jumlah maksimum permintaan lookup yang belum dikonfirmasi untuk operasi lookup.
client.lookup.batch-timeout	Duration	100ms	Waktu maksimum untuk menunggu batch lookup terisi. Jika timeout ini tercapai, batch lookup ditutup dan dikirim.
client.lookup.max-retries	Integer	Integer.MAX_VALUE	Jumlah kali klien mengirim ulang permintaan lookup yang gagal dengan error yang berpotensi sementara. Nilai lebih dari nol mengaktifkan pengiriman ulang.

Parameter lainnya

Parameter	Tipe	Bawaan	Deskripsi
bootstrap.servers	List	(None)	Tentukan daftar awal node yang digunakan untuk terhubung ke kluster Fluss, dalam format `host1:port1,host2:port2,...`. Klien menggunakan daftar ini untuk menemukan topologi kluster lengkap, sehingga tidak perlu mencakup semua server. Kami merekomendasikan mengonfigurasi beberapa node untuk ketersediaan tinggi guna menghindari single point of failure.
client.id	String	""	String ID yang dikirim ke server bersama setiap permintaan. Menggunakan ID yang bermakna (misalnya, `order-service-prod`) membantu mengidentifikasi sumber permintaan dalam log sisi server.
client.connect-timeout	Duration	120s	Timeout koneksi untuk klien Netty.
client.request-timeout	Duration	30s	Timeout untuk menyelesaikan permintaan. Saat `client.writer.acks` diatur ke `all` (-1), ini menentukan waktu maksimum klien menunggu tulisan sepenuhnya dikonfirmasi. Bawaan: 30s.
client.filesystem.security.token.renewal.backoff	Duration	1h	Interval waktu untuk menunggu sebelum mencoba mengambil token keamanan baru untuk sistem file setelah kegagalan pengambilan.
client.filesystem.security.token.renewal.time-ratio	Double	0.75	Rasio dari masa berlaku token yang menentukan kapan harus memperbarui token. Misalnya, rasio 0.75 berarti pembaruan dicoba setelah 75% dari periode validitas token berlalu.
client.metrics.enabled	Boolean	false	Mengaktifkan metrik klien. Saat diaktifkan, klien mengumpulkan dan melaporkan metrik melalui reporter metrik JMX.