Realtime Compute for Apache Flink:Window TVF

Referensi cepat

Modifikasi yang menjaga kompatibilitas penuh

Modifikasi berikut memungkinkan penerapan dilanjutkan dari data state yang ada tanpa dampak apa pun terhadap hasil agregasi.

• Menghapus metrik agregat non-distinct. Data state untuk metrik yang dihapus dibuang, dan metrik yang tersisa dilanjutkan dari nilai yang tersimpan tanpa gangguan.

• Mengubah posisi metrik agregat non-distinct. Mengubah urutan kolom agregat dalam daftar SELECT tidak memengaruhi state. Contoh berikut memindahkan max(c) sebelum sum(b), dan penerapan tetap sepenuhnya kompatibel.

  -- Asli
  SELECT a, sum(b), max(c)
  FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
  GROUP BY a, window_start, window_end;
  
  -- Diurut ulang — sepenuhnya kompatibel
  SELECT a, max(c), sum(b)
  FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
  GROUP BY a, window_start, window_end;

• Menambah atau menghapus bidang atribut jendela. Menambah atau menghapus window_start, window_end, atau window_time dari daftar SELECT tidak memengaruhi state.

  -- Asli
  SELECT a, sum(b), max(c), window_start
  FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
  GROUP BY a, window_start, window_end;
  
  -- Tambah window_end — sepenuhnya kompatibel
  SELECT a, sum(b), max(c), window_start, window_end
  FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
  GROUP BY a, window_start, window_end;
  
  -- Hapus window_start — sepenuhnya kompatibel
  SELECT a, sum(b), max(c)
  FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
  GROUP BY a, window_start, window_end;

• Tidak ada metrik agregat sebelum dan sesudah modifikasi. Jika baik SQL asli maupun yang diperbarui tidak menyertakan metrik agregat, state sepenuhnya kompatibel.

• Mengubah urutan kunci pengelompokan terkait jendela. Mengubah urutan kunci window_start atau window_end dalam klausa GROUP BY aman selama urutan relatif semua kunci pengelompokan lain tetap sama.

  -- Asli
  SELECT a, sum(b), max(c)
  FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
  GROUP BY a, c, window_end, window_start;
  
  -- Tukar window_end dan window_start — sepenuhnya kompatibel
  SELECT a, sum(b), max(c)
  FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
  GROUP BY a, window_end, c, window_start;

Modifikasi yang menyebabkan kompatibilitas sebagian

Modifikasi ini memungkinkan penerapan menggunakan kembali data state untuk metrik yang tidak berubah. Metrik yang baru ditambahkan mulai mengakumulasi dari nol saat penerapan di-restart; state metrik yang dihapus dibuang.

• Menambahkan metrik agregat non-distinct. Metrik baru mulai menghitung dari nol saat penerapan dimulai. Metrik yang sudah ada tidak terpengaruh.

• Menambah dan menghapus metrik agregat non-distinct secara bersamaan. Metrik yang ditambahkan mulai dari nol; state metrik yang dihapus dibuang; metrik yang tidak berubah dilanjutkan secara normal.

• Memodifikasi metrik agregat non-distinct (termasuk mengganti fungsi agregat atau logika perhitungan bidang inputnya). Metrik asli dianggap dihapus—state-nya dibuang. Metrik baru mulai dari nol.

Contoh berikut menggambarkan kasus-kasus tersebut (menambah, menghapus, dan memodifikasi metrik). Untuk penghapusan murni tanpa penambahan, lihat modifikasi kompatibel untuk menghapus metrik agregat non-distinct.

-- Asli
SELECT a, sum(b), max(c)
FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
GROUP BY a, window_start, window_end;

-- Tambah count(c): kompatibilitas sebagian.
-- sum(b) dan max(c) dilanjutkan dari state yang tersimpan. count(c) mulai dari 0.
SELECT a, sum(b), max(c), count(c)
FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
GROUP BY a, window_start, window_end;

-- Tambah count(c) dan hapus sum(b) secara bersamaan: kompatibilitas sebagian.
-- max(c) dilanjutkan dari state yang tersimpan. State sum(b) dibuang. count(c) mulai dari 0.
SELECT a, max(c), count(c)
FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
GROUP BY a, window_start, window_end;

