Realtime Compute for Apache Flink:Parameter kunci

Catatan penggunaan

Sebelum pekerjaan dimulai, jika Anda tidak secara eksplisit mengatur parameter ini di bagian Parameters, managed memory digunakan untuk mem-buffer data dalam mode mini-batch. Salah satu kondisi berikut akan memicu komputasi akhir dan output data:

• Pesan watermark diterima dari operator MiniBatchAssigner.

• Managed memory penuh.

• Sebelum checkpoint dimulai.

• Pekerjaan dihentikan.

Nilai yang tersedia

• -1 (default): Menunjukkan bahwa managed memory digunakan untuk mem-buffer data.

• Nilai Long negatif lainnya: Sama seperti pengaturan default.

• Nilai Long positif lainnya: Menunjukkan bahwa heap memory digunakan untuk mem-buffer data. Saat jumlah catatan yang dibuffer mencapai nilai ini (N), sistem secara otomatis memicu operasi output.

Contoh

1. Buat Pekerjaan streaming SQL, salin kode SQL berikut, lalu deploy pekerjaan tersebut.

   CREATE TEMPORARY TABLE s1 (
     a INT,
     b INT,
     ts TIMESTAMP(3),
     PRIMARY KEY (a) NOT ENFORCED,
     WATERMARK FOR ts AS ts - INTERVAL '1' SECOND
   ) WITH (
     'connector'='datagen',
     'rows-per-second'='1',
     'fields.ts.kind'='random',
     'fields.ts.max-past'='5s',
     'fields.b.kind'='random',
     'fields.b.min'='0',
     'fields.b.max'='10'
   );
   CREATE TEMPORARY TABLE sink (
     a INT,
     b BIGINT,
     PRIMARY KEY (a) NOT ENFORCED
   ) WITH (
     'connector'='print'
   );
   INSERT INTO sink SELECT a, sum(b) FROM s1 GROUP BY a;

2. Di tab Deployment Details halaman Job O&M, di field Other Configurations bagian Runtime Parameter Configuration, atur parameter table.exec.mini-batch.enabled: true dan table.exec.mini-batch.allow-latency: 2s, dan jangan atur table.exec.mini-batch.size agar menggunakan nilai default -1.

3. Start pekerjaan tersebut. Di tab Status, topologi pekerjaan mencakup operator MiniBatchAssigner, LocalGroupAggregate, dan GlobalGroupAggregate.

Catatan penggunaan

• Parameter ini hanya berlaku di VVR 8.0.8 dan versi setelahnya. Di versi sebelum VVR 8.0.8, perilakunya setara dengan mengatur parameter ini ke false.

• Saat mengubah nilai dari false ke true, jumlah data yang dikirim ke hilir dari operator MinibatchGlobalAgg dan MinibatchAgg dapat meningkat, sehingga memberikan tekanan lebih besar pada operator hilir.

Nilai yang tersedia

• false (default): Saat state TTL diaktifkan, operator MinibatchGlobalAgg dan MinibatchAgg tidak mengeluarkan data ke hilir jika hasil agregasi tidak berubah.

• true: Saat state TTL diaktifkan, operator MinibatchGlobalAgg dan MinibatchAgg tetap mengeluarkan catatan (record) terbaru (duplikat) ke hilir meskipun hasil agregasi tidak berubah.

Contoh

1. Buat Pekerjaan streaming SQL, salin kode SQL berikut, lalu deploy pekerjaan tersebut.

   create temporary table src(
       a int,
       b string
   ) with (
       'connector' = 'datagen',
       'rows-per-second' = '10',
       'fields.a.min' = '1',
       'fields.a.max' = '1',
       'fields.b.length' = '3'
   );
   create temporary table snk(
       a int,
       max_length_b bigint
   ) with (
       'connector' = 'blackhole'
   );
   insert into snk select a, max(CHAR_LENGTH(b)) from src group by a; 

2. Di bagian Other Configurations area Runtime Parameter Configuration pada tab Deployment Details halaman Job O&M, atur parameter table.exec.mini-batch.enabled: true dan table.exec.mini-batch.allow-latency: 2s untuk mengaktifkan optimasi Minibatch Aggregate.

3. Start pekerjaan tersebut. Di tab Status, pekerjaan mencakup operator MinibatchGlobalAggregate. Klik tanda "+" pada operator tersebut untuk melihat bahwa operator GlobalGroupAggregate tidak mengirim data ke hilir saat hasil agregasi tidak berubah.

   Operator tersebut menampilkan RecordsIn sebesar 19 dan RecordsOut sebesar 1, yang berarti 19 catatan input hanya menghasilkan 1 output agregat.

4. Hentikan pekerjaan tersebut, lalu tambahkan parameter table.exec.agg.mini-batch.output-identical-enabled: true ke Other Configurations di bagian Running Parameter Configuration pada halaman Deployment Details halaman Job O&M.

5. Start pekerjaan tersebut. Di tab Status, Anda dapat melihat pekerjaan mencakup operator MinibatchGlobalAggregate. Klik tanda "+" pada operator tersebut untuk mengamati bahwa operator GlobalGroupAggregate kini mengirim data ke hilir meskipun hasil agregasi tidak berubah. Setelah pekerjaan dimulai ulang, tab Status menunjukkan bahwa RecordsIn dan RecordsOut operator GlobalGroupAggregate keduanya bernilai 94. Ini menunjukkan bahwa dengan table.exec.agg.mini-batch.output-identical-enabled: true diaktifkan, operator tersebut mengirim data ke hilir meskipun hasil agregasi tidak berubah.

Kasus penggunaan

• Gunakan optimasi ini untuk mempertahankan urutan pemrosesan per kunci dalam join tabel dimensi ketika aliran memiliki sedikit pesan dengan kunci pembaruan yang sama dalam periode waktu tertentu.

