Realtime Compute for Apache Flink:Pernyataan JOIN untuk tabel dimensi

Kategori parameter penyetelan dan cara meneruskannya

Penyetelan JOIN tabel dimensi melibatkan tiga kategori parameter. Setiap kategori memiliki cara tersendiri untuk diteruskan, dan menggunakan mekanisme yang salah menyebabkan petunjuk diabaikan tanpa peringatan.

Timpa opsi konektor dengan petunjuk OPTIONS

Petunjuk OPTIONS() memungkinkan Anda menimpa opsi WITH suatu tabel langsung dalam kueri, tanpa mengubah pernyataan CREATE TABLE. Petunjuk ini umumnya digunakan untuk menyetel perilaku cache.

SELECT t.id, t.name, w.phoneNumber
FROM kafka_input AS t
LEFT JOIN phoneNumber /*+ OPTIONS(
  'lookup.cache' = 'PARTIAL',
  'lookup.partial-cache.max-rows' = '1000'
) */ FOR SYSTEM_TIME AS OF PROCTIME() AS w
ON t.name = w.name;

Setiap kunci di dalam OPTIONS harus merupakan opsi WITH yang benar-benar didukung oleh konektor.

Konfigurasi perilaku lookup dengan petunjuk LOOKUP

Petunjuk LOOKUP berperilaku konsisten dengan versi komunitas Apache Flink. Petunjuk ini mengonfigurasi mode sinkron/asinkron, pengulangan, dan perilaku shuffle untuk operasi join tunggal. Untuk detailnya, lihat Apache Flink Lookup Hint.

• Petunjuk LOOKUP hanya tersedia di VVR 8.0 dan versi lebih baru.

• Opsi 'shuffle' = 'true' hanya tersedia di VVR 8.0.8 dan versi lebih baru.

• Alias didukung di VVR 8.0 dan versi lebih baru. Jika tabel dimensi dirujuk dengan alias, petunjuk harus menggunakan alias tersebut.

-- Terapkan strategi shuffle hanya ke tabel dimensi dim1.
SELECT /*+ LOOKUP('table'='dim1', 'shuffle' = 'true') */ ...
FROM src AS T
LEFT JOIN dim1 FOR SYSTEM_TIME AS OF PROCTIME() ON T.a = dim1.a
LEFT JOIN dim2 FOR SYSTEM_TIME AS OF PROCTIME() ON T.b = dim2.b;

-- Terapkan strategi shuffle ke kedua dim1 dan dim2.
SELECT /*+ LOOKUP('table'='dim1', 'shuffle' = 'true'), LOOKUP('table'='dim2', 'shuffle' = 'true') */ ...
FROM src AS T
LEFT JOIN dim1 FOR SYSTEM_TIME AS OF PROCTIME() ON T.a = dim1.a
LEFT JOIN dim2 FOR SYSTEM_TIME AS OF PROCTIME() ON T.b = dim2.b;

-- Saat dim1 dirujuk dengan alias D1, petunjuk harus menggunakan alias tersebut.
SELECT /*+ LOOKUP('table'='D1', 'shuffle' = 'true') */ ...
FROM src AS T
LEFT JOIN dim1 FOR SYSTEM_TIME AS OF PROCTIME() AS D1 ON T.a = D1.a
LEFT JOIN dim2 FOR SYSTEM_TIME AS OF PROCTIME() AS D2 ON T.b = D2.b;

-- Terapkan strategi shuffle ke kedua dim1 dan dim2 menggunakan alias mereka.
SELECT /*+ LOOKUP('table'='D1', 'shuffle' = 'true'), LOOKUP('table'='D2', 'shuffle' = 'true') */ ...
FROM src AS T
LEFT JOIN dim1 FOR SYSTEM_TIME AS OF PROCTIME() AS D1 ON T.a = D1.a
LEFT JOIN dim2 FOR SYSTEM_TIME AS OF PROCTIME() AS D2 ON T.b = D2.b;

Atur TableConfig tingkat pekerjaan dengan SET

Kunci TableConfig seperti table.exec.* adalah pengaturan global tingkat pekerjaan. Kunci-kunci ini harus diteruskan melalui SET; baik petunjuk LOOKUP maupun OPTIONS tidak menerimanya.

Kunci TableConfig umum untuk JOIN tabel dimensi:

-- Atur default tingkat pekerjaan di awal pekerjaan.
SET 'table.exec.async-lookup.buffer-capacity' = '3072';
SET 'table.exec.async-lookup.timeout' = '180s';

INSERT INTO sink_table
SELECT ...
FROM src_table
LEFT JOIN dim_table FOR SYSTEM_TIME AS OF PROCTIME() AS d
ON src_table.key = d.key;

Kontrol shuffle dengan petunjuk strategi join

Petunjuk strategi join mengontrol perilaku shuffle untuk join tabel dimensi. Petunjuk yang tersedia adalah SHUFFLE_HASH, REPLICATED_SHUFFLE_HASH, dan SKEW. Tabel berikut menunjukkan strategi join mana yang cocok untuk setiap strategi cache.

