Realtime Compute for Apache Flink:JOIN statements for dimension tables

Effect

The SHUFFLE_HASH join strategy allows mainstream data to be shuffled based on join keys before the JOIN operation is performed. If the cache policy is LRU, the cache hit ratio is increased and the number of I/O requests is reduced. If the cache policy is ALL, the memory usage is reduced. You can specify multiple dimension tables in each SHUFFLE_HASH join hint.

Limits

If you use the SHUFFLE_HASH join strategy, the memory overhead is reduced. However, additional network overheads are introduced because upstream data needs to be shuffled based on join keys. Therefore, the SHUFFLE_HASH join strategy is not suitable for the following scenarios:

• The mainstream data has severe data skew on join keys. If you use the SHUFFLE_HASH join strategy to join data, the join operator may cause a performance bottleneck due to data skew. This may cause severe backpressure in streaming deployments or severe long tails in batch deployments. In this scenario, we recommend that you use the REPLICATED_SHUFFLE_HASH join strategy.

• If a dimension table contains a small amount of data and does not have a memory bottleneck during table loading when the cache policy is ALL, the memory overheads saved by using the SHUFFLE_HASH join strategy may not be cost-effective, compared with the additional network overheads introduced by using the SHUFFLE_HASH join strategy.

Sample code

-- Enable the SHUFFLE_HASH join strategy only for the dimension table 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

-- Enable the SHUFFLE_HASH join strategy for dimension tables dim1 and 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

-- You must use the alias D1 for the dimension table dim1 in the hint to enable the SHUFFLE_HASH join strategy. 
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

-- Use aliases for dimension tables dim1 and dim2 in the hint to enable the SHUFFLE_HASH join strategy. 
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

Effect

The effect of REPLICATED_SHUFFLE_HASH is basically the same as the effect of SHUFFLE_HASH. However, REPLICATED_SHUFFLE_HASH randomly scatters the mainstream data that has the same key to the specified number of concurrent threads to resolve the performance bottleneck caused by data skew. You can specify multiple dimension tables in each REPLICATED_SHUFFLE_HASH join hint.

Limits

• You must configure the table.exec.skew-join.replicate-num parameter to specify the number of buckets that contain skewed data. The default value of this parameter is 16. The value of this parameter cannot be greater than the number of concurrent threads on the join operator of the dimension table. For more information about how to configure this parameter, see How do I configure parameters for deployment running?

• Update streams are not supported. If the mainstream is an update stream and you use the REPLICATED_SHUFFLE_HASH join strategy, an error is returned.

Sample code

-- Enable the REPLICATED_SHUFFLE_HASH join strategy for the dimension table 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

-- You must use the alias for the dimension table dim1 in the hint to enable the REPLICATED_SHUFFLE_HASH join.
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

Effect

If the specified table has data skew, the optimizer uses the REPLICATED_SHUFFLE_HASH join strategy for the JOIN operation on the dimension table. SKEW is only a syntactic sugar and the REPLICATED_SHUFFLE_HASH join strategy is actually used at the underlying layer.

Limits

• You can specify only one table in each SKEW hint.

• The name of the table must be the name of the primary table that has data skew instead of a dimension table.

• Update streams are not supported. If the mainstream is an update stream and you use the SKEW join strategy, an error is returned.

Sample code

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