-- Ubah max(c) menjadi min(c): kompatibilitas sebagian.
-- sum(b) dilanjutkan dari state yang tersimpan.
-- State max(c) dibuang. min(c) mulai dari 0.
SELECT a, sum(b), min(c)
FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
GROUP BY a, window_start, window_end;

Mengubah logika perhitungan bidang input juga dianggap sebagai modifikasi. Contoh berikut mengubah max(c) menjadi max(substring(c, 1, 5)) menggunakan temporary view. State max(c) asli dibuang, dan max(c) dalam kueri baru (di mana c adalah lima karakter pertama dari nilai asli) mulai dari nol.

-- Asli
SELECT a, sum(b), max(c)
FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
GROUP BY a, window_start, window_end;

-- Ubah logika bidang input melalui temporary view: kompatibilitas sebagian.
-- sum(b) dilanjutkan dari state yang tersimpan. State max(c) (asli) dibuang.
-- max(c) pada view baru mulai dari 0.
CREATE TEMPORARY VIEW MyView AS SELECT a, b, substring(c, 1, 5) AS c, ts FROM MyTable;
SELECT a, sum(b), max(c)
FROM TABLE(TUMBLE(TABLE MyView, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
GROUP BY a, window_start, window_end;

Modifikasi yang menyebabkan ketidakkompatibelan penuh

Modifikasi ini mengharuskan penerapan membuang seluruh data state. Penerapan dimulai ulang dari awal seolah-olah tidak ada state sebelumnya.

• Memodifikasi atribut jendela. Mengubah tipe jendela (misalnya, dari TUMBLE ke HOP), ukuran jendela, atau atribut waktu menyebabkan ketidakkompatibelan penuh.

  -- Asli
  SELECT a, sum(b), max(c), window_start
  FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
  GROUP BY a, window_start, window_end;
  
  -- Ubah tipe jendela menjadi HOP: tidak kompatibel
  SELECT a, sum(b), max(c),
    hop_start(ts, INTERVAL '1' MINUTE, INTERVAL '2' MINUTE) AS window_start
  FROM MyTable
  GROUP BY a, hop(ts, INTERVAL '1' MINUTE, INTERVAL '2' MINUTE);
  
  -- Ubah ukuran jendela dari 1 menit menjadi 2 menit: tidak kompatibel
  SELECT a, sum(b), max(c),
    tumble_start(ts, INTERVAL '2' MINUTE) AS window_start
  FROM MyTable
  GROUP BY a, tumble(ts, INTERVAL '2' MINUTE);
  
  -- Ubah atribut waktu dari ts ke proctime: tidak kompatibel
  SELECT a, sum(b), max(c),
    tumble_start(ts, INTERVAL '1' MINUTE) AS window_start
  FROM MyTable
  GROUP BY a, tumble(proctime, INTERVAL '1' MINUTE);

• Menambah, menghapus, atau memodifikasi kunci pengelompokan (termasuk mengubah logika perhitungan bidang kunci pengelompokan).

  -- Asli
  SELECT a, sum(b), max(c),
    tumble_start(ts, INTERVAL '1' MINUTE) AS window_start
  FROM MyTable
  GROUP BY a, tumble(ts, INTERVAL '1' MINUTE);
  
  -- Tambah kunci pengelompokan d: tidak kompatibel
  SELECT a, sum(b), max(c),
    tumble_start(ts, INTERVAL '1' MINUTE) AS window_start
  FROM MyTable
  GROUP BY a, d, tumble(ts, INTERVAL '1' MINUTE);

• Mengubah urutan kunci pengelompokan non-jendela. Mengubah urutan kunci pengelompokan yang tidak terkait fungsi jendela menyebabkan ketidakkompatibelan penuh.

  -- Asli
  SELECT a, sum(b), max(c)
  FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
  GROUP BY a, c, window_end, window_start;
  
  -- Tukar a dan c: tidak kompatibel
  SELECT a, sum(b), max(c)
  FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
  GROUP BY c, a, window_end, window_start;

• Menambah, menghapus, atau memodifikasi metrik agregat distinct. Metrik agregat distinct (dihitung menggunakan fungsi agregat distinct seperti COUNT(DISTINCT ...)) selalu menyebabkan ketidakkompatibelan penuh saat ditambah, dihapus, atau diubah.

  -- Asli
  SELECT a, sum(b), count(DISTINCT b), max(c), count(DISTINCT c),
    tumble_start(ts, INTERVAL '1' MINUTE) AS window_start
  FROM MyTable
  GROUP BY a, tumble(ts, INTERVAL '1' MINUTE);
  
  -- Hapus count(DISTINCT b): tidak kompatibel
  SELECT a, sum(b), max(c), count(DISTINCT c),
    tumble_start(ts, INTERVAL '1' MINUTE) AS window_start
  FROM MyTable
  GROUP BY a, tumble(ts, INTERVAL '1' MINUTE);

• Menghapus semua metrik agregat. Seluruh state dibuang dan tidak ada data state yang digunakan kembali.

  -- Asli
  SELECT a, sum(b), count(DISTINCT b), max(c), count(DISTINCT c),
    tumble_start(ts, INTERVAL '1' MINUTE) AS window_start
  FROM MyTable
  GROUP BY a, tumble(ts, INTERVAL '1' MINUTE);
  
  -- Hapus semua metrik agregat: tidak kompatibel
  SELECT a, tumble_start(ts, INTERVAL '1' MINUTE) AS window_start
  FROM MyTable
  GROUP BY a, tumble(ts, INTERVAL '1' MINUTE);

• Menambahkan metrik agregat ketika SQL asli tidak memiliki metrik agregat. Beralih dari nol metrik agregat ke satu atau lebih menyebabkan ketidakkompatibelan penuh.

  -- Asli (tidak ada metrik agregat)
  SELECT a, b
  FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
  GROUP BY a, b, window_end, window_start;
  
  -- Tambah count(a): tidak kompatibel
  SELECT a, b, count(a)
  FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
  GROUP BY a, b, window_end, window_start;

• Hanya ada satu metrik agregat dan logika perhitungannya berubah.

  -- Asli (satu metrik)
  INSERT INTO MySink SELECT a, sum(b)
  FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' SECOND))
  GROUP BY a, window_start, window_end;
  
  -- Ubah perhitungan input melalui temporary view: tidak kompatibel
  CREATE TEMPORARY VIEW MyView AS SELECT a, b + 1 AS b, ts FROM MyTable;
  INSERT INTO MySink SELECT a, sum(b)
  FROM TABLE(TUMBLE(TABLE MyView, DESCRIPTOR(ts), INTERVAL '1' SECOND))
  GROUP BY a, window_start, window_end;

• Himpunan metrik agregat sepenuhnya berbeda sebelum dan sesudah modifikasi.

  -- Asli
  INSERT INTO MySink SELECT a, sum(b)
  FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' SECOND))
  GROUP BY a, c, window_start, window_end;
  
  -- Ganti sum(b) dengan min(b): tidak kompatibel
  INSERT INTO MySink SELECT a, min(b)
  FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' SECOND))
  GROUP BY a, c, window_start, window_end;

• Menambah atau menghapus klausa perhitungan bidang antara fungsi jendela dan GROUP BY. Klausa perhitungan bidang mengubah window_start, window_end, atau window_time antara langkah TVF dan GROUP BY.

  -- Asli
  SELECT a, sum(b), max(c), window_start, window_end
  FROM (SELECT a, b, c, window_start, window_end
        FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE)))
  GROUP BY a, window_start, window_end;
  
  -- Tambah klausa perhitungan bidang: tidak kompatibel
  SELECT a, sum(b), max(c), window_start, window_end
  FROM (SELECT a, b, c,
               window_start + (INTERVAL '1' SECOND) AS window_start,
               window_end
        FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE)))
  GROUP BY a, window_start, window_end;

