Realtime Compute for Apache Flink:Instructions JOIN pour les tables de dimension

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;

Options prises en charge

Comportement de l'option shuffle

L'option shuffle affecte la stratégie de brassage de la jointure de la table de dimension. Le comportement effectif dans chaque scénario est le suivant.

Exemples de code

-- Apply the shuffle strategy only to dimension table 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;

-- Apply the shuffle strategy to both dim1 and 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;

-- When dim1 is referenced by the alias D1, the hint must use the alias.
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;

-- Apply the shuffle strategy to both dim1 and dim2 using their aliases.
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;

-- Set job-level defaults at the beginning of the job.
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;

SHUFFLE_HASH

Effet : L'utilisation d'une stratégie de hachage avec brassage dans une jointure de table de dimension brasse le flux principal par clé de jointure avant la jointure. Combinée à la stratégie de cache LRU, elle augmente le taux de succès du cache et réduit les requêtes d'E/S ; combinée à la stratégie de cache ALL, elle réduit l'utilisation de la mémoire. Un seul indice SHUFFLE_HASH peut spécifier plusieurs tables de dimension.

Limites : SHUFFLE_HASH réduit l'utilisation de la mémoire, mais les données en amont doivent être brassées par clé de jointure, ce qui introduit une surcharge réseau supplémentaire. Évitez-le dans les deux cas suivants :

• Le flux principal est fortement déséquilibré sur la clé de jointure. L'utilisation de SHUFFLE_HASH ferait du nœud de jointure un goulot d'étranglement, provoquant une contre-pression sévère pour les jobs en continu ou des queues longues pour les jobs par lots. Utilisez plutôt REPLICATED_SHUFFLE_HASH.

• La table de dimension est petite et la stratégie de cache ALL ne présente aucun goulot d'étranglement mémoire. Dans ce cas, la mémoire économisée par SHUFFLE_HASH ne justifie pas la surcharge réseau supplémentaire.

Exemple de code :

-- Enable SHUFFLE_HASH for dim1 only.
SELECT /*+ SHUFFLE_HASH(dim1) */ ...

-- Enable SHUFFLE_HASH for both dim1 and dim2.
SELECT /*+ SHUFFLE_HASH(dim1, dim2) */ ...

-- When dim1 is referenced by the alias D1, the hint must use the alias.
SELECT /*+ SHUFFLE_HASH(D1) */ ...

REPLICATED_SHUFFLE_HASH

Effet : REPLICATED_SHUFFLE_HASH se comporte essentiellement de la même manière que SHUFFLE_HASH, sauf qu'il distribue aléatoirement les lignes du flux principal ayant la même clé sur N instances parallèles. Cela permet de résoudre le goulot d'étranglement de performance causé par le déséquilibre des données. Un seul indice REPLICATED_SHUFFLE_HASH peut spécifier plusieurs tables de dimension.

Limites :

• Configurez le nombre de buckets pour les données déséquilibrées avec table.exec.skew-join.replicate-num (par défaut : 16). Sa valeur ne peut pas dépasser le parallélisme du nœud de jointure de dimension. Pour plus de détails, consultez la section Optimisation SQL au niveau du job.

• Les flux de mise à jour ne sont pas pris en charge. L'utilisation de cet indice lorsque le flux principal est un flux de mise à jour génère une erreur.

Exemple de code :

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

SKEW

Effet : Lorsque la table spécifiée est connue pour être déséquilibrée, l'optimiseur applique la stratégie de hachage avec brassage répliqué. SKEW est un sucre syntaxique ; il est implémenté en interne avec un hachage de brassage répliqué.

Limites :

• Chaque indice SKEW ne peut spécifier qu'une seule table.

• Le nom de la table doit faire référence à la table principale présentant le déséquilibre, et non à la table de dimension.

• Les flux de mise à jour ne sont pas pris en charge. L'utilisation de cet indice lorsque le flux principal est un flux de mise à jour génère une erreur.

Exemple de code :

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

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;

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;

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;