Realtime Compute for Apache Flink:Sentencias JOIN para tablas de dimensiones

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;

Opciones compatibles

Comportamiento de la opción shuffle

La opción shuffle afecta la estrategia de redistribución de la unión de tabla de dimensiones. El comportamiento efectivo en cada escenario es el siguiente.

Ejemplos de código

-- 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

Efecto: Usar una estrategia shuffle-hash en un JOIN de tabla de dimensiones redistribuye el flujo principal por clave de unión antes de la unión. Combinada con la estrategia de caché LRU, aumenta la tasa de aciertos de caché y reduce las solicitudes de E/S; combinada con la estrategia de caché ALL, reduce el uso de memoria. Un solo hint SHUFFLE_HASH puede especificar varias tablas de dimensiones.

Límites: SHUFFLE_HASH reduce el uso de memoria, pero los datos ascendentes deben redistribuirse por clave de unión, lo que introduce sobrecarga de red adicional. Evítelo en los siguientes dos casos:

• El flujo principal tiene un sesgo significativo en la clave de unión. Usar SHUFFLE_HASH convertiría el nodo de unión en un cuello de botella, causando contrapresión severa en trabajos de streaming o colas largas en trabajos por lotes. Use REPLICATED_SHUFFLE_HASH en su lugar.

• La tabla de dimensiones es pequeña y la estrategia de caché ALL no presenta cuellos de botella de memoria. En este caso, la memoria ahorrada por SHUFFLE_HASH no justifica la sobrecarga de red adicional.

Ejemplo de código:

-- 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

Efecto: REPLICATED_SHUFFLE_HASH se comporta esencialmente igual que SHUFFLE_HASH, excepto que distribuye aleatoriamente las filas del flujo principal con la misma clave entre N instancias paralelas. Esto aborda el cuello de botella de rendimiento causado por el sesgo de datos. Un solo hint REPLICATED_SHUFFLE_HASH puede especificar varias tablas de dimensiones.

Límites:

• Configure la cantidad de buckets para datos sesgados con table.exec.skew-join.replicate-num (predeterminado: 16). Su valor no puede exceder el paralelismo del nodo de unión de dimensiones. Para más detalles, consulte Ajuste SQL a nivel de trabajo.

• No se admiten flujos de actualización. Usar este hint cuando el flujo principal es un flujo de actualización genera un error.

Ejemplo de código:

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

SKEW

Efecto: Cuando se sabe que la tabla especificada tiene sesgo, el optimizador aplica la estrategia replicated-shuffle-hash. SKEW es azúcar sintáctica; internamente se implementa con redistributed shuffle hash.

Límites:

• Cada hint SKEW solo puede especificar una tabla.

• El nombre de la tabla debe referirse a la tabla principal que tiene el sesgo, no a la tabla de dimensiones.

• No se admiten flujos de actualización. Usar este hint cuando el flujo principal es un flujo de actualización genera un error.

Ejemplo de código:

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;