Panduan Cepat Join Flink SQL - Realtime Compute for Apache Flink

Flink SQL mendukung join kompleks pada tabel dinamis menggunakan berbagai semantik kueri dan jenis join. Hindari membuat produk Kartesius karena tidak didukung dan dapat menyebabkan kegagalan kueri. Secara default, urutan join tidak dioptimalkan. Untuk performa lebih baik, cantumkan tabel dalam klausa FROM dari frekuensi pembaruan terendah ke tertinggi.

Ikhtisar join

Join type	Description	Constraints
Regular joins	Regular join adalah jenis join paling umum. Setiap catatan baru atau perubahan di salah satu sisi join akan terlihat dan memengaruhi seluruh hasil join.	Sintaks regular join fleksibel dan mendukung operasi INSERT, UPDATE, dan DELETE pada tabel input. Namun, regular join menyimpan data input dari kedua sisi dalam state-nya, yang dapat tumbuh tanpa batas. Untuk mencegah ukuran state berlebihan, Anda dapat mengatur waktu hidup (TTL) state, tetapi hal ini dapat mengurangi akurasi hasil Anda.
Interval joins	Interval join mengembalikan produk Kartesius dari catatan yang memenuhi kondisi join dan batasan waktu.	Memerlukan setidaknya satu kondisi join kesetaraan dan kondisi pembatas waktu yang berlaku untuk kedua sisi. Rentang waktu dapat didefinisikan sebagai satu kondisi (seperti <, <=, >=, atau >), menggunakan kondisi `BETWEEN` , atau sebagai kondisi kesetaraan pada atribut waktu dengan tipe yang sama (waktu pemrosesan atau event time) di kedua tabel.
Temporal joins	Temporal join menggabungkan aliran dengan versioned table. Join ini menggunakan event time atau waktu pemrosesan untuk mencocokkan data dengan versi tabel yang tepat pada titik waktu tertentu.	Memerlukan kedua tabel menggunakan semantik waktu yang sama (waktu pemrosesan atau event time). Perhatikan siklus hidup hasil join; kondisi join biasanya didasarkan pada timestamp tertentu.
Lookup joins	Lookup join biasanya digunakan untuk memperkaya tabel dengan data yang diambil dari sistem eksternal. Join ini memerlukan satu tabel memiliki atribut waktu pemrosesan dan tabel lainnya didukung oleh dimension table.	Memerlukan satu tabel memiliki atribut waktu pemrosesan, sedangkan tabel lainnya harus menggunakan konektor sumber lookup. Diperlukan juga kondisi join kesetaraan antara kedua tabel.
Lateral joins	Lateral join menghubungkan tabel dengan output dari fungsi bernilai tabel. Setiap baris dari tabel kiri digabungkan dengan semua baris yang dihasilkan oleh pemanggilan fungsi bernilai tabel yang sesuai.	Memerlukan kondisi join `ON` TRUE yang konstan dalam klausa, seperti `JOIN LATERAL TABLE(table_func(order_id)) t(res) ON TRUE`.

Regular joins

Empat jenis regular join yang umum adalah:

INNER JOIN: Mengembalikan catatan yang memenuhi kondisi join di kedua tabel (irisan).
LEFT JOIN: Mengembalikan semua catatan dari tabel kiri, bahkan jika tidak ada catatan yang cocok di tabel kanan (mempertahankan tabel kiri).
RIGHT JOIN: Mengembalikan semua catatan dari tabel kanan, bahkan jika tidak ada catatan yang cocok di tabel kiri (mempertahankan tabel kanan).
FULL OUTER JOIN: Mengembalikan gabungan kedua tabel, termasuk catatan yang cocok maupun tidak cocok.

Diagram join reguler

Contoh regular join

Masuk ke Realtime Compute for Apache Flink console.
Pada daftar ruang kerja, temukan ruang kerja target dan klik Console di kolom Actions.
Di panel navigasi sebelah kiri, klik Development > ETL.

Klik ikon , lalu klik New Blank Stream Draft. Masukkan name, pilih engine version, lalu klik Create.

