08 - Kafka + Apache Flink: Pipeline Analytics Real-Time e Iceberg Sink

Quando Kafka Streams non basta: leggere da Kafka e scrivere in un data lake

Hai già un flusso di ordini su Kafka e un servizio che lo elabora con Kafka Streams. Poi arrivano richieste di un altro tipo: unire il flusso degli ordini a quello dei pagamenti con un vincolo di tempo, calcolare finestre su più topic, e scrivere il risultato in un formato che gli analisti interrogano con SQL, non in un altro topic. Per questo secondo lavoro si usa spesso Apache Flink: un motore di elaborazione distribuito, con un cluster proprio, che legge da Kafka e scrive altrove. La domanda utile non è «quale è migliore», ma «che cosa cambia se lo scelgo»: dove gira, chi tiene lo stato, cosa succede se si ferma una partizione.

Questo articolo si concentra su ciò che puoi capire senza un cluster Flink: il tempo dell'evento e i watermark, che decidono quando una finestra è chiusa. Lo dimostra un modello in Java 21 che ho compilato ed eseguito. Il resto (la sintassi SQL per tabelle Kafka, finestre, tabella Iceberg) è uno schema da adattare, non eseguito, perché richiede Flink e Kafka: la sintassi viene dalla documentazione di Flink 2.3 (versione «stable») e di Apache Iceberg 1.12, consultate il 2026-11-12. Non riporto benchmark né cifre di throughput.

Tempo dell'evento e watermark: quando una finestra è chiusa?

Una finestra «ordini dei minuti 0-5» si può chiudere solo quando sai che non arriveranno altri eventi di quei minuti. Ma gli eventi arrivano in ordine sparso. Flink risolve il problema con i watermark: un segnale che dice «non arriveranno più eventi con tempo inferiore a T». Nel DDL SQL lo dichiari con WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND, che significa: considera ammissibile un disordine fino a 5 secondi. La documentazione di Flink aggiunge una regola decisiva per Kafka: quando un operatore ha più ingressi, il suo tempo è il minimo dei tempi degli ingressi, e per la sorgente Kafka il watermark in uscita è il minimo tra i watermark delle partizioni lette. Se una partizione non riceve eventi, il suo watermark non avanza e blocca tutto; il rimedio documentato è l'opzione table.exec.source.idle-timeout.

Il modello applica queste regole a un operatore con due partizioni e finestre tumbling da 5 minuti. Il watermark ammette un minuto di disordine.

import java.util.*;

public class Watermark {
    static final long MIN = 60_000L, RITARDO = 1 * MIN;            // il watermark ammette 1 minuto di disordine

    // Un "operatore" con due partizioni in ingresso: il suo tempo è il MINIMO dei watermark delle partizioni.
    static class Operatore {
        final long[] massimo = {Long.MIN_VALUE, Long.MIN_VALUE};   // evento più recente visto per ogni partizione
        final boolean[] inattiva = {false, false};
        final Map<Long, Integer> conteggi = new TreeMap<>();       // finestra (inizio, in minuti) -> conteggio
        final Set<Long> chiuse = new TreeSet<>();
        int scartati = 0;

        long watermark() {
            long w = Long.MAX_VALUE;
            for (int p = 0; p < 2; p++) {
                if (inattiva[p]) continue;                          // una partizione dichiarata inattiva non frena più
                w = Math.min(w, massimo[p] == Long.MIN_VALUE ? Long.MIN_VALUE : massimo[p] - RITARDO);
            }
            return w;
        }

        void ricevi(int partizione, long minuto) {
            long t = minuto * MIN, inizio = (t / (5 * MIN)) * 5;     // finestre tumbling da 5 minuti
            if (chiuse.contains(inizio)) { scartati++; return; }    // la finestra è già stata emessa: ritardatario
            conteggi.merge(inizio, 1, Integer::sum);
            massimo[partizione] = Math.max(massimo[partizione], t);
            for (long i : new ArrayList<>(conteggi.keySet()))       // emette le finestre chiuse dal watermark
                if (watermark() >= (i + 5) * MIN && chiuse.add(i))
                    System.out.println("   emessa finestra [" + i + "," + (i + 5) + ") = " + conteggi.get(i));
        }
    }

