Realtime Compute for Apache Flink:Otimizar Flink SQL

execution.checkpointing.interval: 180s
table.exec.state.ttl: 129600000
table.exec.mini-batch.enabled: true
table.exec.mini-batch.allow-latency: 5s
table.optimizer.distinct-agg.split.enabled: true

Dimensionar recursos do JobManager

Para implantações paralelas, aumente os recursos do JobManager na seção Resources da aba Configuration. Por exemplo:

• Job Manager CPU: 4

• Job Manager Memory: 8 GiB

Dimensionar recursos do TaskManager

Em topologias complexas, aumente os recursos do TaskManager na seção Resources da aba Configuration. Por exemplo:

• Task Manager CPU: 2

• Task Manager Memory: 4 GiB

Ativar miniBatch

O miniBatch armazena em buffer os registros recebidos e os processa em conjunto, reduzindo os acessos ao estado por lote. Isso melhora o throughput ao custo de uma latência ligeiramente maior. O recurso miniBatch aciona o processamento de micro-lotes com base em mensagens de evento inseridas na source em um intervalo especificado.

Quando usar: Cenários de agregação de dados que não exigem latência ultrabaixa.

Como ativar: Adicione o seguinte ao campo Other Configuration na seção Parameters da aba Configuration. Configure custom deployment parameters.

table.exec.mini-batch.enabled: true
table.exec.mini-batch.allow-latency: 5s

Ativar LocalGlobal

O LocalGlobal divide a agregação em duas fases — pré-agregação local em cada nó upstream e, em seguida, agregação global — semelhante ao combine e reduce do MapReduce. A agregação local pré-filtra dados de hotspot, reduzindo o volume que chega à fase global.

Quando usar: Funções de agregação comuns como SUM, COUNT, MAX, MIN e AVG com problemas de hotspot de dados.

Pré-requisitos:

• O miniBatch deve estar ativado. O LocalGlobal usa o intervalo do mini-batch para determinar quantos registros acumular antes da agregação local.

• A função de agregação deve implementar o método merge (AggregateFunction).

Status: Ativado por padrão quando o miniBatch está ativo.

Verificar: Verifique se a topologia final contém um nó GlobalGroupAggregate ou LocalGroupAggregate.

Ativar PartialFinal para COUNT DISTINCT

O LocalGlobal é menos eficaz para COUNT DISTINCT porque a agregação local não consegue remover chaves distintas duplicadas. O PartialFinal resolve isso dividindo a agregação em duas camadas: a primeira dispersa os dados pelo hash da chave distinta, e a segunda realiza a agregação final.

Quando usar: Consultas COUNT DISTINCT com grandes volumes de dados em que o desempenho da agregação é um gargalo.

Como ativar: Adicione o seguinte ao campo Other Configuration. Configure custom deployment parameters.

table.optimizer.distinct-agg.split.enabled: true

Limitações:

• Não é possível usar o PartialFinal em SQL que contenha funções de agregação definidas pelo usuário (UDAFs).

• Ative apenas para grandes volumes de dados. O PartialFinal introduz um shuffle de rede adicional.

Verificar: Confirme se a agregação de uma camada mudou para agregação de duas camadas na topologia final.

Usar AGG WITH FILTER em vez de CASE WHEN

Ao calcular COUNT DISTINCT no mesmo campo sob condições diferentes, use a cláusula FILTER em vez de CASE WHEN. O otimizador reconhece argumentos de filtro diferentes na mesma chave distinta e usa uma única instância de estado compartilhada, reduzindo o acesso ao estado e seu tamanho. Testes de desempenho mostram uma melhoria de 2x em relação ao CASE WHEN.

Antes (menos eficiente):

COUNT(DISTINCT visitor_id) AS uv_total,
COUNT(DISTINCT CASE WHEN is_wireless = 'y' THEN visitor_id ELSE NULL END) AS uv_wireless

Depois (desempenho 2x superior):

COUNT(DISTINCT visitor_id) AS uv_total,
COUNT(DISTINCT visitor_id) FILTER (WHERE is_wireless = 'y') AS uv_wireless

Usar Delta Aggregation (agregação sem estado)

A Delta Aggregation fornece agregação sem estado para agregação de grupo sem janela em streaming. Diferente da agregação de grupo regular, que mantém continuamente um acumulador de agregação para cada grupo no estado do Flink, a Delta Aggregation responde a uma alteração de dados usando a chave de agrupamento para buscar assincronamente os registros de detalhe atuais do grupo e recalcular o resultado agregado a partir desses registros.

