import pandas as pd

Kafka Connect: integrazione senza codice

Prima o poi un team deve portare dentro Kafka i dati che vivono in un database, in un SaaS o in un sistema operativo, e la tentazione è scrivere un microservizio diverso per ogni fonte. Kafka Connect promette di evitarlo offrendo un’integrazione standardizzata e configurabile. La promessa “senza codice” però va presa con cautela: il valore reale dipende da come gestisci configurazione, offset, errori e osservabilità. Questa lezione mostra dove finisce il no code e dove comincia l’ingegneria vera.

Il problema che Connect risolve

Senza Connect, ogni nuova sorgente richiede un consumer o un producer scritto a mano, con la sua logica di retry, di gestione errori e di tracciamento degli offset. Moltiplica questo per dieci sistemi e ottieni dieci pezzi di codice quasi identici da mantenere, ognuno con i suoi bug. Connect sposta quel lavoro in una configurazione dichiarativa eseguita da un runtime già pronto.

Il modo giusto di leggere un connector è come una checklist di integrazione: sorgente, destinazione, converter, transform, dead letter queue, retry e lag. Un connector è affidabile quando fallisce in modo osservabile e recuperabile, non quando nasconde la complessità dietro una configurazione di poche righe. La domanda da tenere a mente è dove possono finire i dati quando qualcosa va storto.

Source e sink connector

Connect distingue due ruoli. Un source connector legge da un sistema esterno e scrive su Kafka: è il caso di Debezium per il change data capture da PostgreSQL, o di un connector S3 che legge file. Un sink connector fa il percorso inverso, legge da Kafka e scrive su un sistema esterno come ClickHouse, S3, Elasticsearch o una destinazione JDBC.

La maggior parte delle pipeline di analytics combina i due: si cattura un cambiamento da un database operativo con un source connector e lo si riversa nel data warehouse con un sink. Il punto delicato è quasi sempre la trasformazione e la gestione degli errori nel mezzo, non il trasporto in sé.

Esempio: sink connector ClickHouse

{
  "name": "clickhouse-sink",
  "config": {
    "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
    "topics": "user_events",
    "clickhouse.url": "jdbc:clickhouse://localhost:8123",
    "clickhouse.table": "user_events",
    "tasks.max": "4",
    "errors.tolerance": "all",
    "errors.deadletterqueue.topic.name": "dlq_user_events"
  }
}

Con questa configurazione, Connect consuma automaticamente dal topic user_events, scrive su ClickHouse e dirotta gli eventi problematici sul topic dlq_user_events per ispezione. Il parametro tasks.max controlla quante istanze parallele lavorano, mentre errors.tolerance su all evita che un singolo messaggio malformato fermi l’intero connector.

Dead letter queue

La dead letter queue è il pattern che permette di gestire gli errori senza bloccare la pipeline. Quando un messaggio non può essere scritto, per esempio perché ha un formato non valido o viola un vincolo, invece di fermare il connector lo si reindirizza a un topic DLQ. Da lì un processo separato, o un analista, può ispezionarlo e correggere il problema mentre il flusso principale continua a scorrere.

È una scelta che cambia il comportamento del sistema sotto stress. Senza DLQ, un evento corrotto può fermare l’ingestione di tutto il resto; con la DLQ, l’errore resta circoscritto e visibile, e questo è esattamente quel “fallire in modo osservabile e recuperabile” che separa un’integrazione robusta da una fragile.

Single message transform

Connect supporta trasformazioni leggere sui messaggi senza bisogno di uno stream processor esterno:

"transforms": "addPrefix",
"transforms.addPrefix.type": "org.apache.kafka.connect.transforms.RegexRouter",
"transforms.addPrefix.regex": ".*",
"transforms.addPrefix.replacement": "clickhouse.$0"

Le single message transform vanno bene per operazioni semplici come rinominare un topic o aggiungere un campo. Per logica più articolata, che incrocia più messaggi o tiene stato, conviene passare a Kafka Streams o a Flink invece di forzare le SMT oltre il loro scopo.

Come impostare la scelta di integrazione

La prima domanda non è “quale connector uso” ma “quale decisione dovrà migliorare grazie a questa integrazione”. Una pipeline ha valore se riduce l’incertezza di una scelta concreta, non se aggiunge un’altra freccia tra due riquadri in un diagramma. Conviene definire in modo esplicito l’unità di lavoro (il topic, l’evento, lo schema, il connector stesso), il segnale che osservi (lag, throughput, tasso di errore, compatibilità dello schema) e la baseline rispetto a cui leggi quel segnale.

Un modo ordinato di procedere segue questa sequenza:

1. definire il problema di integrazione in linguaggio di business;
2. identificare l’unità corretta, che sia evento, record o tabella sorgente;
3. controllare se il connector cattura davvero il fenomeno o solo una parte;
4. costruire una metrica di salute interpretabile, tipicamente lag e tasso di errore;
5. segmentare per fonte e per tipo di evento, perché un problema può nascondersi in un solo canale;
6. trasformare il risultato in una raccomandazione verificabile.

Un caso di scelta

