Realtime Compute for Apache Flink:Actualización parcial (construcción de tablas anchas)

Por qué usar la actualización parcial para construir tablas anchas

En los pipelines de datos de streaming tradicionales, la construcción de una tabla ancha suele requerir la unión de múltiples streams o tablas mediante una clave primaria. Por ejemplo, un sistema de recomendaciones en tiempo real necesita consolidar actualizaciones de diversas fuentes, como preferencias de usuario, compras, clics y carritos de compra, para crear una vista de 360 grados del cliente.

Aunque Apache Flink puede realizar uniones de múltiples streams, este enfoque presenta varios puntos problemáticos en escenarios a gran escala:

• Tamaño de estado grande: Todos los eventos deben almacenarse en caché hasta que se complete la unión, lo que puede provocar que el estado crezca indefinidamente.

• Cuellos de botella de rendimiento: La sobrecarga elevada de checkpoints y la contrapresión frecuente pueden reducir el throughput general y la estabilidad.

• Depuración difícil: Un estado complejo y opaco hace que la resolución de problemas sea costosa y consuma mucho tiempo.

• Consistencia de datos deficiente: Establecer un Time-to-Live (TTL) puede causar que los eventos se descarten prematuramente, lo que genera resultados incorrectos.

Estos desafíos evidencian la necesidad de un nuevo método para construir tablas anchas que evite uniones complejas y reduzca significativamente el consumo de recursos y la sobrecarga de mantenimiento.

Construcción de tablas anchas con la actualización parcial de Fluss

Fluss introduce un método para construir tablas anchas que utiliza un mecanismo de actualización parcial. Esto le permite enriquecer datos con información dimensional directamente dentro de un trabajo de procesamiento de streams, lo que elimina la necesidad de una unión compleja de múltiples streams.

A diferencia de los métodos tradicionales, Fluss permite que cada fuente de datos actualice de forma independiente su propio conjunto de columnas en una sola tabla ancha basándose en una clave primaria. Por ejemplo, puede definir una tabla user_profile con user_id como clave primaria, que contenga todas las columnas posibles. Cada stream de datos solo necesita escribir los datos parciales que posee, y el motor de almacenamiento de Fluss fusiona automáticamente estas actualizaciones por clave primaria.

El mecanismo subyacente funciona de la siguiente manera: cuando llega una actualización parcial, Fluss busca el registro existente para esa clave primaria y actualiza únicamente las columnas proporcionadas en el evento actual, dejando las demás columnas sin cambios. La nueva versión fusionada se escribe en disco en tiempo real, lo que garantiza que cada registro siempre esté en su estado más reciente y completo. Este proceso es similar a una actualización incremental en una base de datos, donde solo se modifican los campos cambiados en lugar de todo el registro. Este enfoque reduce drásticamente la carga de estado y la complejidad computacional del trabajo de Flink, haciendo que el pipeline de datos sea más ligero, eficiente y fácil de mantener.

Ejemplo

-- Recommendations – model scores
CREATE TABLE IF NOT EXISTS `fluss-demo`.fluss.recommendations (
    user_id  STRING,
    item_id  STRING,
    rec_score DOUBLE,
    rec_ts   TIMESTAMP(3),
    PRIMARY KEY (user_id, item_id) NOT ENFORCED
) WITH ('bucket.num' = '3');
-- Impressions – how often we showed something
CREATE TABLE IF NOT EXISTS `fluss-demo`.fluss.impressions (
    user_id STRING,
    item_id STRING,
    imp_cnt INT,
    imp_ts  TIMESTAMP(3),
    PRIMARY KEY (user_id, item_id) NOT ENFORCED
) WITH ('bucket.num' = '3');
-- Clicks – user engagement
CREATE TABLE IF NOT EXISTS `fluss-demo`.fluss.clicks (
    user_id  STRING,
    item_id  STRING,
    click_cnt INT,
    clk_ts    TIMESTAMP(3),
    PRIMARY KEY (user_id, item_id) NOT ENFORCED
) WITH ('bucket.num' = '3');
-- Result wide table
CREATE TABLE IF NOT EXISTS `fluss-demo`.fluss.user_rec_wide (
    user_id   STRING,
    item_id   STRING,
    rec_score DOUBLE,   -- updated by recs stream
    imp_cnt   INT,      -- updated by impressions stream
    click_cnt INT,      -- updated by clicks stream
    PRIMARY KEY (user_id, item_id) NOT ENFORCED
) WITH ('bucket.num' = '3');

-- Recommendations – model scores
INSERT INTO `fluss-demo`.fluss.recommendations VALUES
    ('user_101','prod_501',0.92 , TIMESTAMP '2025-05-16 09:15:02'),
    ('user_101','prod_502',0.78 , TIMESTAMP '2025-05-16 09:15:05'),
    ('user_102','prod_503',0.83 , TIMESTAMP '2025-05-16 09:16:00'),
    ('user_103','prod_501',0.67 , TIMESTAMP '2025-05-16 09:16:20'),
    ('user_104','prod_504',0.88 , TIMESTAMP '2025-05-16 09:16:45');
-- Impressions – how often each (user,item) was shown
INSERT INTO `fluss-demo`.fluss.impressions VALUES
    ('user_101','prod_501', 3, TIMESTAMP '2025-05-16 09:17:10'),
    ('user_101','prod_502', 1, TIMESTAMP '2025-05-16 09:17:15'),
    ('user_102','prod_503', 7, TIMESTAMP '2025-05-16 09:18:22'),
    ('user_103','prod_501', 4, TIMESTAMP '2025-05-16 09:18:30'),
    ('user_104','prod_504', 2, TIMESTAMP '2025-05-16 09:18:55');
-- Clicks – user engagement
INSERT INTO `fluss-demo`.fluss.clicks VALUES
    ('user_101','prod_501', 1, TIMESTAMP '2025-05-16 09:19:00'),
    ('user_101','prod_502', 2, TIMESTAMP '2025-05-16 09:19:07'),
    ('user_102','prod_503', 1, TIMESTAMP '2025-05-16 09:19:12'),
    ('user_103','prod_501', 1, TIMESTAMP '2025-05-16 09:19:20'),
    ('user_104','prod_504', 1, TIMESTAMP '2025-05-16 09:19:25');