Lookup join - Realtime Compute for Apache Flink - Alibaba Cloud Documentation Center

Dokumen ini menjelaskan cara melakukan penggabungan tabel dimensi.

Penggabungan tabel dimensi (lookup join)

Lookup join merupakan implementasi khusus dari penggabungan tabel dimensi. Untuk informasi lebih lanjut, lihat Dimension Table JOIN Statement.
Di Fluss, hanya tabel kunci primer yang dapat digunakan sebagai tabel dimensi. Kondisi penggabungan harus mencakup semua kolom kunci primer dari tabel dimensi. Jika Anda menetapkan bucket key sebagai kunci gabungan saat membuat tabel, kondisi penggabungan juga dapat berupa subset awalan dari kunci primer. Untuk informasi lebih lanjut, lihat Prefix lookup berdasarkan bucket key.
Jika tabel dimensi adalah tabel kunci primer terpartisi, kondisi penggabungan juga harus mencakup kunci partisi.
Secara default, penggabungan tabel dimensi di Fluss menggunakan mode asinkron untuk throughput yang lebih tinggi. Anda dapat beralih ke mode sinkron dengan mengatur petunjuk SQL 'lookup.async' = 'false'.

Contoh SQL untuk menggabungkan tabel dimensi

Buat dua tabel dimensi

USE CATALOG `fluss_catalog`;
USE `my_db`;

CREATE TABLE `orders` (
  `o_orderkey` INT NOT NULL,
  `o_custkey` INT NOT NULL,
  `o_orderstatus` CHAR(1) NOT NULL,
  `o_totalprice` DECIMAL(15, 2) NOT NULL,
  `o_orderdate` DATE NOT NULL,
  `o_orderpriority` CHAR(15) NOT NULL,
  `o_clerk` CHAR(15) NOT NULL,
  `o_shippriority` INT NOT NULL,
  `o_comment` STRING NOT NULL,
  `o_dt` STRING NOT NULL,
  PRIMARY KEY (o_orderkey) NOT ENFORCED
);

CREATE TABLE `customer` (
  `c_custkey` INT NOT NULL,
  `c_name` STRING NOT NULL,
  `c_address` STRING NOT NULL,
  `c_nationkey` INT NOT NULL,
  `c_phone` CHAR(15) NOT NULL,
  `c_acctbal` DECIMAL(15, 2) NOT NULL,
  `c_mktsegment` CHAR(10) NOT NULL,
  `c_comment` STRING NOT NULL,
  PRIMARY KEY (c_custkey) NOT ENFORCED
);

Gabungkan kedua tabel dimensi

CREATE TEMPORARY TABLE `lookup_join_sink` (
  `order_key`        INT NOT NULL,
  `order_totalprice` DECIMAL(15, 2) NOT NULL,
  `customer_name`    STRING NOT NULL,
  `customer_address` STRING NOT NULL
) WITH ('connector' = 'blackhole');

-- Mode asinkron (default)
INSERT INTO `lookup_join_sink`
SELECT `o`.`o_orderkey`, `o`.`o_totalprice`, `c`.`c_name`, `c`.`c_address`
FROM (SELECT `orders`.*, PROCTIME() AS `ptime` FROM `orders`) AS `o`
LEFT JOIN `customer` FOR SYSTEM_TIME AS OF `o`.`ptime` AS `c`
ON `o`.`o_custkey` = `c`.`c_custkey`;

Gabungkan dalam mode sinkron

-- Mode sinkron
INSERT INTO `lookup_join_sink`
SELECT `o`.`o_orderkey`, `o`.`o_totalprice`, `c`.`c_name`, `c`.`c_address`
FROM (SELECT `orders`.*, PROCTIME() AS `ptime` FROM `orders`) AS `o`
LEFT JOIN `customer` /*+ OPTIONS('lookup.async' = 'false') */ FOR SYSTEM_TIME AS OF `o`.`ptime` AS `c`
ON `o`.`o_custkey` = `c`.`c_custkey`;

Gabungkan tabel dimensi terpartisi

Jika tabel dimensi adalah tabel kunci primer terpartisi, kondisi penggabungan harus mencakup semua kolom kunci primer, yaitu kunci primer dan kunci partisi.

