Realtime Compute for Apache Flink:Optimize 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

Escalado de recursos de JobManager

Para despliegues paralelos, aumente los recursos de JobManager en la sección Resources de la pestaña Configuration. Por ejemplo:

• Job Manager CPU: 4

• Job Manager Memory: 8 GiB

Escalado de recursos de TaskManager

Para topologías complejas, aumente los recursos de TaskManager en la sección Resources de la pestaña Configuration. Por ejemplo:

• Task Manager CPU: 2

• Task Manager Memory: 4 GiB

Habilitación de miniBatch

MiniBatch almacena en búfer los registros entrantes y los procesa en conjunto, lo que reduce los accesos al estado por cada lote. Esto mejora el throughput a costa de una latencia ligeramente mayor. La función miniBatch activa el procesamiento por micro-lotes según los mensajes de evento insertados en el origen a un intervalo especificado.

Cuándo usarlo: Escenarios de agregación de datos donde no se requiere una latencia ultrabaja.

Cómo habilitarlo: Agregue lo siguiente al campo Other Configuration en la sección Parameters de la pestaña Configuration. Configure custom deployment parameters.

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

Habilitación de LocalGlobal

LocalGlobal divide la agregación en dos fases: una preagregación local en cada nodo ascendente y luego una agregación global, similar a las fases combine y reduce de MapReduce. La agregación local prefiltrea los datos de puntos críticos, lo que reduce el volumen que llega a la fase global.

Cuándo usarlo: Funciones de agregación comunes como SUM, COUNT, MAX, MIN y AVG cuando existen problemas de puntos críticos de datos.

Requisitos previos:

• MiniBatch debe estar habilitado. LocalGlobal utiliza el intervalo de mini-batch para determinar cuántos registros acumular antes de la agregación local.

• La función de agregación debe implementar el método merge (AggregateFunction).

Estado: Habilitado de forma predeterminada cuando miniBatch está activo.

Verificación: Compruebe si la topología final contiene un nodo GlobalGroupAggregate o LocalGroupAggregate.

Habilitación de PartialFinal para COUNT DISTINCT

LocalGlobal es menos eficaz para COUNT DISTINCT porque la agregación local no puede eliminar claves distinct duplicadas. PartialFinal soluciona esto al dividir la agregación en dos capas: la primera dispersa los datos según el hash de la clave distinct, y la segunda realiza la agregación final.

Cuándo usarlo: Consultas COUNT DISTINCT con grandes volúmenes de datos donde el rendimiento de la agregación representa un cuello de botella.

Cómo habilitarlo: Agregue lo siguiente al campo Other Configuration. Configure custom deployment parameters.

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

Limitaciones:

• PartialFinal no se puede utilizar en sentencias SQL que contengan funciones de agregación definidas por el usuario (UDAF).

• Habilítelo solo para grandes volúmenes de datos. PartialFinal introduce una operación shuffle de red adicional.

Verificación: Compruebe si la agregación de una capa cambia a una agregación de dos capas en la topología final.

Uso de AGG WITH FILTER en lugar de CASE WHEN

Al calcular COUNT DISTINCT sobre el mismo campo bajo diferentes condiciones, utilice la cláusula FILTER en lugar de CASE WHEN. El optimizador reconoce distintos argumentos de filtro sobre la misma clave distinct y utiliza una única instancia de estado compartida, lo que reduce el acceso al estado y su tamaño. Las pruebas de rendimiento muestran una mejora de 2x respecto a 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

Después (rendimiento 2x superior):

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

Uso de Delta Aggregation (agregación sin estado)

Delta Aggregation proporciona una agregación sin estado para la agregación por grupos sin ventanas en streaming. A diferencia de la agregación por grupos regular, que mantiene continuamente un acumulador de agregación para cada grupo en el estado de Flink, Delta Aggregation responde a un cambio de datos utilizando la clave de agrupación para buscar de forma asíncrona los registros de detalle actuales del grupo y recalcular el resultado agregado a partir de dichos registros.

Como resultado, la cantidad de estado de agregación en el lado de Flink ya no depende del número histórico de grupos. El despliegue no necesita expandir continuamente su estado de agregación a medida que aumenta el número de grupos. Esto reduce la cantidad de datos persistidos en los checkpoints, acorta el tiempo de recuperación y estabiliza el uso de recursos en despliegues de agregación por grupos a gran 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';

Separación de clave-valor para joins de dos streams

En VVR 6.0.1 y versiones posteriores, el motor infiere automáticamente si debe habilitar la separación de clave-valor para operadores JOIN de dos streams, lo que mejora el rendimiento típico de los joins en más de un 40%.

Configure el parámetro table.exec.join.kv-separate para controlar este comportamiento:

Habilitación de miniBatch para joins

De forma predeterminada, el operador join regular procesa los registros uno por uno: busca el estado homólogo mediante la clave de join, actualiza el estado y luego emite los resultados. Este patrón por registro incrementa la sobrecarga del backend de estado, especialmente con RocksDB, y puede causar una amplificación severa de registros en joins en cascada.

MiniBatch para joins aborda esto con dos optimizaciones principales:

1. Plegado de registros: Pliega los registros en el búfer para reducir el volumen de datos antes del proceso de join.

2. Supresión de salida: Suprime resultados intermedios redundantes durante el procesamiento por lotes.