Contoh ini menunjukkan cara menggunakan join untuk menghubungkan baris antar tabel, mengaitkan nama kode superhero dengan nama aslinya.

CREATE TEMPORARY TABLE NOC (
  agent_id STRING,
  codename STRING
)
WITH (
  'connector' = 'faker',   -- Konektor Faker menghasilkan data simulasi.
  'fields.agent_id.expression' = '#{regexify ''(1|2|3|4|5){1}''}',  -- Menghasilkan bilangan acak dari 1 hingga 5.
  'fields.codename.expression' = '#{superhero.name}',   -- Fungsi bawaan Faker yang menghasilkan nama superhero secara acak.
  'number-of-rows' = '10'   -- Menentukan bahwa 10 baris data dihasilkan.
);

CREATE TEMPORARY TABLE RealNames (
  agent_id STRING,
  name     STRING
)
WITH (
  'connector' = 'faker',
  'fields.agent_id.expression' = '#{regexify ''(1|2|3|4|5){1}''}',
  'fields.name.expression' = '#{Name.full_name}',  -- Fungsi bawaan Faker yang menghasilkan nama secara acak.
  'number-of-rows' = '10'
);

SELECT
    name,
    codename
FROM NOC
INNER JOIN RealNames ON NOC.agent_id = RealNames.agent_id;  -- Jika agent_id (1-5) di kedua tabel sama, nama dan nama kode dikembalikan.

Di pojok kanan atas, klik Debug, pilih kluster debug, lalu klik OK. Jika Anda belum memiliki session cluster, lihat Create a session cluster.
Setelah debugging selesai, hasilnya adalah tabel dengan dua kolom, name dan codename, yang menampilkan data keluaran Regular Join.

Untuk informasi lebih lanjut tentang penggunaan regular join, lihat Regular join statements.

Interval joins

Interval join menggabungkan catatan dari dua aliran yang berada dalam interval waktu tertentu. Jenis join ini biasanya digunakan untuk menghubungkan event dari dua aliran yang terjadi berdekatan dalam waktu.

Diagram interval join

Contoh interval join

Contoh ini menggabungkan pesanan dengan pengiriman. Hasilnya difilter hanya untuk pesanan yang dikirim dalam waktu tiga jam setelah ditempatkan.

Buat draft ETL seperti yang dijelaskan di Regular joins.

CREATE TEMPORARY TABLE orders (
  id INT,
  order_time AS TIMESTAMPADD(HOUR, CAST(FLOOR(RAND()*(1-5+1)+5)*(-1) AS INT), CURRENT_TIMESTAMP)    -- Mendapatkan waktu acak dari 2, 3, atau 4 jam yang lalu berdasarkan waktu lokal.
)
WITH (
  'connector' = 'datagen',     -- Konektor datagen dapat menghasilkan data acak secara berkala.
  'rows-per-second'='10',      -- Laju pembuatan data acak, 10 baris/detik.
  'fields.id.kind'='sequence', -- Generator urutan.
  'fields.id.start'='1',       -- Urutan dimulai dari 1.
  'fields.id.end'='100'        -- Urutan berakhir di 100.
);

CREATE TEMPORARY TABLE shipments (
  order_id INT,
  shipment_time AS TIMESTAMPADD(HOUR, CAST(FLOOR(RAND()*(1-5+1))+1 AS INT), CURRENT_TIMESTAMP)   -- Mendapatkan waktu acak dari 0, 1, atau 2 jam yang lalu berdasarkan waktu lokal.
)
WITH (
  'connector' = 'datagen',
  'rows-per-second'='5',
  'fields.order_id.kind'='sequence',
  'fields.order_id.start'='1',
  'fields.order_id.end'='100'
);

SELECT
  o.id AS order_id,
  o.order_time,
  s.shipment_time,
  TIMESTAMPDIFF(HOUR,o.order_time,s.shipment_time) AS hour_diff    -- Selisih waktu antara order_time dan shipment_time.