• Mengubah apakah GROUP BY berisi window_start dan window_end secara bersamaan. Jika klausa GROUP BY berubah dari menyertakan kedua bidang tersebut menjadi hanya salah satu (atau sebaliknya), penerapan menjadi tidak kompatibel.

  -- Asli (GROUP BY hanya berisi window_start)
  SELECT a, sum(b), max(c), window_start
  FROM (SELECT a, b, c, window_start
        FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE)))
  GROUP BY a, window_start;
  
  -- Tambah window_end ke GROUP BY: tidak kompatibel
  SELECT a, sum(b), max(c), window_start, window_end
  FROM (SELECT a, b, c, window_start, window_end
        FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE)))
  GROUP BY a, window_start, window_end;

• Menambah atau menghapus GROUPING SETS, CUBE, atau ROLLUP. Klausa-klausa ini mengelompokkan data secara terpisah berdasarkan window_start dan window_end, yang mengubah tata letak state internal.

  -- Asli
  SELECT a, sum(b), max(c), window_start, window_end
  FROM (SELECT a, b, c, window_start, window_end
        FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE)))
  GROUP BY a, window_start, window_end;
  
  -- Tambah GROUPING SETS: tidak kompatibel
  SELECT a, sum(b), max(c), window_start, window_end
  FROM (SELECT a, b, c, window_start, window_end
        FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE)))
  GROUP BY GROUPING SETS ((a), (window_start), (window_end));

