Realtime Compute for Apache Flink:Lookup join

Efek

Strategi join SHUFFLE_HASH melakukan shuffle data utama berdasarkan kunci gabungan sebelum operasi JOIN. Jika kebijakan cache adalah LRU, rasio hit cache meningkat dan jumlah permintaan I/O berkurang. Jika kebijakan cache adalah ALL, penggunaan memori berkurang. Anda dapat menentukan beberapa tabel dimensi dalam setiap petunjuk join SHUFFLE_HASH.

Batasan

Meskipun overhead memori berkurang, strategi join SHUFFLE_HASH menimbulkan overhead jaringan tambahan karena data hulu perlu di-shuffle berdasarkan kunci gabungan. Oleh karena itu, strategi ini tidak cocok untuk skenario berikut:

• Data utama mengalami kesenjangan data parah pada kunci gabungan. Penggunaan strategi join SHUFFLE_HASH dalam kasus ini dapat menyebabkan bottleneck performa pada operator join, yang berpotensi menimbulkan tekanan balik parah dalam penerapan streaming atau long tails dalam penerapan batch. Dalam skenario ini, kami merekomendasikan menggunakan strategi join REPLICATED_SHUFFLE_HASH.

• Jika tabel dimensi berisi data dalam jumlah kecil dan tidak mengalami bottleneck memori selama pemuatan tabel ketika kebijakan cache adalah ALL, penghematan overhead memori yang diperoleh mungkin tidak sebanding dengan overhead jaringan tambahan yang ditimbulkan.

Kode contoh

-- Aktifkan strategi join SHUFFLE_HASH hanya untuk tabel dimensi dim1.
SELECT /*+ SHUFFLE_HASH(dim1) */
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

-- Aktifkan strategi join SHUFFLE_HASH untuk tabel dimensi dim1 dan dim2.
SELECT /*+ SHUFFLE_HASH(dim1, dim2) */
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

-- Gunakan alias D1 untuk tabel dimensi dim1 dalam petunjuk untuk mengaktifkan strategi join SHUFFLE_HASH.
SELECT /*+ SHUFFLE_HASH(D1) */
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

-- Gunakan alias untuk tabel dimensi dim1 dan dim2 dalam petunjuk untuk mengaktifkan strategi join SHUFFLE_HASH.
SELECT /*+ SHUFFLE_HASH(D1, D2) */
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

Efek

Efek REPLICATED_SHUFFLE_HASH pada dasarnya sama dengan SHUFFLE_HASH. Namun, REPLICATED_SHUFFLE_HASH menyebarkan data utama yang memiliki kunci yang sama secara acak ke sejumlah thread konkuren tertentu untuk mengatasi bottleneck performa akibat kesenjangan data. Anda dapat menentukan beberapa tabel dimensi dalam setiap petunjuk join REPLICATED_SHUFFLE_HASH.

Batasan

• Anda harus mengonfigurasi parameter table.exec.skew-join.replicate-num untuk menentukan jumlah bucket yang berisi data miring. Nilai default parameter ini adalah 16 dan tidak boleh melebihi jumlah thread konkuren pada operator join tabel dimensi. Untuk informasi selengkapnya tentang cara mengonfigurasi parameter ini, lihat Bagaimana cara mengonfigurasi parameter running kustom untuk pekerjaan?

• Aliran pembaruan tidak didukung. Jika aliran utama adalah aliran pembaruan dan Anda menggunakan strategi join REPLICATED_SHUFFLE_HASH, sistem akan mengembalikan error.

Kode contoh

-- Aktifkan strategi join REPLICATED_SHUFFLE_HASH untuk tabel dimensi dim1.
SELECT /*+ REPLICATED_SHUFFLE_HASH(dim1) */ 
FROM src AS T 
LEFT JOIN dim1 FOR SYSTEM_TIME AS OF PROCTIME() AS D1 ON T.a = D1.a

-- Gunakan alias untuk tabel dimensi dim1 dalam petunjuk untuk mengaktifkan strategi join REPLICATED_SHUFFLE_HASH.
SELECT /*+ REPLICATED_SHUFFLE_HASH(D1) */ 
FROM src AS T 
LEFT JOIN dim1 FOR SYSTEM_TIME AS OF PROCTIME() AS D1 ON T.a = D1.a

Efek

Jika tabel yang ditentukan mengalami kesenjangan data, pengoptimal menggunakan strategi join REPLICATED_SHUFFLE_HASH untuk operasi JOIN pada tabel dimensi. SKEW hanyalah syntactic sugar; strategi join REPLICATED_SHUFFLE_HASH yang sebenarnya digunakan di lapisan bawah.

Batasan

• Anda hanya dapat menentukan satu tabel dalam setiap petunjuk SKEW.

• Nama tabel harus merupakan nama tabel primary yang mengalami kesenjangan data, bukan tabel dimensi.

• Aliran pembaruan tidak didukung. Jika aliran utama adalah aliran pembaruan dan Anda menggunakan strategi join SKEW, sistem akan mengembalikan error.

Kode contoh

SELECT /*+ SKEW(src) */  
FROM src AS T 
LEFT JOIN dim1 FOR SYSTEM_TIME AS OF PROCTIME() AS D1 ON T.a = D1.a