Realtime Compute for Apache Flink:Agregasi jendela

Sintaks SESSION TVF jendela: VVR 11.x vs VVR 8.x

Sintaks SESSION TVF jendela berbeda antar versi VVR. Upgrade ke VVR 11.1 atau yang lebih baru untuk menggunakan sintaks lengkap.

SESSION(TABLE data [PARTITION BY(keycols, ...)], DESCRIPTOR(timecol), gap)

SESSION(TABLE data, DESCRIPTOR(timecol), gap)

-- Skema tabel Bid (digunakan di semua contoh di bawah)
> desc Bid;
+-------------+------------------------+------+-----+--------+---------------------------------+
|        name |                   type | null | key | extras |                       watermark |
+-------------+------------------------+------+-----+--------+---------------------------------+
|     bidtime | TIMESTAMP(3) *ROWTIME* | true |     |        | `bidtime` - INTERVAL '1' SECOND |
|       price |         DECIMAL(10, 2) | true |     |        |                                 |
|        item |                 STRING | true |     |        |                                 |
+-------------+------------------------+------+-----+--------+---------------------------------+

-- VVR 11.x: kolom partisi dideklarasikan eksplisit dalam SESSION()
> SELECT window_start, window_end, item, SUM(price) AS total_price
  FROM TABLE(
      SESSION(TABLE Bid PARTITION BY item, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES))
  GROUP BY item, window_start, window_end;

-- VVR 8.x: kolom partisi diinferensi dari GROUP BY
> SELECT window_start, window_end, item, SUM(price) AS total_price
  FROM TABLE(
      SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES))
  GROUP BY item, window_start, window_end;

Kapan agregasi TVF jendela kembali ke mode non-TVF

Ketika sebuah kueri mencakup TVF jendela tetapi tidak memenuhi kondisi agar TVF dan agregasi dapat digabungkan, sistem akan kembali ke rencana eksekusi non-TVF.

TVF jendela dan pernyataan agregasi tidak dapat digabungkan jika salah satu kondisi berikut terpenuhi:

1. Filter atau komputasi pada kolom waktu jendela. window_start, window_end, atau window_time difilter atau dimodifikasi sebelum agregasi.

   -- Filter pada window_start
   > SELECT window_start, window_end, item, SUM(price) AS total_price
       FROM
       (SELECT item, price, window_start, window_end FROM
       TABLE(
       SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES))
       WHERE window_start >= TIMESTAMP '2020-04-15 08:06:00.000')
       GROUP BY item, window_start, window_end;
   
   -- Operasi aritmetika pada window_start
   > SELECT window_start, window_end, item, SUM(price) AS total_price
       FROM
       (SELECT item, price, window_start + (INTERVAL '1' SECOND) AS window_start, window_end FROM
       TABLE(
       SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES)))
       GROUP BY item, window_start, window_end;
   
   -- Konversi tipe pada window_start
   > SELECT window_start, window_end, item, SUM(price) AS total_price
       FROM
       (SELECT item, price, CAST(window_start AS varchar) AS window_start, window_end FROM
       TABLE(
       SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES)))
       GROUP BY item, window_start, window_end;

2. TVF jendela digunakan bersama fungsi bernilai tabel yang didefinisikan pengguna (UDTF).

   > SELECT window_start, window_end, category, SUM(price) AS total_price
       FROM
       (SELECT category, price, window_start, window_end FROM
       TABLE(
       SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES)),
       LATERAL TABLE(category_udtf(item)) AS T(category))
       GROUP BY category, window_start, window_end;

3. Klausa `GROUP BY` tidak mencantumkan `window_start` atau `window_end`.

   > SELECT window_start, item, SUM(price) AS total_price
     FROM TABLE(
         SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES))
     GROUP BY item, window_start;

4. Fungsi agregat yang didefinisikan pengguna Python (UDAF) digunakan.

5. `GROUPING SETS`, `CUBE`, atau `ROLLUP` mengelompokkan secara terpisah berdasarkan `window_start` atau `window_end`.

   > SELECT item, SUM(price) AS total_price
     FROM TABLE(
         SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES))
     GROUP BY GROUPING SETS((item), (window_start), (window_end));
   
   > SELECT item, SUM(price) AS total_price
     FROM TABLE(
         SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES))
     GROUP BY CUBE (item, window_start, window_end);
   
   > SELECT item, SUM(price) AS total_price
     FROM TABLE(
         SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES))
     GROUP BY ROLLUP (item, window_start, window_end);

6. Fungsi agregat diterapkan pada `window_start`, `window_end`, atau `window_time`.

   > SELECT window_start, window_end, item, SUM(price) AS total_price, MAX(window_end) AS max_end
     FROM TABLE(
         SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES))
     GROUP BY item, window_start, window_end;

Kondisi 1: Strategi fase agregasi mengizinkan dua fase

table.optimizer.agg-phase-strategy diatur ke AUTO (bawaan) atau TWO_PHASE. Mengatur ke ONE_PHASE menonaktifkan optimisasi dua fase.

Kondisi 2: Jendela menggunakan waktu event

Jendela harus menggunakan waktu event (rowtime). Jendela waktu pemrosesan tidak didukung.

Kondisi 3: Jenis jendela bukan SESSION

Jenis jendela TUMBLE, HOP, dan CUMULATE didukung. Jendela SESSION tidak mendukung optimisasi dua fase.

Kondisi 4: Semua fungsi agregat mendukung penggabungan parsial

Semua fungsi agregat harus mendukung operasi merge. Fungsi bawaan seperti SUM, COUNT, MIN, MAX, dan AVG didukung. UDAF kustom harus mengimplementasikan metode merge().

Kondisi 5: Aliran input hanya insert dan jendela dapat dikonversi ke bentuk TVF

Semua kondisi berikut harus terpenuhi:

• Aliran input hanya berisi operasi insert.

• table.exec.emit.early-fire.enabled diatur ke false (bawaan).

• table.exec.emit.late-fire.enabled diatur ke false (bawaan).

• Untuk jendela HOP, jendela harus sejajar (ukuran jendela habis dibagi interval slide).

Kondisi 6: Distribusi data belum memenuhi persyaratan partisi

Distribusi data input belum memenuhi persyaratan partisi untuk agregasi. Jika data sudah didistribusikan berdasarkan kunci partisi, pengoptimal menentukan bahwa tidak diperlukan pra-agregasi tambahan dan tidak menghasilkan node Agregasi Lokal.

Referensi parameter