FROM orders o
JOIN shipments s ON o.id = s.order_id   
WHERE 
    o.order_time BETWEEN s.shipment_time - INTERVAL '3' HOUR AND s.shipment_time;  -- Memfilter pesanan yang ditempatkan dalam waktu tiga jam sebelum waktu pengiriman.

Di pojok kanan atas, klik Debug lalu klik OK. Hasil debug adalah sebagai berikut.

Untuk informasi lebih lanjut tentang interval join, lihat Interval join statements.

Temporal joins

Temporal join digunakan dengan tabel dinamis, yang isinya berubah seiring waktu. Join ini menggabungkan aliran event dengan versi tabel dinamis yang berlaku pada saat setiap event terjadi.

Diagram temporal join

Contoh temporal join

Contoh ini menunjukkan skenario bisnis di mana nilai tukar berubah dalam periode waktu berbeda, dan pesanan harus dihitung menggunakan nilai tukar yang berlaku pada saat transaksi.

(Opsional) Hasilkan data simulasi

Buat draft ETL seperti yang dijelaskan di Regular joins.

Gunakan konektor Faker untuk menghasilkan data simulasi dan tulis ke Kafka sebagai tabel nilai tukar dinamis menggunakan konektor Upsert Kafka.

CREATE TEMPORARY TABLE currency_rates (
  `currency_code` STRING,
  `eur_rate` DECIMAL(6,4),
  `rate_time` TIMESTAMP(3),
  WATERMARK FOR `rate_time` AS rate_time - INTERVAL '15' SECOND,
  PRIMARY KEY (currency_code) NOT ENFORCED
) WITH (
  'connector' = 'upsert-kafka',
  'topic' = 'currency_rates',
  'properties.bootstrap.servers' = '${secret_values.kafkahost}',
  'key.format' = 'raw',
  'value.format' = 'json'
);

CREATE TEMPORARY TABLE transactions (
  `id` STRING,
  `currency_code` STRING,
  `total` DECIMAL(10,2),
  `transaction_time` TIMESTAMP(3),
  WATERMARK FOR `transaction_time` AS transaction_time - INTERVAL '30' SECOND
) WITH (
  'connector' = 'kafka',
  'topic' = 'transactions',
  'properties.bootstrap.servers' = '${secret_values.kafkahost}',
  'key.format' = 'raw',
  'key.fields' = 'id',
  'value.format' = 'json'
);

CREATE TEMPORARY TABLE currency_rates_faker (
  `currency_code` STRING,
  `eur_rate` DECIMAL(6,4),
  `rate_time` TIMESTAMP(3)
)
WITH (
  'connector' = 'faker',
  'fields.currency_code.expression' = '#{Currency.code}',
  'fields.eur_rate.expression' = '#{Number.randomDouble ''4'',''0'',''10''}',
  'fields.rate_time.expression' = '#{date.past ''15'',''SECONDS''}',
  'rows-per-second' = '2'
);

CREATE TEMPORARY TABLE transactions_faker (
  `id` STRING,
  `currency_code` STRING,
  `total` DECIMAL(10,2),
  `transaction_time` TIMESTAMP(3)
)
WITH (
  'connector' = 'faker',
  'fields.id.expression' = '#{Internet.UUID}',
  'fields.currency_code.expression' = '#{Currency.code}',
  'fields.total.expression' = '#{Number.randomDouble ''2'',''10'',''1000''}',
  'fields.transaction_time.expression' = '#{date.past ''30'',''SECONDS''}',
  'rows-per-second' = '2'
);

BEGIN STATEMENT SET;

INSERT INTO currency_rates
SELECT * FROM currency_rates_faker;

INSERT INTO transactions
SELECT * FROM transactions_faker;

END;

Di pojok kanan atas, klik Deploy untuk menyebarluaskan pekerjaan.
Di panel navigasi sebelah kiri, klik O&M > Deployments. Temukan pekerjaan target, klik Start di kolom Actions, pilih Initial Mode, lalu klik Start.

Buat draft ETL baru seperti yang dijelaskan di bagian Regular joins untuk membaca data simulasi guna debugging.

