Realtime Compute for Apache Flink:Optimalkan Flink SQL

Skalakan sumber daya JobManager

Untuk penerapan paralel, tingkatkan sumber daya JobManager di bagian Resources pada tab Configuration. Contohnya:

• Job Manager CPU: 4

• Job Manager Memory: 8 GiB

Skalakan sumber daya TaskManager

Untuk topologi kompleks, tingkatkan sumber daya TaskManager di bagian Resources pada tab Configuration. Contohnya:

• Task Manager CPU: 2

• Task Manager Memory: 4 GiB

Aktifkan miniBatch

MiniBatch menyimpan catatan masuk dalam buffer dan memprosesnya secara bersamaan, sehingga mengurangi akses status per batch. Hal ini meningkatkan throughput dengan sedikit peningkatan latensi. Fitur miniBatch memicu pemrosesan mikro-batch berdasarkan pesan event yang dimasukkan di sumber pada interval tertentu.

Kapan digunakan: Skenario agregasi data yang tidak memerlukan latensi ultra-rendah.

Cara mengaktifkan: Tambahkan konfigurasi berikut ke bidang Other Configuration di bagian Parameters pada tab Configuration. Konfigurasikan parameter penerapan kustom.

table.exec.mini-batch.enabled: true
table.exec.mini-batch.allow-latency: 5s

Aktifkan LocalGlobal

LocalGlobal membagi agregasi menjadi dua tahap — pra-agregasi lokal di setiap node hulu, lalu agregasi global — mirip dengan combine dan reduce pada MapReduce. Agregasi lokal melakukan pra-filter terhadap data hotspot, sehingga mengurangi volume data yang mencapai tahap global.

Kapan digunakan: Fungsi agregat umum seperti SUM, COUNT, MAX, MIN, dan AVG yang mengalami masalah hotspot data.

Prasyarat:

• MiniBatch harus diaktifkan. LocalGlobal menggunakan interval mini-batch untuk menentukan jumlah catatan yang diakumulasi sebelum agregasi lokal.

• Fungsi agregat harus mengimplementasikan metode merge (AggregateFunction).

Status: Diaktifkan secara default saat miniBatch aktif.

Verifikasi: Periksa topologi akhir untuk keberadaan node GlobalGroupAggregate atau LocalGroupAggregate.

Aktifkan PartialFinal untuk COUNT DISTINCT

LocalGlobal kurang efektif untuk COUNT DISTINCT karena agregasi lokal tidak dapat menghapus duplikat kunci distinct. PartialFinal mengatasi hal ini dengan membagi agregasi menjadi dua lapisan: lapisan pertama menyebarkan data berdasarkan hash kunci distinct, lapisan kedua melakukan agregasi akhir.

Kapan digunakan: Kueri COUNT DISTINCT dengan volume data besar di mana performa agregasi menjadi bottleneck.

Cara mengaktifkan: Tambahkan konfigurasi berikut ke bidang Other Configuration. Konfigurasikan parameter penerapan kustom.

table.optimizer.distinct-agg.split.enabled: true

Batasan:

• PartialFinal tidak dapat digunakan dalam SQL yang mengandung fungsi agregat yang didefinisikan pengguna (UDAF).

• Aktifkan hanya untuk volume data besar. PartialFinal memperkenalkan shuffle jaringan tambahan.

Verifikasi: Periksa apakah agregasi satu lapisan berubah menjadi agregasi dua lapisan dalam topologi akhir.

Gunakan AGG WITH FILTER alih-alih CASE WHEN

Saat menghitung COUNT DISTINCT pada bidang yang sama dengan kondisi berbeda, gunakan klausa FILTER alih-alih CASE WHEN. Pengoptimal mengenali argumen filter berbeda pada kunci distinct yang sama dan menggunakan satu instance status bersama, sehingga mengurangi akses dan ukuran status. Uji performa menunjukkan peningkatan 2x dibandingkan CASE WHEN.

Sebelum (kurang efisien):

COUNT(DISTINCT visitor_id) AS uv_total,
COUNT(DISTINCT CASE WHEN is_wireless = 'y' THEN visitor_id ELSE NULL END) AS uv_wireless

Setelah (performa 2x):

COUNT(DISTINCT visitor_id) AS uv_total,
COUNT(DISTINCT visitor_id) FILTER (WHERE is_wireless = 'y') AS uv_wireless

Gunakan Delta Aggregation (agregasi tanpa status)

Delta Aggregation menyediakan agregasi tanpa status untuk agregasi kelompok streaming non-windowed. Berbeda dengan agregasi kelompok reguler yang terus-menerus memelihara akumulator agregat untuk setiap kelompok dalam status Flink, Delta Aggregation merespons perubahan data dengan menggunakan kunci pengelompokan untuk melakukan lookup asinkron terhadap catatan detail terkini dari kelompok tersebut dan menghitung ulang hasil agregat dari catatan tersebut.