Como resultado, a quantidade de estado de agregação no lado do Flink não depende mais do número histórico de grupos. A implantação não precisa expandir continuamente seu estado de agregação à medida que o número de grupos cresce. Isso reduz a quantidade de dados persistidos em checkpoints, diminui o tempo de recuperação e estabiliza o uso de recursos em implantações de agregação de grupo em grande escala.

SET 'table.exec.async-lookup.buffer-capacity' = '100';
SET 'table.exec.async-lookup.timeout' = '3 min';

SET 'table.exec.mini-batch.enabled' = 'true';
SET 'table.exec.mini-batch.allow-latency' = '10s';
SET 'table.exec.mini-batch.size' = '1000';

SET 'table.exec.delta-agg.cache-enabled' = 'true';
SET 'table.exec.delta-agg.cache-size' = '2000';

+U(..., 120) → Matches the condition and is written to the result table
+U(..., 80)  → Does not match the condition and is filtered out

SET 'table.optimizer.delta-agg.ignore-non-unique-key-filter' = 'true';

Separação de chave-valor para joins de dois fluxos

No VVR 6.0.1 e posterior, o engine infere automaticamente se deve ativar a separação de chave-valor para operadores JOIN de dois fluxos, melhorando o desempenho típico de join em mais de 40%.

Configure o parâmetro table.exec.join.kv-separate para controlar esse comportamento:

Ativar miniBatch para joins

Por padrão, o operador de join regular processa os registros individualmente: busca o estado correspondente pela chave de join, atualiza o estado e gera os resultados. Esse padrão por registro aumenta a sobrecarga do backend de estado — especialmente com RocksDB — e pode causar amplificação severa de registros em joins em cascata.

O miniBatch para joins resolve isso com duas otimizações principais:

1. Compactação de registros: Compacta registros no buffer para reduzir o volume de dados antes do processo de join.

2. Supressão de saída: Suprime resultados intermediários redundantes durante o processamento em lote.

O miniBatch aciona o processamento quando a latência máxima permitida é atingida, o limite de tamanho do lote é alcançado ou ocorre um checkpoint.

Versão do VVR: 8.0.4 ou posterior.

Como ativar: Adicione o seguinte ao campo Other Configuration. Configure custom deployment parameters.

table.exec.mini-batch.enabled: true
table.exec.mini-batch.allow-latency: 5s
table.exec.stream.join.mini-batch-enabled: true

Mais indicado para: Outer joins em cascata com amplificação de mensagens. Operadores upstream usam colapso ou mesclagem de mensagens para suprimir a amplificação em operadores downstream. Por exemplo:

SELECT a.id AS a_id, a.a_content, B.id AS b_id, B.b_content
FROM a LEFT JOIN
  (SELECT * FROM b
      LEFT JOIN c ON b.prd_id = c.id) B
ON a.id = B.id

Algoritmos TopN

O algoritmo usado depende se o fluxo de dados de entrada é estático ou dinâmico:

O nome do algoritmo aparece no nome do nó da topologia.

Alternar de RetractRank para UpdateFastRank

O UpdateFastRank exige todas as seguintes condições:

1. O fluxo de entrada não contém mensagens DELETE ou UPDATE_BEFORE. Verifique com EXPLAIN:

      EXPLAIN CHANGELOG_MODE <query_statement_or_insert_statement_or_statement_set>

2. O fluxo de entrada contém informações de chave primária (ex.: colunas usadas em uma cláusula GROUP BY).

3. Os campos ORDER BY são atualizados monotonamente na ordem oposta à classificação. Por exemplo, ORDER BY COUNT DESC, ORDER BY COUNT_DISTINCT DESC ou ORDER BY SUM(valor positivo) DESC.

Exemplo: Para ORDER BY SUM DESC, filtre por valores positivos para garantir a monotonicidade:

INSERT INTO print_test
SELECT
  cate_id,
  seller_id,
  stat_date,
  pay_ord_amt
FROM (
    SELECT
      *,
      ROW_NUMBER() OVER (
        -- The PARTITION BY columns must appear in the subquery's GROUP BY.
        -- Include a time field to prevent disorder when state expires.
        PARTITION BY cate_id, stat_date
        ORDER BY pay_ord_amt DESC
      ) AS rownum
    FROM (
        SELECT
          cate_id,
          seller_id,
          stat_date,
          -- SUM is monotonically increasing because only positive values are included.
          SUM(total_fee) FILTER (WHERE total_fee >= 0) AS pay_ord_amt
        FROM random_test
        WHERE total_fee >= 0
        GROUP BY seller_id, stat_date, cate_id
      ) a
  )