-- Kunci primer: (c_custkey, dt). Kunci partisi: dt
CREATE TABLE `customer_partitioned` (
  `c_custkey` INT NOT NULL,
  `c_name` STRING NOT NULL,
  `c_address` STRING NOT NULL,
  `c_nationkey` INT NOT NULL,
  `c_phone` CHAR(15) NOT NULL,
  `c_acctbal` DECIMAL(15, 2) NOT NULL,
  `c_mktsegment` CHAR(10) NOT NULL,
  `c_comment` STRING NOT NULL,
  `dt` STRING NOT NULL,
  PRIMARY KEY (`c_custkey`, `dt`) NOT ENFORCED
)
PARTITIONED BY (`dt`)
WITH (
  'table.auto-partition.enabled' = 'true',
  'table.auto-partition.time-unit' = 'year'
);

Selain kunci primer, kondisi penggabungan juga harus mencakup kunci partisi.

INSERT INTO `lookup_join_sink`
SELECT `o`.`o_orderkey`, `o`.`o_totalprice`, `c`.`c_name`, `c`.`c_address`
FROM (SELECT `orders`.*, PROCTIME() AS `ptime` FROM `orders`) AS `o`
LEFT JOIN `customer_partitioned` FOR SYSTEM_TIME AS OF `o`.`ptime` AS `c`
ON `o`.`o_custkey` = `c`.`c_custkey` AND `o`.`o_dt` = `c`.`dt`;

Untuk informasi lebih lanjut tentang tabel terpartisi, lihat Data partitioning.

Prefix lookup berdasarkan bucket key

Ikhtisar

Ketika tabel kunci primer digunakan sebagai tabel dimensi, jika Anda menetapkan bucket key sebagai kunci gabungan saat membuat tabel, kondisi penggabungan dapat berupa subset awalan dari kunci primer tabel dimensi, bukan seluruh kunci primer. Ini disebut prefix lookup.

Prefix lookup secara default menggunakan mode asinkron untuk throughput yang lebih tinggi. Anda dapat beralih ke mode sinkron dengan mengatur petunjuk SQL 'lookup.async' = 'false'.

Batasan

Hanya tabel kunci primer Fluss yang dapat digunakan sebagai tabel dimensi untuk prefix lookup.
Anda harus menetapkan bucket key sebagai kunci gabungan saat membuat tabel. Untuk informasi lebih lanjut tentang cara menetapkan bucket key, lihat Data bucketing.
Kondisi penggabungan harus berupa subset awalan dari kunci primer tabel dimensi. Untuk tabel kunci primer terpartisi, kondisi penggabungan adalah awalan dari kunci primer (tidak termasuk kunci partisi) ditambah kunci partisi.
Prefix lookup tidak dapat digunakan bersamaan dengan lookup.insert-if-not-exists.

Contoh

Tabel terpartisi

Jika tabel dimensi adalah tabel kunci primer terpartisi, kondisi penggabungan adalah awalan dari kunci primer (tidak termasuk kunci partisi) ditambah kunci partisi. Pada contoh berikut, kunci primer tabel dimensi adalah (c_custkey, c_nationkey, dt), bucket key adalah c_custkey, dan kunci partisi adalah dt.

-- Kunci primer: (c_custkey, c_nationkey, dt). Bucket key: c_custkey. Kunci partisi: dt
CREATE TABLE `customer_partitioned_with_bucket_key` (
  `c_custkey` INT NOT NULL,
  `c_name` STRING NOT NULL,
  `c_address` STRING NOT NULL,
  `c_nationkey` INT NOT NULL,
  `c_phone` CHAR(15) NOT NULL,
  `c_acctbal` DECIMAL(15, 2) NOT NULL,
  `c_mktsegment` CHAR(10) NOT NULL,
  `c_comment` STRING NOT NULL,
  `dt` STRING NOT NULL,
  PRIMARY KEY (`c_custkey`, `c_nationkey`, `dt`) NOT ENFORCED
)
PARTITIONED BY (`dt`)
WITH (
  'bucket.key' = 'c_custkey',
  'table.auto-partition.enabled' = 'true',
  'table.auto-partition.time-unit' = 'year'
);

