Belajar Apache Flink - Table API & SQL

#Pendahuluan

Empat episode terakhir berkutat di DataStream API yang imperatif. Episode 9 ini menawarkan rute yang jauh lebih singkat: Flink SQL. Dengan SQL, kalian menulis query streaming — lengkap dengan windowing, join, dan agregasi — tanpa menulis satu baris pun logika operator manual. Untuk analitik, ini biasanya satu-satunya API yang kalian butuhkan.

Kita akan mengenal TableEnvironment dan catalog, menulis DDL untuk mendeklarasikan source dan sink, lalu menerapkan windowing, joins, dan aggregations. Di akhir, kita menyinggung temporal tables dan pola CDC — bagaimana Flink melihat tabel database yang terus berubah sebagai stream.

#TableEnvironment dan Catalog

#Environment untuk Executable

TableEnvironment adalah titik masuk semua query. Ada dua mode: streaming (default untuk Flink) dan batch:

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;
 
EnvironmentSettings settings = EnvironmentSettings.inStreamingMode();
TableEnvironment tableEnv = TableEnvironment.create(settings);
 
tableEnv.executeSql("CREATE TABLE orders (user_id STRING, amount BIGINT, event_ts TIMESTAMP(3))");
tableEnv.executeSql("INSERT INTO agg_orders SELECT user_id, SUM(amount) FROM orders ...");

tableEnv.executeSql menjalankan pernyataan DDL, DML, dan query sekaligus. Ini pola utama program Table API: deklarasikan, jalankan, selesai.

#Catalog untuk Metadata Terpusat

Catalog menyimpan metadata tabel agar query bisa ditulis tanpa mengulang DDL di setiap job. Flink punya built-in catalog (misalnya untuk Hive) dan generic JDBC catalog. Dengan catalog, tim berbagi definisi tabel yang sama — sumber kebenaran tunggal.

CREATE CATALOG orders_catalog WITH (
  'type' = 'generic_in_memory',
  'default-database' = 'analytics'
);
USE CATALOG orders_catalog;

#DDL dan DML dengan Flink SQL

#CREATE TABLE untuk Source dan Sink

DDL di Flink SQL menggabungkan skema data dan konfigurasi konektor dalam satu pernyataan:

CREATE TABLE orders (
  user_id  STRING,
  amount   BIGINT,
  event_ts TIMESTAMP(3),
  WATERMARK FOR event_ts AS event_ts - INTERVAL '5' SECOND
) WITH (
  'connector' = 'kafka',
  'topic' = 'orders',
  'properties.bootstrap.servers' = 'localhost:9092',
  'format' = 'json'
);

Perhatikan klausa WATERMARK FOR event_ts — konsep watermark dari episode 5 diekspresikan langsung di DDL. Tanpa baris ini, windowing event time tidak berjalan.

#INSERT INTO untuk Menulis Hasil

Hasil query ditulis ke tabel sink dengan INSERT:

INSERT INTO agg_orders
SELECT user_id, SUM(amount) AS total
FROM orders
GROUP BY user_id, TUMBLE(event_ts, INTERVAL '5' MINUTE);

#Streaming SQL Patterns

#Windowing dengan Fungsi Bawaan

Flink SQL menyediakan fungsi window TUMBLE, HOP, dan SESSION yang langsung dipakai di GROUP BY:

SELECT user_id, COUNT(*) AS jumlah, SUM(amount) AS total
FROM orders
GROUP BY user_id, TUMBLE(event_ts, INTERVAL '1' MINUTE);
 
SELECT user_id, COUNT(*) AS aktivitas
FROM clicks
GROUP BY user_id, SESSION(event_ts, INTERVAL '30' MINUTE);

TUMBLE(event_ts, INTERVAL '1' MINUTE) adalah tumbling window satu menit, dan SESSION mengelompokkan event berdasarkan gap aktivitas. Sintaks windowing ini jauh lebih ringkas daripada DataStream API.

#Joins antar Stream

SQL juga menyederhanakan join antar stream:

SELECT o.order_id, o.amount, p.status
FROM orders o
JOIN payments p
  ON o.order_id = p.order_id
  AND o.event_ts BETWEEN p.event_ts - INTERVAL '5' MINUTE
  AND p.event_ts + INTERVAL '5' MINUTE;

Pattern interval join seperti di atas membatasi kecocokan pada jendela waktu tertentu — penting karena stream tidak punya batas yang jelas.

#Temporal Tables dan CDC

Temporal table adalah tabel yang isinya berubah mengikuti waktu — ideal untuk data dimensi. CDC (Change Data Capture) memungkinkan Flink membaca perubahan database melalui Debezium:

CREATE TABLE users (
  id   INT PRIMARY KEY NOT ENFORCED,
  nama STRING,
  updated_at TIMESTAMP(3)
) WITH (
  'connector' = 'mysql-cdc',
  'hostname' = 'localhost',
  'port' = '3306',
  'username' = 'flink',
  'password' = 'secret',
  'database-name' = 'app',
  'table-name' = 'users'
);

Connector mysql-cdc mengubah setiap perubahan baris di MySQL menjadi event stream — tabel "hidup" yang terus diperbarui tanpa polling manual. Ini fondasi arsitektur change-data-capture modern.

#Menggunakan SQL Client

Untuk eksperimen cepat tanpa menulis kode Java, gunakan SQL client bawaan Flink:

$FLINK_HOME/bin/sql-client.sh

Di dalam prompt SQL client, kalian bisa menulis DDL dan query secara interaktif. Perintah sql-client.sh adalah alat paling cepat untuk memvalidasi query sebelum memasukkannya ke aplikasi.

#Penutup

Episode 9 memperkenalkan jalur deklaratif Flink: TableEnvironment dan catalog untuk metadata, DDL yang menggabungkan skema dan konektor, fungsi window TUMBLE, HOP, dan SESSION, join antar stream, serta temporal tables dan CDC untuk data yang berubah-ubah. Flink SQL membuat analitik streaming bisa ditulis dalam hitungan baris.

Inti yang harus dibawa pulang:

• TableEnvironment adalah titik masuk semua program Table API dan SQL.
• Catalog menyimpan definisi tabel agar bisa dipakai ulang lintas job.
• Watermark dideklarasikan di DDL dengan klausa WATERMARK FOR.
• TUMBLE, HOP, dan SESSION mengekspresikan windowing langsung di SQL.
• Connector mysql-cdc menjadikan perubahan database sebagai stream event.

Di episode 10 selanjutnya kita akan membahas complex event processing (CEP) — mendefinisikan pattern dengan library CEP Flink, mencocokkan event berurutan, menerapkannya untuk fraud detection dan anomaly detection, serta menangani timed patterns dan pattern states. Di sinilah Flink menjadi "otak" yang mendeteksi pola tersembunyi dalam aliran data.