Flink SQL admite joins regulares (también conocidos como joins de doble flujo) sobre tablas dinámicas. A diferencia de los joins por lotes, los resultados se actualizan continuamente conforme llegan nuevos datos a cualquiera de los flujos, lo que garantiza que la salida final sea coherente con la que produciría un join por lotes sobre los mismos datos.
Contexto
Los joins regulares almacenan en búfer todos los registros históricos de ambos flujos de entrada dentro del estado de Flink, de modo que cualquier fila entrante pueda compararse con todas las filas vistas previamente en el otro flujo. Esto implica que el estado crece de forma indefinida mientras el trabajo se mantenga en ejecución. En trabajos de larga duración con flujos de alto volumen, este crecimiento ilimitado del estado puede agotar la memoria y provocar fallos.
Para limitar el tamaño del estado, establezca un tiempo de vida (TTL) del estado mediante el parámetro table.exec.state.ttl o la sugerencia JOIN_STATE_TTL específica por flujo (consulte Hints). Tenga en cuenta que la eliminación de registros del estado puede afectar la precisión de los resultados: las filas que hayan expirado ya no coincidirán con los datos nuevos.
Si el crecimiento ilimitado del estado representa un problema, considere también como alternativas los joins de intervalo, los joins de búsqueda y los joins de ventana.
Sintaxis
tableReference [, tableReference ]*
| tableExpression [ NATURAL | INNER ] [ { LEFT | RIGHT | FULL } [ OUTER ] ] JOIN tableExpression [ joinCondition ]
| tableExpression CROSS JOIN tableExpression
| tableExpression [ CROSS | OUTER ] APPLY tableExpression
joinCondition:
ON booleanExpression
| USING '(' column [, column ]* ')'tableReference: el nombre de la tabla.tableExpression: la expresión de tabla.joinCondition: la condición de unión.
Coloque primero la tabla con menor frecuencia de actualización y al final la que tenga mayor frecuencia. Esta práctica reduce la cantidad de estado que debe evaluarse por cada fila entrante.
Tipos de join
INNER JOIN
Devuelve únicamente las filas where la condición de unión se cumple en ambos lados.
SELECT id, productName, status
FROM orders o
JOIN shipments s
ON o.id = s.orderId;LEFT JOIN
Devuelve todas las filas de la tabla izquierda. Las filas sin coincidencia en el lado derecho generan valores NULL en las columnas correspondientes a la derecha.
SELECT o.id, o.productName, s.status
FROM orders o
LEFT JOIN shipments s
ON o.id = s.orderId;RIGHT JOIN
Devuelve todas las filas de la tabla derecha. Las filas sin coincidencia en el lado izquierdo generan valores NULL en las columnas correspondientes a la izquierda.
SELECT o.id, o.productName, s.status
FROM orders o
RIGHT JOIN shipments s
ON o.id = s.orderId;FULL OUTER JOIN
Devuelve todas las filas de ambas tablas. Las filas sin coincidencia en ninguno de los lados generan valores NULL en las columnas sin correspondencia.
SELECT o.id, o.productName, s.status
FROM orders o
FULL OUTER JOIN shipments s
ON o.id = s.orderId;CROSS JOIN
Devuelve el producto cartesiano de ambas tablas; cada fila del lado izquierdo se combina con cada fila del lado derecho.
SELECT *
FROM orders o
CROSS JOIN shipments s;Sugerencias
En Ververica Runtime (VVR) 8.0.1 y versiones posteriores, utilice hints para especificar distintos valores de tiempo de vida (TTL) para los estados de los flujos izquierdo y derecho. Esto reduce la cantidad de datos de estado que deben mantenerse.
La sugerencia JOIN_STATE_TTL aplica exclusivamente a joins regulares. No es compatible con joins de búsqueda, de intervalo ni de ventana. Las sugerencias para joins regulares son experimentales y su sintaxis podría cambiar en futuras versiones.
Sintaxis
-- VVR 8.0.1 and later
SELECT /*+ JOIN_STATE_TTL('tableReference1' = 'ttl1' [, 'tableReference2' = 'ttl2']*) */ ...
-- VVR 8.0.7 and later also support the Apache Flink syntax
SELECT /*+ STATE_TTL('tableReference1' = 'ttl1' [, 'tableReference2' = 'ttl2']*) */ ...Establezca tableReference como un nombre de tabla, un nombre de vista o un alias. Si define un alias para una tabla, utilícelo también en la sugerencia.
Si especifica un TTL solo para uno de los flujos, el otro utilizará el TTL del despliegue definido por el parámetro table.exec.state.ttl. El valor predeterminado es de 1.5 días. Para más detalles, consulte Basic configurations.
Ejemplos
Todos los ejemplos utilizan JOIN_STATE_TTL para definir tiempos de vida independientes para cada flujo. VVR 8.0.7 y versiones posteriores también aceptan la sintaxis equivalente STATE_TTL que se muestra en los comentarios.
Uso de un alias
SELECT /*+ JOIN_STATE_TTL('o' = '3d', 'p' = '1d') */
o.rowtime, o.productid, o.orderid, o.units, p.name, p.unitprice
FROM Orders AS o
JOIN Products AS p
ON o.productid = p.productid;
-- VVR 8.0.7 and later
SELECT /*+ STATE_TTL('o' = '3d', 'p' = '1d') */
o.rowtime, o.productid, o.orderid, o.units, p.name, p.unitprice
FROM Orders AS o
JOIN Products AS p
ON o.productid = p.productid;Uso de un nombre de tabla
SELECT /*+ JOIN_STATE_TTL('Orders' = '3d', 'Products' = '1d') */ *
FROM Orders
JOIN Products
ON Orders.productid = Products.productid;
-- VVR 8.0.7 and later
SELECT /*+ STATE_TTL('Orders' = '3d', 'Products' = '1d') */ *
FROM Orders
JOIN Products
ON Orders.productid = Products.productid;Uso de un nombre de vista
CREATE TEMPORARY VIEW v AS
SELECT id, ...
FROM (
SELECT id, ROW_NUMBER() OVER (PARTITION BY ... ORDER BY ..) AS rn
FROM src1
WHERE ...
) tmp
WHERE rn = 1;
SELECT /*+ JOIN_STATE_TTL('v' = '1d', 'b' = '3d') */ v.* , b.*
FROM v
LEFT JOIN src2 AS b ON v.id = b.id;
-- VVR 8.0.7 and later
SELECT /*+ STATE_TTL('v' = '1d', 'b' = '3d') */ v.* , b.*
FROM v
LEFT JOIN src2 AS b ON v.id = b.id;Ejemplos
Ejemplo 1: Unión de las tablas Orders y Shipments
Datos de prueba
Tabla 1. Orders
|
id |
productName |
ordertime |
|
1 |
phone_a |
2025-05-01 10:00:00.0 |
|
2 |
notebook_x |
2025-05-01 10:02:00.0 |
|
3 |
phone_b |
2025-05-01 10:03:00.0 |
|
4 |
pad_m |
2025-05-01 10:05:00.0 |
Tabla 2. Shipments
|
shipId |
orderId |
status |
shipTime |
|
101 |
1 |
shipped |
2025-05-01 11:00:00.0 |
|
102 |
2 |
delivered |
2025-05-01 17:00:00.0 |
|
103 |
3 |
shipped |
2025-05-01 12:00:00.0 |
|
104 |
4 |
shipped |
2025-05-01 11:30:00.0 |
Instrucción de prueba
SELECT id, productName, status
FROM orders o
JOIN shipments s
ON o.id = s.orderId;Resultados de la prueba
|
id |
productName |
status |
|
1 |
phone_a |
shipped |
|
2 |
notebook_x |
delivered |
|
3 |
phone_b |
shipped |
|
4 |
pad_m |
shipped |
Ejemplo 2: Unión de las tablas datahub_stream1 y datahub_stream2
Datos de prueba
Tabla 3. datahub_stream1
|
a (BIGINT) |
b (BIGINT) |
c (VARCHAR) |
|
0 |
10 |
test11 |
|
1 |
10 |
test21 |
Tabla 4. datahub_stream2
|
a (BIGINT) |
b (BIGINT) |
c (VARCHAR) |
|
0 |
10 |
test11 |
|
1 |
10 |
test21 |
|
0 |
10 |
test31 |
|
1 |
10 |
test41 |
Instrucción de prueba
SELECT s1.c, s2.c
FROM datahub_stream1 AS s1
JOIN datahub_stream2 AS s2
ON s1.a = s2.a
WHERE s1.a = 0;Resultados de la prueba
|
s1.c (VARCHAR) |
s2.c (VARCHAR) |
|
test11 |
test11 |
|
test11 |
test31 |