Akibatnya, jumlah status agregasi di sisi Flink tidak lagi bergantung pada jumlah historis kelompok. Penerapan tidak perlu terus memperluas status agregasinya seiring bertambahnya jumlah kelompok. Hal ini mengurangi jumlah data yang dipersist di checkpoint, memperpendek waktu pemulihan, dan menstabilkan penggunaan sumber daya untuk penerapan agregasi kelompok skala besar.

SET 'table.exec.async-lookup.buffer-capacity' = '100';
SET 'table.exec.async-lookup.timeout' = '3 min';

SET 'table.exec.mini-batch.enabled' = 'true';
SET 'table.exec.mini-batch.allow-latency' = '10s';
SET 'table.exec.mini-batch.size' = '1000';

SET 'table.exec.delta-agg.cache-enabled' = 'true';
SET 'table.exec.delta-agg.cache-size' = '2000';

+U(..., 120) → Sesuai kondisi dan ditulis ke tabel hasil
+U(..., 80)  → Tidak sesuai kondisi dan difilter

SET 'table.optimizer.delta-agg.ignore-non-unique-key-filter' = 'true';

Pemisahan key-value untuk join dua aliran

Pada VVR 6.0.1 dan versi lebih baru, mesin secara otomatis menyimpulkan apakah akan mengaktifkan pemisahan key-value untuk operator JOIN dua aliran, meningkatkan performa join khas lebih dari 40%.

Konfigurasikan parameter table.exec.join.kv-separate untuk mengontrol perilaku ini:

Aktifkan miniBatch untuk join

Secara default, operator join reguler memproses catatan satu per satu: mencari status pasangan berdasarkan kunci gabungan, memperbarui status, lalu menghasilkan output. Pola per-catatan ini meningkatkan beban backend status — terutama dengan RocksDB — dan dapat menyebabkan amplifikasi catatan parah pada join bertingkat.

MiniBatch untuk join mengatasi hal ini dengan dua optimasi inti:

1. Pelipatan catatan: Melipat catatan dalam buffer untuk mengurangi volume data sebelum proses join.

2. Penekanan output: Menekan hasil antara yang redundan selama pemrosesan batch.

MiniBatch memicu pemrosesan saat latensi maksimum tercapai, batas ukuran batch tercapai, atau terjadi checkpoint.

Versi VVR: 8.0.4 atau lebih baru.

Cara mengaktifkan: Tambahkan konfigurasi berikut ke bidang Other Configuration. Konfigurasikan parameter penerapan kustom.

table.exec.mini-batch.enabled: true
table.exec.mini-batch.allow-latency: 5s
table.exec.stream.join.mini-batch-enabled: true

Paling cocok untuk: Join outer bertingkat dengan amplifikasi pesan. Operator hulu menggunakan collapse atau merge pesan untuk menekan amplifikasi pada operator hilir. Contohnya:

SELECT a.id AS a_id, a.a_content, B.id AS b_id, B.b_content
FROM a LEFT JOIN
  (SELECT * FROM b
      LEFT JOIN c ON b.prd_id = c.id) B
ON a.id = B.id

Algoritma TopN

Algoritma yang digunakan bergantung pada apakah aliran data masukan bersifat statis atau dinamis:

Nama algoritma muncul dalam nama node topologi.

Beralih dari RetractRank ke UpdateFastRank

UpdateFastRank memerlukan semua kondisi berikut:

1. Aliran input tidak berisi pesan DELETE atau UPDATE_BEFORE. Verifikasi dengan EXPLAIN:

      EXPLAIN CHANGELOG_MODE <query_statement_or_insert_statement_or_statement_set>

2. Aliran input berisi informasi kunci primer (misalnya, kolom yang digunakan dalam klausa GROUP BY).

3. Bidang ORDER BY diperbarui secara monotonik dalam urutan kebalikan dari pengurutan. Misalnya, ORDER BY COUNT DESC, ORDER BY COUNT_DISTINCT DESC, atau ORDER BY SUM(nilai positif) DESC.

Contoh: Untuk ORDER BY SUM DESC, filter nilai positif untuk menjamin monotonicity:

INSERT INTO print_test
SELECT
  cate_id,
  seller_id,
  stat_date,
  pay_ord_amt
FROM (
    SELECT
      *,
      ROW_NUMBER() OVER (
        -- Kolom PARTITION BY harus muncul dalam GROUP BY subkueri.
        -- Sertakan bidang waktu untuk mencegah ketidakteraturan saat status kedaluwarsa.
        PARTITION BY cate_id, stat_date
        ORDER BY pay_ord_amt DESC
      ) AS rownum
    FROM (
        SELECT
          cate_id,
          seller_id,
          stat_date,
          -- SUM meningkat secara monotonik karena hanya nilai positif yang disertakan.
          SUM(total_fee) FILTER (WHERE total_fee >= 0) AS pay_ord_amt
        FROM random_test
        WHERE total_fee >= 0
        GROUP BY seller_id, stat_date, cate_id
      ) a
  )
WHERE rownum <= 100;

