Realtime Compute for Apache Flink:Instruções JOIN para tabelas de dimensão

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;

Opções suportadas

Comportamento da opção shuffle

A opção shuffle afeta a estratégia de shuffle da junção com a tabela de dimensão. O comportamento efetivo em cada cenário é descrito abaixo.

Exemplos 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

Efeito: O uso de uma estratégia shuffle-hash em um JOIN de tabela de dimensão realiza o shuffle do fluxo principal pela chave de junção antes da junção. Combinada com a estratégia de cache LRU, aumenta a taxa de acerto do cache e reduz solicitações de E/S; combinada com a estratégia de cache ALL, reduz o uso de memória. Um único hint SHUFFLE_HASH pode especifique várias tabelas de dimensão.

Limites: O SHUFFLE_HASH reduz o uso de memória, mas os dados upstream precisam passar por shuffle pela chave de junção, o que introduz sobrecarga de rede adicional. Evite-o nos dois casos seguintes:

• O fluxo principal apresenta forte distorção na chave de junção. O uso de SHUFFLE_HASH tornaria o nó de junção um gargalo, causando backpressure severo em jobs de streaming ou long tails em jobs em lote. Use REPLICATED_SHUFFLE_HASH.

• A tabela de dimensão é pequena e a estratégia de cache ALL não apresenta gargalo de memória. Nesse caso, a economia de memória proporcionada pelo SHUFFLE_HASH não compensa a sobrecarga de rede adicional.

Exemplo 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

Efeito: O REPLICATED_SHUFFLE_HASH comporta-se essencialmente como o SHUFFLE_HASH, exceto por distribuir aleatoriamente as linhas do fluxo principal com a mesma chave entre N instâncias paralelas. Isso resolve o gargalo de desempenho causado pela distorção de dados. Um único hint REPLICATED_SHUFFLE_HASH pode especifique várias tabelas de dimensão.

Limites:

• configure a contagem de buckets para dados distorcidos com table.exec.skew-join.replicate-num (padrão: 16). Seu valor não pode exceder o paralelismo do nó de junção de dimensão. Para detalhes, consulte Ajuste de SQL no nível do job.

• Fluxos de atualização não são suportados. O uso deste hint quando o fluxo principal é um fluxo de atualização gera um erro.

Exemplo de código:

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

SKEW

Efeito: Quando se sabe que a tabela especificada possui distorção, o otimizador aplica a estratégia replicated-shuffle-hash. SKEW é um açúcar sintático — internamente, é implementado com replicated shuffle hash.

Limites:

• Cada hint SKEW pode especifique apenas uma tabela.

• O nome da tabela deve referir-se à tabela principal que possui a distorção, não à tabela de dimensão.

• Fluxos de atualização não são suportados. O uso deste hint quando o fluxo principal é um fluxo de atualização gera um erro.

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