    public static void main(String[] args) {
        System.out.println("== entrambe le partizioni ricevono eventi");
        Operatore a = new Operatore();
        a.ricevi(0, 1); a.ricevi(1, 2); a.ricevi(0, 7); a.ricevi(1, 8); a.ricevi(0, 4);
        System.out.println("   ritardatari scartati: " + a.scartati);

        System.out.println("== la partizione 1 smette di ricevere dopo il minuto 2");
        Operatore b = new Operatore();
        b.ricevi(1, 2);
        for (long m = 1; m <= 20; m++) b.ricevi(0, m);
        System.out.println("   finestre emesse: " + b.chiuse + "  (watermark = " + b.watermark() / MIN + " min)");

        System.out.println("== stessa situazione, partizione 1 dichiarata inattiva");
        b.inattiva[1] = true;
        b.ricevi(0, 21);
        System.out.println("   finestre emesse: " + b.chiuse + "  (watermark = " + b.watermark() / MIN + " min)");
    }
}

// Output:
// == entrambe le partizioni ricevono eventi
//    emessa finestra [0,5) = 2
//    ritardatari scartati: 1
// == la partizione 1 smette di ricevere dopo il minuto 2
//    finestre emesse: []  (watermark = 1 min)
// == stessa situazione, partizione 1 dichiarata inattiva
//    emessa finestra [0,5) = 5
//    emessa finestra [5,10) = 5
//    emessa finestra [10,15) = 5
//    emessa finestra [15,20) = 5
//    finestre emesse: [0, 5, 10, 15]  (watermark = 20 min)

Nella prima scena la finestra [0,5) si chiude solo quando il minuto 8 arriva dalla partizione 1 e porta il watermark a 6 (il minimo tra 7-1 e 8-1): ha contato 2 eventi, i minuti 1 e 2. L'evento col minuto 4, arrivato dopo, trova la finestra già emessa e viene scartato (ritardatari scartati: 1). È il prezzo di ogni watermark: più è stretto, prima emetti; più è largo, meno eventi tardivi perdi. Nella seconda scena la partizione 1 si ferma dopo il minuto 2 mentre la 0 prosegue fino al minuto 20: nessuna finestra viene emessa, e il watermark resta a 1 minuto. Nella terza dichiari la partizione 1 inattiva (è ciò che fa l'idle-timeout di Flink): il watermark salta a 20 e le quattro finestre accumulate escono tutte insieme.

La conseguenza pratica è che un topic con partizioni sbilanciate o vuote è il tipico motivo per cui una query Flink «non produce nulla». Il problema non è il tuo SQL: è un watermark fermo. Ricordalo quando i risultati non compaiono: guarda prima il watermark, poi la query. Il parallelo con Kafka Streams è utile ma non identico: lì, come visto nell'articolo su Streams, il tempo che scorre è quello dello stream (il massimo timestamp visto) e il margine è il grace period.

Le tabelle Kafka in Flink SQL, scritte come vuole la documentazione

In Flink SQL un topic è una tabella. Il connector kafka richiede almeno topic, properties.bootstrap.servers e un format; properties.group.id è l'identità del gruppo; scan.startup.mode decide da dove partire. Per scrivere un risultato aggiornabile (una riga che cambia nel tempo) esiste il connector upsert-kafka, per il quale la documentazione raccomanda di dichiarare la chiave primaria nel DDL, e che richiede formati separati per chiave e valore (key.format, value.format). Lo schema qui sotto mette insieme le due cose: una tabella sorgente con il watermark, una tabella di destinazione di tipo upsert, e la query continua che le collega.