CREATE TEMPORARY TABLE currency_rates (
  `currency_code` STRING,
  `eur_rate` DECIMAL(6,4),
  `rate_time` TIMESTAMP(3),
  WATERMARK FOR `rate_time` AS rate_time - INTERVAL '15' SECOND,
  PRIMARY KEY (currency_code) NOT ENFORCED
) WITH (
  'connector' = 'upsert-kafka',
  'topic' = 'currency_rates',
  'properties.bootstrap.servers' = '${secret_values.kafkahost}',
  'properties.auto.offset.reset' = 'earliest',
  'properties.group.id' = 'currency_rates',
  'key.format' = 'raw',
  'value.format' = 'json'
);

CREATE TEMPORARY TABLE transactions (
  `id` STRING,
  `currency_code` STRING,
  `total` DECIMAL(10,2),
  `transaction_time` TIMESTAMP(3),
  WATERMARK FOR `transaction_time` AS transaction_time - INTERVAL '30' SECOND
) WITH (
  'connector' = 'kafka',
  'topic' = 'transactions',
  'properties.bootstrap.servers' = '${secret_values.kafkahost}',
  'properties.auto.offset.reset' = 'earliest',
  'properties.group.id' = 'transactions',
  'key.format' = 'raw',
  'key.fields' = 'id',
  'value.format' = 'json'
);

SELECT
  t.id,
  t.total * c.eur_rate AS total_eur,
  c.eur_rate,
  t.total,
  c.currency_code,
  c.rate_time,
  t.transaction_time
FROM transactions t
JOIN currency_rates FOR SYSTEM_TIME AS OF t.transaction_time AS c
ON t.currency_code = c.currency_code 
;

Di pojok kanan atas, klik Debug lalu klik OK.

Nilai tukar berubah dua kali, pada 20:16:11 dan 20:35:22. Transaksi terjadi pada 20:35:14. Pada saat itu, nilai tukar belum diperbarui. Oleh karena itu, perhitungan untuk transaksi ini harus menggunakan nilai tukar yang berlaku sejak 20:16:11.

Lookup joins

Lookup join memperkaya aliran data dengan data statis atau data yang berubah lambat dari sistem eksternal. Misalnya, Anda dapat menggabungkan aliran pesanan real-time dengan data referensi, seperti informasi produk yang disimpan dalam database relasional. Join ini memerlukan satu tabel memiliki atribut waktu pemrosesan dan tabel lainnya terhubung melalui konektor lookup, seperti konektor MySQL.

Diagram lookup join

Catatan

Anda harus menyertakan FOR SYSTEM_TIME AS OF PROCTIME() untuk menggabungkan setiap catatan aliran dengan snapshot tabel dimensi pada waktu tersebut. Jika data tabel dimensi berubah nanti, hasil join tidak terpengaruh.
Kondisi ON harus mencakup kondisi kesetaraan pada bidang yang dapat digunakan tabel dimensi untuk pencarian.

Contoh lookup join

Contoh ini menunjukkan cara memperkaya data pesanan dengan data statis dari konektor eksternal untuk menambahkan nama produk.

Buat draft ETL seperti yang dijelaskan di Regular joins.

CREATE TEMPORARY TABLE orders ( 
    order_id STRING,
    product_id INT,
    order_total INT
) WITH (
  'connector' = 'faker',   -- Konektor Faker menghasilkan data simulasi.
  'fields.order_id.expression' = '#{Internet.uuid}',   -- Menghasilkan UUID acak.
  'fields.product_id.expression' = '#{number.numberBetween ''1'',''5''}',   -- Menghasilkan bilangan acak dari 1 hingga 5.
  'fields.order_total.expression' = '#{number.numberBetween ''1000'',''5000''}',  -- Menghasilkan bilangan acak dari 1000 hingga 5000.
  'number-of-rows' = '10'  -- Jumlah baris data yang dihasilkan.
);