Modifikasi dengan kompatibilitas tidak diketahui

Untuk pola-pola ini, mesin tidak dapat menentukan terlebih dahulu apakah data state kompatibel. Perlakukan modifikasi ini sebagai berpotensi tidak kompatibel dan validasi perilaku setelah restart.

• Klausa perhitungan bidang ada baik sebelum maupun sesudah modifikasi. Ketika klausa perhitungan bidang ada dalam SQL asli dan yang dimodifikasi (meskipun tidak berubah), kompatibilitasnya tidak diketahui.

• Klausa filter pada bidang jendela (window_start, window_end, window_time). Menambah, menghapus, atau mempertahankan filter pada bidang jendela semuanya menghasilkan kompatibilitas tidak diketahui.

  -- Asli
  SELECT a, sum(b), max(c), window_start, window_end
  FROM (SELECT a, b, c, window_start, window_end
        FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE)))
  GROUP BY a, window_start, window_end;
  
  -- Tambah klausa filter: kompatibilitas tidak diketahui
  SELECT a, sum(b), max(c), window_start, window_end
  FROM (SELECT a, b, c, window_start, window_end
        FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
        WHERE window_start >= TIMESTAMP '2024-04-15 08:06:00.000')
  GROUP BY a, window_start, window_end;

• Fungsi tabel yang didefinisikan pengguna (UDTF). Menambah, menghapus, atau mempertahankan UDTF dalam SQL semuanya menghasilkan kompatibilitas tidak diketahui.

  -- Asli
  SELECT a, sum(b), length(c), window_start, window_end
  FROM (SELECT a, b, c, window_start, window_end
        FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE)))
  GROUP BY a, c, window_start, window_end;
  
  -- Tambah UDTF: kompatibilitas tidak diketahui
  SELECT a, sum(b), length(c), window_start, window_end, c1, c2
  FROM (SELECT a, b, c, window_start, window_end, c1, c2
        FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE)),
             LATERAL TABLE(split(c)) AS T(c1, c2))
  GROUP BY a, c, window_start, window_end, c1, c2;

• Fungsi agregat yang didefinisikan pengguna Python (UDAF). Menambah, menghapus, atau mempertahankan UDAF Python semuanya menghasilkan kompatibilitas tidak diketahui.

  -- Asli
  SELECT a, sum(b), max(c), window_start
  FROM (SELECT a, b, c, window_start
        FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE)))
  GROUP BY a, window_start;
  
  -- Tambah UDAF Python: kompatibilitas tidak diketahui
  SELECT a, sum(b), c, window_start
  FROM (SELECT a, b, weighted_avg(c) AS c, window_start
        FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
        GROUP BY a, b, window_start)
  GROUP BY a, c, window_start;

• Agregasi pada window_start, window_end, atau window_time. Menerapkan fungsi agregat pada bidang jendela—menambah, menghapus, atau mempertahankannya—menghasilkan kompatibilitas tidak diketahui.

  -- Asli
  SELECT a, sum(b), max(c), window_start, window_end
  FROM (SELECT a, b, c, window_start, window_end
        FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE)))
  GROUP BY a, window_start, window_end;
  
  -- Tambah agregasi pada window_start: kompatibilitas tidak diketahui
  SELECT a, sum(b), max(c), MAX(window_start) AS ag, window_end
  FROM (SELECT a, b, c, window_start, window_end
        FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(ts), INTERVAL '1' MINUTE)))
  GROUP BY (a, window_start, window_end);

• GROUPING SETS, CUBE, atau ROLLUP ada baik sebelum maupun sesudah modifikasi. Jika klausa GROUP BY menggunakan GROUPING SETS, CUBE, atau ROLLUP baik sebelum maupun sesudah perubahan, kompatibilitasnya tidak diketahui.