MiniBatch activa el procesamiento cuando se alcanza la latencia máxima permitida, se alcanza el límite de tamaño del lote o se produce un checkpoint.

Versión de VVR: 8.0.4 o posterior.

Cómo habilitarlo: Agregue lo siguiente al 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

Ideal para: Joins externos en cascada con amplificación de mensajes. Los operadores ascendentes utilizan colapso o fusión de mensajes para suprimir la amplificación en los operadores descendentes. Por ejemplo:

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

El algoritmo utilizado depende de si el stream de datos de entrada es estático o dinámico:

El nombre del algoritmo aparece en el nombre del nodo de la topología.

Cambio de RetractRank a UpdateFastRank

UpdateFastRank requiere que se cumplan todas las siguientes condiciones:

1. El stream de entrada no contiene mensajes DELETE o UPDATE_BEFORE. Verifíquelo con EXPLAIN:

      EXPLAIN CHANGELOG_MODE <query_statement_or_insert_statement_or_statement_set>

2. El stream de entrada contiene información de clave principal (por ejemplo, columnas utilizadas en una cláusula GROUP BY).

3. Los campos ORDER BY se actualizan monótonamente en el orden opuesto al de clasificación. Por ejemplo, ORDER BY COUNT DESC, ORDER BY COUNT_DISTINCT DESC u ORDER BY SUM(valor positivo) DESC.

Ejemplo: Para ORDER BY SUM DESC, filtre por valores positivos para garantizar la monotonicidad:

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;

Reducción del volumen de salida

No incluya rownum en la salida SELECT final. Ordene los resultados en el frontend para reducir el volumen de escritura en el sink. Top-N.

Aumento del tamaño de la caché TopN

TopN mantiene una caché de estado para reducir las lecturas de disco. Calcule la tasa de aciertos de la caché con esta fórmula:

cache_hit = cache_size * parallelism / top_n / partition_key_num

Ejemplo: Con Top100, un tamaño de caché predeterminado de 10,000 registros, paralelismo de 50 y 100,000 claves de partición:

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

Una tasa de aciertos del 5% significa que la mayoría de las lecturas van al disco, lo que causa inestabilidad en las métricas de state seek y degrada el rendimiento. Aumente el tamaño de la caché:

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

Con 200,000 entradas de caché:

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

Si el recuento de claves de partición es grande, aumente también el tamaño de la caché TopN y la memoria heap. Configure a deployment.

Inclusión de un campo de tiempo en PARTITION BY

Agregue un campo de tiempo como day a la cláusula PARTITION BY. Sin él, los resultados de TopN se desordenan cuando los datos de estado caducan debido al TTL.

Sintaxis

La deduplicación es un caso especial de TopN. Utilice ROW_NUMBER() con una cláusula OVER para asignar un número de fila por partición y luego conserve solo 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

Comportamiento del atributo de tiempo:

• proctime (tiempo de procesamiento): La deduplicación se basa en el momento en que se procesan los registros. Los resultados pueden variar entre ejecuciones.

• rowtime (tiempo de evento): La deduplicación se basa en el momento en que se produjeron los registros. Los resultados son deterministas.

Keep FirstRow

Conserva el primer registro por clave de deduplicación. El estado almacena solo datos de clave principal, por lo que el acceso al estado es altamente eficiente.

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

Deduplica la tabla T por la columna b, conservando el registro más antiguo según el tiempo de procesamiento. En lugar de declarar un atributo proctime, llame a la función PROCTIME().

Keep LastRow

Conserva el registro más reciente por clave de deduplicación. Esto tiene un rendimiento ligeramente superior al de la función LAST_VALUE.

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

Deduplica la tabla T por las columnas b y d, conservando el registro más reciente según el tiempo de evento.

Preferencia por funciones integradas sobre UDF

Las funciones integradas están optimizadas para serialización, deserialización y operaciones a nivel de bytes. Reemplace las UDF con equivalentes integrados siempre que sea posible.

Uso de delimitadores de un solo carácter en KEYVALUE

La función KEYVALUE(content, keyValueSplit, keySplit, keyName) se ejecuta aproximadamente un 30% más rápido cuando keyValueSplit y keySplit son caracteres únicos como : o ,. Con delimitadores de un solo carácter, el motor busca la clave objetivo directamente en los datos binarios sin analizar sintácticamente toda la entrada.

Patrones del operador LIKE

Escape de guiones bajos: El guion bajo (_) es un comodín de un solo carácter en SQL. Para coincidir con un guion bajo literal, utilice un carácter de escape:

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

Sin el escape, LIKE '%seller_id%' también coincide con seller#id, sellerxid, seller1id y otras cadenas no deseadas.

Evite las expresiones regulares

Las expresiones regulares pueden ser 100 veces más lentas que las operaciones aritméticas y pueden entrar en bucles infinitos en casos extremos, bloqueando los despliegues. Utilice LIKE en su lugar siempre que sea posible.

Para los casos que requieren expresiones regulares, consulte REGEXP y REGEXP_REPLACE.

Sintaxis

La sintaxis sigue el estándar Apache Calcite SQL:

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

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

Hints de join

Los hints de consulta modifican el plan de ejecución dentro del bloque de consulta actual. Actualmente, Flink solo admite hints de join, tanto para dimension table joins como para regular joins.