WHERE rownum <= 100;

Reduzir volume de saída

Não inclua rownum na saída SELECT final. Ordene os resultados no frontend para reduzir o volume de gravação no sink. Top-N.

Aumentar o tamanho do cache TopN

O TopN mantém um cache de estado para reduzir leituras em disco. Calcule a taxa de acerto do cache com esta fórmula:

cache_hit = cache_size * parallelism / top_n / partition_key_num

Exemplo: Com Top100, tamanho de cache padrão de 10.000 registros, paralelismo de 50 e 100.000 chaves de partição:

10,000 * 50 / 100 / 100,000 = 5% hit ratio

Uma taxa de acerto de 5% significa que a maioria das leituras vai para o disco, causando instabilidade nas métricas de state seek e degradando o desempenho. Aumente o tamanho do cache:

table.exec.rank.topn-cache-size: 200000

Com 200.000 entradas de cache:

200,000 * 50 / 100 / 100,000 = 100% hit ratio

Se a contagem de chaves de partição for grande, aumente também o tamanho do cache TopN e a memória heap. Configure a deployment.

Incluir um campo de tempo em PARTITION BY

Adicione um campo de tempo como day à cláusula PARTITION BY. Sem ele, os resultados do TopN ficam desordenados quando os dados de estado expiram devido ao TTL.

Sintaxe

A deduplicação é um caso especial de TopN. Use ROW_NUMBER() com uma cláusula OVER para atribuir um número de linha por partição e, em seguida, mantenha apenas rownum = 1:

SELECT *
FROM (
   SELECT *,
    ROW_NUMBER() OVER (PARTITION BY col1[, col2..]
     ORDER BY timeAttributeCol [ASC|DESC]) AS rownum
   FROM table_name)
WHERE rownum = 1

Comportamento do atributo de tempo:

• proctime (tempo de processamento): A deduplicação baseia-se no momento em que os registros são processados. Os resultados podem variar entre execuções.

• rowtime (tempo de evento): A deduplicação baseia-se no momento em que os registros foram produzidos. Os resultados são determinísticos.

Keep FirstRow

Mantém o primeiro registro por chave de deduplicação. O estado armazena apenas dados da chave primária, tornando o acesso ao estado altamente eficiente.

SELECT *
FROM (
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY b ORDER BY proctime) AS rowNum
  FROM T
)
WHERE rowNum = 1

Deduplica a tabela T pela coluna b, mantendo o registro mais antigo por tempo de processamento. Em vez de declarar um atributo proctime, chame a função PROCTIME().

Keep LastRow

Mantém o registro mais recente por chave de deduplicação. Isso tem um desempenho ligeiramente superior à função LAST_VALUE.

SELECT *
FROM (
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY b, d ORDER BY rowtime DESC) AS rowNum
  FROM T
)
WHERE rowNum = 1

Deduplica a tabela T pelas colunas b e d, mantendo o registro mais recente por tempo de evento.

Preferir funções integradas em vez de UDFs

As funções integradas são otimizadas para serialização, desserialização e operações no nível de byte. Substitua UDFs por equivalentes integrados sempre que possível.

Usar delimitadores de caractere único em KEYVALUE

A função KEYVALUE(content, keyValueSplit, keySplit, keyName) executa cerca de 30% mais rápido quando keyValueSplit e keySplit são caracteres únicos, como : ou ,. Com delimitadores de caractere único, o engine busca a chave alvo diretamente nos dados binários sem analisar toda a entrada.

Padrões do operador LIKE

Escapar underscores: O underscore (_) é um curinga de caractere único em SQL. Para corresponder a um underscore literal, use um caractere de escape:

LIKE '%seller/_id%' ESCAPE '/'

Sem o escape, LIKE '%seller_id%' também corresponde a seller#id, sellerxid, seller1id e outras strings não intencionais.

Evitar expressões regulares

Expressões regulares podem ser 100x mais lentas que operações aritméticas e podem entrar em loops infinitos em casos extremos, bloqueando implantações. Use LIKE sempre que possível.

Para casos que exigem expressões regulares, consulte REGEXP e REGEXP_REPLACE.

Sintaxe

A sintaxe segue o Apache Calcite SQL:

SELECT /*+ hint [, hint ] */ ...

-- Where:
-- hint:       hintName(hintOption [, hintOption]*)
-- hintOption:  simpleIdentifier | numericLiteral | stringLiteral

Join hints

Query hints modificam o plano de execução dentro do bloco de consulta atual. Atualmente, o Flink suporta apenas join hints, tanto para dimension table joins quanto para regular joins.