-- Schema da adattare (non eseguito: serve Flink SQL e un cluster Kafka). Sintassi dalla documentazione Flink 2.3.
CREATE TABLE ordini (
    order_id   STRING,
    user_id    STRING,
    amount     DECIMAL(10, 2),
    created_at BIGINT,                                       -- millisecondi, dal payload
    event_time AS TO_TIMESTAMP_LTZ(created_at, 3),           -- il tempo dell'evento, non quello di arrivo
    WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
) WITH (
    'connector' = 'kafka',
    'topic' = 'orders',
    'properties.bootstrap.servers' = 'kafka-1:9092',
    'properties.group.id' = 'flink-analytics',
    'scan.startup.mode' = 'earliest-offset',
    'format' = 'json'
);

-- Il risultato va su un topic "upsert": serve la chiave primaria, e un formato per chiave e uno per valore.
CREATE TABLE spesa_per_finestra (
    window_start TIMESTAMP(3),
    user_id      STRING,
    ordini       BIGINT,
    totale       DECIMAL(12, 2),
    PRIMARY KEY (window_start, user_id) NOT ENFORCED
) WITH (
    'connector' = 'upsert-kafka',
    'topic' = 'spesa-per-finestra',
    'properties.bootstrap.servers' = 'kafka-1:9092',
    'key.format' = 'json',
    'value.format' = 'json'
);

INSERT INTO spesa_per_finestra
SELECT window_start, user_id, COUNT(*), SUM(amount)
FROM TABLE(TUMBLE(TABLE ordini, DESCRIPTOR(event_time), INTERVAL '10' MINUTES))
GROUP BY window_start, window_end, user_id;

Tre punti da notare. La funzione TUMBLE restituisce la relazione originale più le colonne window_start, window_end e window_time, ed è per questo che il GROUP BY le include. La chiave primaria dell'uscita, (window_start, user_id), è ciò che permette all'upsert di sostituire il valore di una finestra quando arrivano dati tardivi che la correggono. E NOT ENFORCED è una dichiarazione, non un controllo: Flink non verifica l'unicità, sei tu a garantirla. Se ti serve unire il flusso degli ordini a quello dei pagamenti, la forma giusta è l'interval join: richiede almeno un predicato di uguaglianza e una condizione che limiti il tempo da entrambi i lati, per esempio o.order_time BETWEEN s.ship_time - INTERVAL '4' HOUR AND s.ship_time come nella documentazione. Funziona solo su tabelle che solo aggiungono righe e con attributi temporali, e grazie ai limiti di tempo Flink può liberare lo stato vecchio.

Garanzie: Flink, checkpoint e transazioni Kafka

Flink salva a intervalli (execution.checkpointing.interval, nessun valore predefinito: devi impostarlo) uno checkpoint dello stato e delle posizioni di lettura. Se un nodo cade, il job riparte dall'ultimo checkpoint. La documentazione dei checkpoint prevede la scelta tra exactly-once, preferibile per la maggior parte delle applicazioni, e at-least-once, adatto ad applicazioni a latenza bassissima. Verso Kafka, la scrittura exactly-once è un'opzione del sink: sink.delivery-guarantee vale per default at-least-once e accetta exactly-once e none; con exactly-once devi impostare anche sink.transactional-id-prefix, perché Flink usa le transazioni di Kafka. Il corollario è quello dell'articolo sulle transazioni: i consumatori read_committed a valle vedranno i dati solo dopo la conferma della transazione, quindi con la latenza di un checkpoint.

Iceberg: la tabella che Flink scrive a ogni checkpoint