-- Aktifkan SHUFFLE_HASH hanya untuk dim1.
SELECT /*+ SHUFFLE_HASH(dim1) */ ...

-- Aktifkan SHUFFLE_HASH untuk kedua dim1 dan dim2.
SELECT /*+ SHUFFLE_HASH(dim1, dim2) */ ...

-- Saat dim1 dirujuk dengan alias D1, petunjuk harus menggunakan alias tersebut.
SELECT /*+ SHUFFLE_HASH(D1) */ ...

SELECT /*+ REPLICATED_SHUFFLE_HASH(dim1) */ ...

SELECT /*+ SKEW(src) */  ...

Contoh 1: JOIN tabel dimensi dasar

Bentuk paling dasar: memperkaya aliran Kafka dengan tabel dimensi MySQL. Tidak ada petunjuk penyetelan yang digunakan.

CREATE TEMPORARY TABLE kafka_input (
  id   BIGINT,
  name VARCHAR,
  age  BIGINT
) WITH (
  'connector' = 'kafka',
  'topic' = '<yourTopic>',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'properties.group.id' = '<yourKafkaConsumerGroupId>',
  'format' = 'csv'
);

CREATE TEMPORARY TABLE phoneNumber (
  name        VARCHAR,
  phoneNumber BIGINT,
  PRIMARY KEY (name) NOT ENFORCED
) WITH (
  'connector' = 'mysql',
  'hostname' = '<yourHostname>',
  'port' = '3306',
  'username' = '<yourUsername>',
  'password' = '<yourPassword>',
  'database-name' = '<yourDatabaseName>',
  'table-name' = '<yourTableName>'
);

CREATE TEMPORARY TABLE result_infor (
  id          BIGINT,
  phoneNumber BIGINT,
  name        VARCHAR
) WITH (
  'connector' = 'blackhole'
);

INSERT INTO result_infor
SELECT
  t.id,
  w.phoneNumber,
  t.name
FROM kafka_input AS t
JOIN phoneNumber FOR SYSTEM_TIME AS OF PROCTIME() AS w
ON t.name = w.name;

Contoh 2: Aktifkan cache dengan petunjuk OPTIONS dan konfigurasi lookup asinkron dengan petunjuk LOOKUP

Skenario penyetelan: QPS tabel dimensi tidak mencukupi. Anda ingin mengaktifkan cache parsial pada tabel dimensi dan mengaktifkan lookup asinkron dengan antrian buffer yang lebih besar.

INSERT INTO user_behavior_wide
SELECT /*+ LOOKUP('table' = 't2', 'async' = 'true', 'capacity' = '3072') */
  t1.member_id AS member_id,
  t2.tag AS tag
FROM user_behavior_datagen AS t1
LEFT JOIN fluss.fluss.user_active_info /*+ OPTIONS(
    'lookup.cache' = 'PARTIAL',
    'lookup.partial-cache.max-rows' = '1000'
  ) */
FOR SYSTEM_TIME AS OF PROCTIME() AS t2
ON t1.member_id = t2.member_id;

• Opsi konektor (lookup.cache, lookup.partial-cache.max-rows) diteruskan melalui petunjuk OPTIONS() yang ditempatkan tepat setelah referensi tabel dimensi; ini menimpa opsi WITH yang sesuai saat kueri.

• Perilaku lookup (async, capacity) dikonfigurasi melalui petunjuk LOOKUP() yang ditempatkan setelah SELECT. OPTIONS dan LOOKUP adalah dua jenis petunjuk yang berbeda dan harus ditulis secara terpisah; menggabungkannya akan diam-diam menonaktifkan petunjuk tersebut.

Contoh 3: Setel buffer lookup asinkron tingkat pekerjaan dengan SET

Skenario penyetelan: beberapa JOIN tabel dimensi dalam pekerjaan yang sama memerlukan buffer lookup asinkron yang lebih besar. Anda ingin perubahan ini berlaku secara global, dan mengaktifkan mode asinkron melalui petunjuk LOOKUP.

SET 'table.exec.async-lookup.buffer-capacity' = '3072';
SET 'table.exec.async-lookup.timeout' = '180s';

INSERT INTO user_behavior_wide
SELECT /*+ LOOKUP('table' = 't2', 'async' = 'true') */
  t1.member_id,
  t2.tag
FROM user_behavior_datagen AS t1
LEFT JOIN fluss.fluss.user_active_info
FOR SYSTEM_TIME AS OF PROCTIME() AS t2
ON t1.member_id = t2.member_id;

• Parameter seperti table.exec.async-lookup.buffer-capacity adalah kunci TableConfig tingkat pekerjaan dan hanya dapat diatur dengan SET. Menempatkannya di dalam OPTIONS atau LOOKUP tidak berpengaruh.

• Jika petunjuk LOOKUP tidak menentukan capacity, nilai global dari SET yang digunakan. Jika keduanya ditentukan, petunjuk LOOKUP memiliki prioritas lebih tinggi pada join individual tersebut.