Un team vuole sincronizzare gli eventi del CRM e gli ordini nel data lake usando connector già esistenti. Prima di procedere conviene controllare il mapping degli schemi, la gestione degli errori, il throughput atteso, la gestione delle credenziali e la possibilità di replay. Un’integrazione che sembra “semplice” può diventare il punto meno governato dell’intera piattaforma proprio perché nessuno la considera codice da presidiare.

La lettura dei segnali segue la solita prudenza. Se una metrica migliora, potrebbe essere un effetto reale o una normale oscillazione, e conviene cercare un confronto e un segmento. Se un solo segmento si muove più degli altri, la media aggregata sta nascondendo una differenza, quindi vanno separate le coorti. Se il costo cresce insieme al risultato, l’impatto va valutato sul margine.

Esempio SQL: una vista di controllo

Il pattern seguente è generico ma eseguibile sulla maggior parte dei warehouse moderni. Serve a costruire una base analitica con metrica, segmento e finestra temporale, così da confrontare periodi e gruppi senza riscrivere ogni volta la logica.

WITH base_events AS (
  SELECT
    user_id,
    account_id,
    event_type,
    event_time,
    DATE_TRUNC('week', event_time) AS week,
    source,
    device_type
  FROM events
  WHERE event_time >= CURRENT_DATE - INTERVAL '180 days'
    AND user_id IS NOT NULL
),
weekly_user_metrics AS (
  SELECT
    week,
    user_id,
    COALESCE(source, 'unknown') AS source,
    COALESCE(device_type, 'unknown') AS device_type,
    COUNT(*) AS total_events,
    COUNT(DISTINCT DATE(event_time)) AS active_days,
    COUNT(DISTINCT event_type) AS event_diversity,
    MAX(CASE WHEN event_type IN ('purchase', 'subscribe', 'activation') THEN 1 ELSE 0 END) AS reached_key_outcome
  FROM base_events
  GROUP BY week, user_id, source, device_type
)
SELECT
  week,
  source,
  device_type,
  COUNT(DISTINCT user_id) AS users,
  ROUND(AVG(active_days), 2) AS avg_active_days,
  ROUND(AVG(event_diversity), 2) AS avg_event_diversity,
  ROUND(AVG(reached_key_outcome) * 100, 2) AS key_outcome_rate
FROM weekly_user_metrics
GROUP BY week, source, device_type
ORDER BY week, source, device_type;

La query non dà la risposta finale. Crea una superficie di osservazione fatta di trend, segmenti e differenze tra canali, da cui formulare ipotesi più precise sulla salute dell’integrazione.

Esempio Python: stabilità e anomalie

Una metrica utile deve essere abbastanza stabile da orientare le decisioni e abbastanza sensibile da segnalare cambiamenti reali. In Python possiamo controllare le variazioni anomale settimana su settimana, utile per accorgersi quando un connector inizia a degradare.

# df contiene: week, segment, users, key_outcome_rate
# key_outcome_rate espresso in percentuale, es. 12.4

df = df.sort_values(['segment', 'week']).copy()
df['previous_rate'] = df.groupby('segment')['key_outcome_rate'].shift(1)
df['wow_change_pp'] = df['key_outcome_rate'] - df['previous_rate']
df['rolling_mean'] = df.groupby('segment')['key_outcome_rate'].transform(
    lambda s: s.rolling(4, min_periods=2).mean()
)
df['rolling_std'] = df.groupby('segment')['key_outcome_rate'].transform(
    lambda s: s.rolling(4, min_periods=2).std()
)
df['z_score'] = (df['key_outcome_rate'] - df['rolling_mean']) / df['rolling_std']

anomalies = df[df['z_score'].abs() >= 2].sort_values('z_score')
print(anomalies[['week', 'segment', 'key_outcome_rate', 'wow_change_pp', 'z_score']])

Il valore è pratico: il controllo evita di reagire a ogni rumore casuale e segnala solo le variazioni che meritano un’indagine. Lo stesso meccanismo può alimentare alert sul lag, review settimanali e retrospettive sulle pipeline.

Errori comuni

Il primo errore è trattare il “no code” come “no engineering” e non presidiare un connector come si presidierebbe del codice. Il secondo è non controllare la qualità del dato in ingresso: schemi che cambiano, eventi duplicati, timezone incoerenti producono conclusioni false anche a valle di un’integrazione tecnicamente perfetta. Il terzo è confondere correlazione e causalità quando si interpretano le metriche di prodotto alimentate dalla pipeline.

Per ridurre questi rischi conviene mantenere almeno tre controlli minimi: una definizione esplicita della metrica di salute, un confronto per fonte e una verifica contro un periodo precedente. E soprattutto una dead letter queue su ogni sink, perché senza un posto dove far cadere gli errori, prima o poi sarà l’intera pipeline a cadere.

Riepilogo

Kafka Connect va trattato come uno strumento decisionale, non come una scorciatoia per evitare di scrivere codice. Il valore nasce quando colleghi problema, dati, metrica e azione, e quando progetti l’integrazione perché fallisca in modo visibile e recuperabile. Una buona pipeline non si misura dal fatto che oggi funziona, ma dalla chiarezza con cui ti dice quando smette di funzionare e cosa farne.

Riferimenti:

• Confluent. (2024). “Kafka Connect Documentation.” docs.confluent.io.
• ClickHouse. (2024). “Kafka Connect Sink.” clickhouse.com/docs.