-- Kondisi penggabungan adalah awalan dari kunci primer (tidak termasuk kunci partisi) ditambah kunci partisi
INSERT INTO `lookup_join_sink`
SELECT `o`.`o_orderkey`, `o`.`o_totalprice`, `c`.`c_name`, `c`.`c_address`
FROM (SELECT `orders`.*, PROCTIME() AS `ptime` FROM `orders`) AS `o`
LEFT JOIN `customer_partitioned_with_bucket_key` FOR SYSTEM_TIME AS OF `o`.`ptime` AS `c`
ON `o`.`o_custkey` = `c`.`c_custkey` AND `o`.`o_dt` = `c`.`dt`;

Tabel tidak terpartisi

Pada contoh berikut, kunci primer tabel dimensi adalah (c_custkey, c_nationkey), bucket key adalah c_custkey, dan kondisi penggabungan hanya menggunakan awalan kunci primer c_custkey.

-- Kunci primer: (c_custkey, c_nationkey). Bucket key: c_custkey
CREATE TABLE `customer_with_bucket_key` (
  `c_custkey` INT NOT NULL,
  `c_name` STRING NOT NULL,
  `c_address` STRING NOT NULL,
  `c_nationkey` INT NOT NULL,
  `c_phone` CHAR(15) NOT NULL,
  `c_acctbal` DECIMAL(15, 2) NOT NULL,
  `c_mktsegment` CHAR(10) NOT NULL,
  `c_comment` STRING NOT NULL,
  PRIMARY KEY (`c_custkey`, `c_nationkey`) NOT ENFORCED
) WITH (
  'bucket.key' = 'c_custkey'
);

-- Prefix lookup dalam mode asinkron (default)
INSERT INTO `lookup_join_sink`
SELECT `o`.`o_orderkey`, `o`.`o_totalprice`, `c`.`c_name`, `c`.`c_address`
FROM (SELECT `orders`.*, PROCTIME() AS `ptime` FROM `orders`) AS `o`
LEFT JOIN `customer_with_bucket_key` FOR SYSTEM_TIME AS OF `o`.`ptime` AS `c`
ON `o`.`o_custkey` = `c`.`c_custkey`;

-- Prefix lookup dalam mode sinkron
INSERT INTO `lookup_join_sink`
SELECT `o`.`o_orderkey`, `o`.`o_totalprice`, `c`.`c_name`, `c`.`c_address`
FROM (SELECT `orders`.*, PROCTIME() AS `ptime` FROM `orders`) AS `o`
LEFT JOIN `customer_with_bucket_key` /*+ OPTIONS('lookup.async' = 'false') */ FOR SYSTEM_TIME AS OF `o`.`ptime` AS `c`
ON `o`.`o_custkey` = `c`.`c_custkey`;

Historical partition lookup

Ikhtisar

Auto partitioning menghapus partisi yang kedaluwarsa berdasarkan kebijakan retensi. Setelah suatu partisi dihapus, lookup join yang mereferensikan partisi tersebut tidak lagi dapat menemukan barisnya, meskipun datanya telah ditier ke Paimon.

Setelah historical partition lookup diaktifkan, lookup kunci primer akan fallback ke Paimon ketika partisi aslinya tidak lagi tersedia. Untuk mengaktifkan fitur ini, atur table.datalake.historical-partition.enabled pada tabel dimensi.

Batasan

Fitur ini dinonaktifkan secara default. Aktifkan dengan mengatur table.datalake.historical-partition.enabled pada tabel dimensi.
Hanya tabel kunci primer Paimon dengan auto partitioning yang diaktifkan dan tepat satu kunci partisi yang didukung.
Setelah fitur diaktifkan, Coordinator membuat dan menyimpan partisi sistem __historical__ untuk mengarahkan lookup ke Paimon. Menonaktifkan opsi ini akan menghapus partisi sistem tersebut.
Klien lookup menggunakan konfigurasi tabel yang diambil saat lookuper dibuat untuk memutuskan apakah akan fallback atau tidak. Setelah Anda mengubah konfigurasi, restart pekerjaan lookup yang perlu mengkueri partisi historis.

Contoh

Aktifkan historical partition lookup pada tabel dimensi.