-- Terhubung ke data produk statis di MySQL.
CREATE TEMPORARY TABLE products (
 product_id INT,
 product_name STRING
)
WITH(
  'connector' = 'mysql',
  'hostname' = '${secret_values.mysqlhost}',
  'port' = '3306',
  'username' = '${secret_values.username}',
  'password' = '${secret_values.password}',
  'database-name' = 'db2024',
  'table-name' = 'products'
);

SELECT 
    o.order_id,
    p.product_name,
    o.order_total,
  CASE 
    WHEN o.order_total > 3000 THEN 1    
    ELSE 0
  END AS is_importance          -- Menambahkan bidang is_importance. Nilainya 1 jika total pesanan melebihi 3000, menandakan pesanan penting.
FROM orders o
JOIN products FOR SYSTEM_TIME AS OF PROCTIME() AS p -- Klausul FOR SYSTEM_TIME AS OF PROCTIME() memastikan bahwa saat operator join memproses baris dari `orders`, setiap baris dari `orders` digabungkan dengan baris `products` yang memenuhi kondisi join.
ON o.product_id = p.product_id;

Tabel products berisi 5 catatan contoh: 1-cola, 2-orange, 3-milk, 4-tea, 5-coffee.

Di pojok kanan atas, klik Debug lalu klik OK. Hasil debug adalah sebagai berikut.

Untuk informasi lebih lanjut tentang lookup join, lihat JOIN statements for dimension tables.

Lateral joins

Lateral join menggunakan subkueri dalam klausa FROM yang dieksekusi untuk setiap baris dari query luar. Ini dapat meningkatkan fleksibilitas dan performa kueri dengan mengurangi pemindaian tabel. Namun, operasi ini dapat menyebabkan penurunan performa jika kueri dalam kompleks atau memproses data dalam jumlah besar.

Contoh lateral join

Contoh ini mengagregasi catatan penjualan untuk menemukan tiga produk teratas berdasarkan total kuantitas penjualan.

Buat draft ETL seperti yang dijelaskan di Regular joins.

CREATE TEMPORARY TABLE sale (
  sale_id STRING,
  product_id INT,
  sale_num INT
)
WITH (
  'connector' = 'faker',    -- Konektor Faker menghasilkan data simulasi.
  'fields.sale_id.expression' = '#{Internet.uuid}',   -- Menghasilkan UUID acak.
  'fields.product_id.expression' = '#{regexify ''(1|2|3|4|5){1}''}',   -- Menghasilkan bilangan acak dari 1 hingga 5.
  'fields.sale_num.expression' = '#{number.numberBetween ''1'',''10''}',  -- Menghasilkan bilangan bulat acak dari 1 hingga 10.
  'number-of-rows' = '50'   -- Menghasilkan 50 baris data.
);

CREATE TEMPORARY TABLE products (
 product_id INT,
 product_name STRING,
 PRIMARY KEY(product_id) NOT ENFORCED
)
WITH(
  'connector' = 'mysql',
  'hostname' = '${secret_values.mysqlhost}',
  'port' = '3306',
  'username' = '${secret_values.username}',
  'password' = '${secret_values.password}',
  'database-name' = 'db2024',
  'table-name' = 'products'
);

SELECT 
    p.product_name,
    s.total_sales
FROM  products p
LEFT JOIN LATERAL
    (SELECT SUM(sale_num) AS total_sales FROM sale WHERE sale.product_id = p.product_id) s ON TRUE
    ORDER BY total_sales DESC
    LIMIT 3;

Di pojok kanan atas, klik Debug lalu klik OK. Hasil debug adalah sebagai berikut.

Dokumen terkait

Jika Anda hanya mempertimbangkan waktu pemrosesan event dan bukan event time-nya, Anda dapat menggunakan processing-time temporal join. Untuk informasi lebih lanjut, lihat Processing-time temporal join statements.
Untuk informasi lebih lanjut tentang penggunaan regular join, lihat Regular join statements.
Untuk informasi lebih lanjut tentang penggunaan interval join, lihat Interval join statements.
Untuk informasi lebih lanjut tentang penggunaan lookup join, lihat JOIN statements for dimension tables.