Apache Iceberg è un formato di tabella per file su storage condiviso: aggiunge metadati e snapshot sopra i file di dati, così che più motori leggano la stessa tabella. La documentazione di Iceberg dice che il sink Flink «guarantees exactly-once semantics» e spiega il meccanismo: i dati scritti restano file temporanei finché il checkpoint non si completa, e solo allora Flink fa il commit di Iceberg. Due conseguenze. Primo: i dati diventano visibili ai lettori all'intervallo di checkpoint, non a ogni riga. Secondo: la documentazione avverte che un commit Iceberg può fallire mentre il checkpoint Flink riesce (o che la notifica di completamento non arrivi), e consiglia di allarmare sul tempo trascorso dall'ultimo commit riuscito.

-- Schema da adattare (non eseguito): catalogo e tabella Iceberg da Flink SQL, sintassi dalla documentazione Iceberg 1.12.
CREATE CATALOG lago WITH (
    'type' = 'iceberg',
    'catalog-type' = 'hadoop',
    'warehouse' = 'file:///data/warehouse',      -- un percorso del tuo storage condiviso
    'property-version' = '1'
);

CREATE DATABASE IF NOT EXISTS lago.analitica;

CREATE TABLE lago.analitica.spesa_per_finestra (
    window_start TIMESTAMP(3),
    user_id      STRING,
    ordini       BIGINT,
    totale       DECIMAL(12, 2)
) PARTITIONED BY (user_id);

-- Un job continuo: i dati diventano visibili in tabella quando Flink completa un checkpoint.
INSERT INTO lago.analitica.spesa_per_finestra
SELECT window_start, user_id, COUNT(*), SUM(amount)
FROM TABLE(TUMBLE(TABLE ordini, DESCRIPTOR(event_time), INTERVAL '10' MINUTES))
GROUP BY window_start, window_end, user_id;

Il catalogo hadoop usa una directory come registro dei metadati ed è il più semplice per provare; in un'infrastruttura condivisa si usano catalogo Hive o REST (stessa sintassi CREATE CATALOG con catalog-type diverso). Una limitazione scritta nella documentazione: Iceberg ha il partizionamento nascosto, ma da Flink DDL non puoi partizionare per una funzione di colonna, solo per colonne. Un'altra nota importante: la documentazione avverte che scaduto lo snapshot o eliminati i file orfani si può corrompere lo stato dei job Flink in scrittura, perché questi ricordano l'ultimo checkpoint committato nel riepilogo dello snapshot: la manutenzione della tabella va fatta con attenzione.

Flink o Kafka Streams? Il criterio

Kafka Streams è una libreria dentro la tua applicazione: nessun cluster aggiuntivo, ma il suo mondo è Kafka (legge e scrive topic, con le garanzie viste). Flink è un sistema con un cluster proprio: costa operazioni in più, ma legge e scrive molti sistemi (Kafka, tabelle Iceberg, database), ha SQL come interfaccia e scala indipendentemente dal servizio applicativo. Se il tuo problema è arricchire eventi dentro un microservizio Java, Streams basta e costa meno. Se il problema è mettere insieme più sorgenti, produrre tabelle per l'analisi e dare SQL a chi non scrive Java, Flink è la strada, con i suoi costi di gestione. In entrambi i casi, il tempo dell'evento e il suo margine di ritardo sono la scelta più importante che farai, perché decidono quali dati perdi e quanto aspetti.

Che cosa sai fare adesso

Sai spiegare perché una finestra si chiude con un watermark e non con l'orologio, che cosa vuol dire che il watermark è il minimo tra le partizioni e perché una partizione vuota può bloccare una query. Sai scrivere in Flink SQL una tabella Kafka con watermark, una finestra tumbling e un'uscita upsert con chiave primaria, e un interval join con vincoli di tempo. Sai che l'exactly-once verso Kafka passa dalle transazioni e verso Iceberg dal checkpoint, e quale latenza ne consegue. Quando vuoi portare gli eventi dal database a Kafka prima di analizzarli, vedi Kafka Connect e Debezium; quando vuoi controllare che la pipeline funzioni, il monitoraggio.