ALTER TABLE `customer_partitioned_with_bucket_key` SET (
  'table.datalake.historical-partition.enabled' = 'true'
);

Untuk informasi lebih lanjut tentang parameter tersebut, lihat Connector parameters.

Insert if not exists

Ikhtisar

Saat lookup join, jika kunci lookup tidak ditemukan di tabel dimensi, penggabungan dilewati secara default. Untuk LEFT JOIN, kolom dari tabel dimensi bernilai NULL. Ketika opsi lookup.insert-if-not-exists diaktifkan dan terjadi lookup miss, Fluss secara otomatis menyisipkan baris baru. Baris ini menggunakan kunci lookup sebagai kunci primernya dan dikembalikan sebagai hasil penggabungan.

Fitur ini sangat berguna untuk membangun tabel kamus secara otomatis (yang harus berupa tabel kunci primer) selama pemrosesan aliran. Kasus penggunaan umumnya adalah memetakan identifier string dengan kardinalitas tinggi, seperti user ID atau device ID, ke ID integer ringkas untuk pemrosesan efisien oleh Aggregation Merge Engine di downstream.

Batasan

Hanya lookup kunci primer yang didukung; prefix lookup tidak didukung.
Tabel dimensi tidak boleh memiliki kolom non-null selain kolom kunci primer dan kolom auto-increment, karena Fluss tidak dapat mengisi nilainya selama penyisipan otomatis.
Aktifkan fitur ini dengan petunjuk SQL berikut: /*+ OPTIONS('lookup.insert-if-not-exists' = 'true') */

Contoh

Contoh berikut menunjukkan cara membangun tabel kamus UID secara otomatis dalam lookup join.

Buat tabel kamus dengan kolom auto-increment.

CREATE TABLE uid_mapping (
  uid VARCHAR NOT NULL,
  uid_int32 INT,PRIMARY KEY (uid) NOT ENFORCED
) WITH ('auto-increment.fields' = 'uid_int32','bucket.num' = '1');

Lakukan lookup join dengan opsi insert-if-not-exists. Saat uid muncul pertama kali, Fluss secara otomatis menyisipkannya ke tabel uid_mapping dan memberikan nilai uid_int32 yang di-auto-increment.

-- Mendaftarkan UID dari tabel aliran ods_events ke tabel kamus uid_mapping
-- dan mengambil ID integer yang sesuai (uid_int32).
SELECT
  ods.country,
  ods.prov,
  ods.city,
  ods.ymd,
  ods.uid,
  dim.uid_int32
FROM ods_events AS ods
JOIN uid_mapping /*+ OPTIONS('lookup.insert-if-not-exists' = 'true') */FOR SYSTEM_TIME AS OF ods.proctime AS dim
  ON dim.uid = ods.uid;

Tabel ods_events berisi data berikut:

Country	Prov	City	Ymd	Uid
CN	Beijing	Haidian	2025-01-01	user_a
CN	Shanghai	Pudong	2025-01-02	user_b
US	California	LA	2025-01-03	user_a
JP	Tokyo	Shibuya	2025-01-04	user_c

Hasil penggabungan adalah sebagai berikut:

Country	Prov	City	Ymd	Uid	Uid_int32
CN	Beijing	Haidian	2025-01-01	user_a	1
CN	Shanghai	Pudong	2025-01-02	user_b	2
US	California	LA	2025-01-03	user_a	1
JP	Tokyo	Shibuya	2025-01-04	user_c	3

Saat user_a muncul pertama kali, ia diberi uid_int32 = 1. Nilai ini digunakan kembali untuk kemunculan berikutnya.
user_b dan user_c masing-masing diberi ID auto-increment baru.

Setelah pekerjaan berjalan, tabel kamus uid_mapping berisi data berikut:

Uid	Uid_int32
user_a	1
user_b	2
user_c	3

Referensi

Untuk opsi konfigurasi yang didukung oleh penggabungan tabel dimensi, lihat Connector parameters.
Untuk informasi lebih lanjut tentang cara menetapkan bucket key, lihat Data bucketing.
Untuk informasi lebih lanjut tentang auto partitioning dan strategi manajemen partisi, lihat Data partitioning.