• Dalam aliran Change Data Capture (CDC) dengan primary key, Anda melakukan penguatan data dengan join tabel dimensi dan menulis ke sink di mana primary key sink cocok dengan primary key sumber. Kunci gabungan (join key) untuk join tabel dimensi berbeda dari primary key, dan join key sisi dimensi adalah primary key. Optimasi ini melakukan shuffle berdasarkan primary key CDC, yang diturunkan sebagai kunci pembaruan (update key). Dibandingkan dengan mengaktifkan optimasi SHUFFLE_HASH untuk skenario yang sama, optimasi ini menghindari pembuatan operator SinkMaterializer sebelum sink pada parallelism yang lebih tinggi. Hal ini menghindari potensi masalah performa dari operator tersebut, terutama state besar yang dapat terakumulasi selama eksekusi jangka panjang. Untuk informasi tentang SinkUpsertMaterializer, lihat Rekomendasi.

• Kunci gabungan (join key) untuk join tabel dimensi berbeda dari primary key, join key sisi dimensi adalah primary key, dan diikuti oleh operator rank. Optimasi ini melakukan shuffle berdasarkan primary key CDC, yang diturunkan sebagai kunci pembaruan (update key). Dibandingkan dengan mengaktifkan optimasi SHUFFLE_HASH untuk skenario yang sama, optimasi ini mencegah UpdateFastRank menurun menjadi RetractRank. Untuk informasi tentang cara mengoptimalkan RetractRank menjadi UpdateFastRank, lihat Teknik optimasi TopN.

Catatan penggunaan

• Jika aliran tidak memiliki kunci pembaruan (update key), seluruh baris digunakan sebagai kunci.

• Throughput menurun jika kunci pembaruan yang sama diperbarui secara sering dalam periode singkat, karena catatan untuk kunci pembaruan yang sama diproses dalam urutan yang ketat.

• Dibandingkan dengan join tabel dimensi asinkron asli, mode Key-Ordered memperkenalkan keyed state. Mengaktifkan atau menonaktifkan mode ini memengaruhi kompatibilitas state.

• Fitur ini hanya berlaku untuk VVR 8.0.10 dan versi setelahnya ketika input join tabel dimensi merupakan update stream dan Anda mengonfigurasi table.exec.async-lookup.output-mode='ALLOW_UNORDERED' dan table.exec.async-lookup.key-ordered-enabled='true'.

Nilai yang tersedia

• false (default): Menonaktifkan mode Key-Ordered.

• true: Mengaktifkan mode Key-Ordered.

Contoh

1. Contoh berikut menggunakan join tabel dimensi Hologres asinkron. Buat Pekerjaan streaming SQL, salin kode SQL berikut, lalu deploy pekerjaan tersebut.

   Untuk informasi lebih lanjut tentang konektor Hologres, lihat Hologres.

   create TEMPORARY table bid_source(
     auction  BIGINT,
     bidder  BIGINT,
     price  BIGINT,
     channel  VARCHAR,
     url  VARCHAR,
     dateTime  TIMESTAMP(3),
     extra  VARCHAR,
     proc_time as proctime(),
     WATERMARK FOR dateTime AS dateTime - INTERVAL '4' SECOND
   ) with (
     'connector' = 'kafka',  -- Konektor aliran non-insert-only
     'topic' = 'user_behavior',
     'properties.bootstrap.servers' = 'localhost:9092',
     'properties.group.id' = 'testGroup',
     'scan.startup.mode' = 'earliest-offset',
     'format' = 'csv'
   );
   CREATE TEMPORARY TABLE users (
       user_id STRING PRIMARY KEY NOT ENFORCED, -- Definisikan primary key
       user_name VARCHAR(255) NOT NULL, 
       age INT NOT NULL
   ) WITH (
       'connector' = 'hologres',    -- Konektor yang mendukung lookup asinkron
       'async' = 'true',
       'dbname' = 'holo db name', --Nama database Hologres Anda
       'tablename' = 'schema_name.table_name', --Nama tabel Hologres yang menerima data
       'username' = 'access id', --ID AccessKey Akun Alibaba Cloud Anda
       'password' = 'access key', --Secret AccessKey Akun Alibaba Cloud Anda
       'endpoint' = 'holo vpc endpoint', --Titik akhir VPC instans Hologres Anda
   );
   CREATE TEMPORARY TABLE bh ( 
       auction  BIGINT,
       age int
   ) WITH (
       'connector' = 'blackhole'
   );
   insert into bh
   SELECT
       bid_source.auction,
       u.age
   FROM bid_source
       JOIN users FOR SYSTEM_TIME AS OF bid_source.proc_time AS u
       ON bid_source.channel = u.user_id;
   

2. Di halaman Job O&M, di tab Deployment Details, di bagian Other Configurations area Runtime Parameter Configuration, atur parameter table.exec.async-lookup.output-mode='ALLOW_UNORDERED' dan table.exec.async-lookup.key-ordered-enabled='true'.

3. Start pekerjaan tersebut. Di tab Status, Anda dapat melihat bahwa atribut async pekerjaan tersebut adalah KEY_ORDERED:true.

table.optimizer.window-join-enabled

Parameter ini mengontrol apakah operasi window join diaktifkan. Saat diaktifkan, Flink mengoptimalkan rencana eksekusi yang sesuai sebagai window join. Untuk jendela (window) kecil, hal ini mengurangi overhead state dan meningkatkan performa. Dibandingkan dengan regular join, window join juga dapat menghindari pengiriman pesan pembaruan ke operator hilir, yang berguna untuk kasus penggunaan yang memerlukan join dengan kondisi jendela waktu kecil.