Kurangi volume output

Jangan sertakan rownum dalam output SELECT akhir. Urutkan hasil di antarmuka depan alih-alih untuk mengurangi volume tulis sink. Top-N.

Tingkatkan ukuran cache TopN

TopN memelihara cache status untuk mengurangi pembacaan disk. Hitung rasio hit cache dengan rumus ini:

cache_hit = cache_size * parallelism / top_n / partition_key_num

Contoh: Dengan Top100, ukuran cache default 10.000 catatan, paralelisme 50, dan 100.000 kunci partisi:

10.000 * 50 / 100 / 100.000 = 5% rasio hit

Rasio hit 5% berarti sebagian besar pembacaan mengakses disk, menyebabkan metrik state seek tidak stabil dan performa menurun. Tingkatkan ukuran cache:

table.exec.rank.topn-cache-size: 200000

Dengan 200.000 entri cache:

200.000 * 50 / 100 / 100.000 = 100% rasio hit

Jika jumlah kunci partisi besar, tingkatkan juga ukuran cache TopN dan memori heap. Konfigurasikan penerapan.

Sertakan bidang waktu dalam PARTITION BY

Tambahkan bidang waktu seperti day ke klausa PARTITION BY. Tanpa itu, hasil TopN menjadi tidak teratur saat data status kedaluwarsa karena TTL.

Sintaks

Deduplikasi merupakan kasus khusus TopN. Gunakan ROW_NUMBER() dengan klausa OVER untuk memberikan nomor baris per partisi, lalu simpan hanya rownum = 1:

SELECT *
FROM (
   SELECT *,
    ROW_NUMBER() OVER (PARTITION BY col1[, col2..]
     ORDER BY timeAttributeCol [ASC|DESC]) AS rownum
   FROM table_name)
WHERE rownum = 1

Perilaku atribut waktu:

• proctime (waktu pemrosesan): Deduplikasi berdasarkan kapan catatan diproses. Hasil dapat berbeda antar eksekusi.

• rowtime (waktu event): Deduplikasi berdasarkan kapan catatan diproduksi. Hasil deterministik.

Keep FirstRow

Menyimpan catatan pertama per kunci deduplikasi. Status hanya menyimpan data kunci primer, sehingga akses status sangat efisien.

SELECT *
FROM (
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY b ORDER BY proctime) AS rowNum
  FROM T
)
WHERE rowNum = 1

Menghapus data duplikat dari tabel T berdasarkan kolom b, dengan mempertahankan catatan paling awal berdasarkan waktu pemrosesan. Alih-alih mendeklarasikan atribut proctime, gunakan fungsi PROCTIME().

Keep LastRow

Menyimpan catatan paling baru per kunci deduplikasi. Performanya sedikit lebih baik daripada fungsi LAST_VALUE.

SELECT *
FROM (
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY b, d ORDER BY rowtime DESC) AS rowNum
  FROM T
)
WHERE rowNum = 1

Mendeduplikasi tabel T berdasarkan kolom b dan d, dengan mempertahankan catatan terbaru berdasarkan waktu event.

Utamakan fungsi bawaan daripada UDF

Fungsi bawaan dioptimalkan untuk serialisasi, deserialisasi, dan operasi tingkat byte. Ganti UDF dengan padanan bawaan bila memungkinkan.

Gunakan delimiter karakter tunggal dalam KEYVALUE

Fungsi KEYVALUE(content, keyValueSplit, keySplit, keyName) berjalan sekitar 30% lebih cepat saat keyValueSplit dan keySplit berupa karakter tunggal seperti : atau ,. Dengan delimiter karakter tunggal, mesin mencari kunci target langsung dalam data biner tanpa menguraikan seluruh input.

Pola operator LIKE

Escape underscore: Karakter underscore (_) merupakan wildcard karakter tunggal dalam SQL. Untuk mencocokkan underscore literal, gunakan karakter escape:

LIKE '%seller/_id%' ESCAPE '/'

Tanpa escaping, LIKE '%seller_id%' juga akan mencocokkan seller#id, sellerxid, seller1id, dan string lain yang tidak diinginkan.

Hindari ekspresi reguler

Ekspresi reguler bisa 100x lebih lambat daripada operasi aritmetika dan dapat masuk ke loop tak hingga dalam kasus tepi, sehingga menghentikan penerapan. Gunakan LIKE jika memungkinkan.

Untuk kasus yang memerlukan ekspresi reguler, lihat REGEXP dan REGEXP_REPLACE.

Sintaks

Sintaks mengikuti Apache Calcite SQL:

SELECT /*+ hint [, hint ] */ ...

-- Where:
-- hint:       hintName(hintOption [, hintOption]*)
-- hintOption:  simpleIdentifier | numericLiteral | stringLiteral

Petunjuk join

Petunjuk kueri memodifikasi rencana eksekusi dalam blok kueri saat ini. Flink saat ini hanya mendukung petunjuk join, baik untuk join tabel dimensi maupun join reguler.