Rilevamento delle anomalie ultra-veloce con la modalità in tempo reale di Apache Spark

Questo post definisce un modello riutilizzabile per carichi di lavoro operativi che fanno davvero la differenza: rilevamento delle frodi, monitoraggio dei sensori IoT, personalizzazione in tempo reale, elaborazione dei segnali di sicurezza; qualsiasi scenario in cui una risposta immediata agli eventi sia fondamentale per i risultati aziendali.

L'obiettivo principale: quando un evento appare sospetto o non valido, segnalarlo immediatamente e instradarlo per l'azione a valle appropriata.

In questo blog mostreremo il rilevamento delle anomalie sulle transazioni della blockchain di Ethereum. Analizzeremo i dati della blockchain di Ethereum e segnaleremo le transazioni con pattern non validi in tempo reale. Nello specifico, rileveremo:

1. Violazioni della qualità dei dati: blocchi in cui gas_used > gas_limit sono fisicamente impossibili secondo il protocollo Ethereum, il che indica corruzione dei dati, bug del producer o errori di parsing dello schema
2. Violazioni dell'igiene del payload: il campo extra_data contenente pattern riconoscibili di PII o credenziali (indirizzi e-mail, token JWT, chiavi di accesso AWS) segnala una fuga di dati o producer configurati in modo errato

Sebbene per la dimostrazione si utilizzino i dati di Ethereum, questa classificazione di eventi "sospetti o non validi" si applica a numerosi casi d'uso di alto valore:

• Rilevamento delle frodi: una transazione presenta pattern anomali → attiva flussi di lavoro di indagine a valle
• Monitoraggio IoT: la lettura di un sensore non rientra nei parametri fisicamente possibili → avvia una risposta automatizzata
• Operazioni di sicurezza: il payload contiene segreti o pattern PII → quarantena in tempo reale con governance unificata
• Motori di personalizzazione: rispondono a specifici eventi comportamentali con offerte contestuali immediate

La stessa logica di rilevamento si applica direttamente a qualsiasi dominio: transazioni finanziarie che contengono PII impreviste, payload IoT con letture dei sensori fuori scala o log degli eventi API contenenti segreti che avrebbero dovuto essere oscurati. Lo stream di Ethereum fornisce un set di dati pulito e riproducibile per dimostrare questo modello su scala.

Informazioni sulla modalità Real-Time di Spark

La modalità Real-Time (RTM) è un nuovo tipo di trigger per Apache Spark™ Structured Streaming che offre una latenza nell'ordine dei millisecondi alle API di Spark — senza la necessità di un motore specializzato separato come Apache Flink.

Mentre la modalità micro-batch predefinita di Structured Streaming funziona come una navetta aeroportuale che aspetta di riempirsi prima di partire, la modalità RTM funziona come un tapis roulant ad alta velocità, elaborando ogni evento non appena arriva. Raggiunge questo obiettivo attraverso tre innovazioni architetturali: flusso continuo di dati (gli eventi vengono elaborati man mano che arrivano, non in blocchi discreti), pianificazione della pipeline (tutte le fasi della query vengono eseguite simultaneamente, senza blocchi) e streaming shuffle (i dati vengono passati immediatamente tra i task in memoria, bypassando il disco).

La modalità RTM è progettata specificamente per carichi di lavoro operativi in cui la latenza influisce direttamente sui risultati aziendali: rilevamento delle frodi, personalizzazione in tempo reale, calcolo delle feature ML e monitoraggio IoT. Per i carichi di lavoro in grado di tollerare 1–2 secondi di latenza, il micro-batch tradizionale rimane la scelta più conveniente.

Perché la modalità RTM rappresenta una svolta

1. Ultra-rapida: latenza inferiore al secondo ora raggiungibile

La modalità Real-Time cambia radicalmente ciò che è possibile fare con Apache Spark. Le latenze end-to-end comprese tra circa 5 ms e circa 300 ms, a seconda della complessità del carico di lavoro, portano Spark in un territorio precedentemente dominato da motori di elaborazione di stream specializzati. Il micro-batch tradizionale offre una latenza di 1–2 secondi; la modalità Real-Time raggiunge tempi compresi tra circa 5 ms e circa 300 ms.
L'architettura ottiene questo risultato grazie a pipeline di esecuzione pre-allocate e al checkpointing asincrono, eliminando l'overhead di pianificazione che tradizionalmente limitava l'elaborazione micro-batch. Per i carichi di lavoro operativi in cui i millisecondi contano (rilevamento delle frodi, monitoraggio IoT, offerte in tempo reale), questo livello di prestazioni è rivoluzionario.

2. Stack semplificato: nessuna tecnologia separata richiesta

Le organizzazioni si scontrano spesso con un malinteso costoso: "Spark non è abbastanza performante per i casi d'uso in tempo reale, quindi abbiamo bisogno di uno stack completamente separato per questo singolo requisito".

Per i carichi di lavoro che tollerano 1–2 secondi di latenza, il micro-batch di Spark inserisce in modo affidabile i dati in Delta Lake con eccellenti caratteristiche di rapporto prezzo/prestazioni. Per i carichi di lavoro operativi che richiedono tempi di risposta inferiori al secondo, la modalità Real-Time elimina completamente la necessità di tecnologie separate — come confermato dai team di Coinbase, DraftKings e MakeMyTrip che hanno consolidato le proprie attività su un unico stack basato su Spark sia per i carichi di lavoro analitici che per quelli operativi.

Con la modalità Real-Time, Spark gestisce carichi di lavoro sia analitici (nell'ordine dei secondi) che operativi (nell'ordine dei millisecondi) all'interno di un'unica piattaforma unificata. Ciò riduce:

• Complessità operativa: un unico stack tecnologico da gestire, monitorare e risolvere i problemi
• Overhead di formazione: le competenze Spark esistenti si trasferiscono direttamente ai casi d'uso in tempo reale
• Attrito di integrazione: nessun passaggio complesso tra motori di streaming separati
• Costo totale di proprietà: il consolidamento riduce i costi operativi, di licenza e di infrastruttura

3. A misura di sviluppatore: semplice modifica del trigger, nessuna riscrittura del codice

Forse l'aspetto più interessante della modalità Real-Time è la sua straordinaria semplicità per gli sviluppatori che hanno già familiarità con Structured Streaming. L'abilitazione di questa potente funzionalità non richiede alcuna migrazione complessa o ristrutturazione fondamentale del codice.

Le organizzazioni possono sbloccare una latenza nell'ordine dei millisecondi semplicemente modificando la configurazione del trigger:

Tutto qui. La stessa familiare API Structured Streaming. La stessa gestione dei checkpoint. La stessa semantica di recapito at-least-once. Basta una sola modifica della configurazione per abilitare l'intelligence operativa inferiore al secondo.

Nota sulle garanzie di recapito: la modalità RTM con sink Kafka fornisce garanzie di recapito at-least-once. I consumer a valle dovrebbero gestire i potenziali duplicati tramite scritture idempotenti o logica di deduplicazione.

Questa integrazione fluida rappresenta un vantaggio fondamentale. I team possono creare prototipi e portare in produzione i carichi di lavoro operativi senza il notevole overhead legato all'apprendimento, alla distribuzione e alla gestione di stack tecnologici completamente separati. Questo approccio accelera notevolmente l'innovazione, riducendo al contempo il rischio tradizionalmente associato all'adozione di nuove funzionalità in tempo reale.

Dopo aver stabilito perché la modalità Real-Time è importante, esaminiamo come implementare questo modello nella pratica. Le sezioni seguenti mostrano una pipeline di guardrail pronta per la produzione — il modello operativo che trasforma queste funzionalità in valore aziendale.

Panoramica dell'architettura: creazione di uno stream di guardrail operativo

Ogni evento in entrata viene valutato immediatamente, producendo un evento a valle arricchito contenente:

• Decisione: ALLOW rispetto a QUARANTINE
• Motivi: spiegazione dettagliata di eventuali flag attivati (violazioni della qualità dei dati, problemi di igiene del payload)

Questo modello operativo funge da unica fonte di verità per il processo decisionale in tempo reale:

• Questo evento deve essere messo in quarantena per essere esaminato?
• In che modo arricchiamo gli eventi in modo che i sistemi a valle possano reagire istantaneamente?

Sebbene la nostra dimostrazione utilizzi i dati dei blocchi di Ethereum, questo pattern si applica universalmente: transazioni finanziarie, letture dei sensori, log di autenticazione, chiamate API—l'architettura rimane coerente.

Validazione su scala

In linea con il nostro impegno nel fornire soluzioni pronte per la produzione, abbiamo validato questo pattern su scala. L'intera blockchain di Ethereum—circa 95 GB distribuiti su 4 partizioni, che rappresentano circa 23 milioni di messaggi—è stata caricata in Kafka per i test.

Definizione delle regole di validazione per la classificazione degli eventi

Implementiamo regole di validazione intenzionalmente semplici e ad alto segnale:

Regola 1: Validazione della pulizia del payload

Scansiona il campo extra_data alla ricerca di pattern che indicano chiaramente dati che non dovrebbero essere presenti nei flussi di produzione. Gli esempi di codice mostrano il rilevamento di pattern di base (indirizzi e-mail, token JWT, formati di chiavi AWS).

Le organizzazioni dovrebbero sostituire queste regole con altre specifiche per i propri requisiti di conformità—pattern PII, identificatori interni, credenziali API e dati sensibili simili.

L'aspetto fondamentale: i guardrail in tempo reale devono far parte della pipeline come elemento di una governance unificata, non essere scoperti durante l'analisi post-incidente.

Regola 2: Validazione della qualità dei dati

gas_used > gas_limit

Questa condizione non dovrebbe mai verificarsi in dati validi. Quando viene rilevata, indica uno dei seguenti problemi:

• Corruzione dei dati durante la trasmissione
• Errori di generazione dei dati lato producer
• Incoerenze nel parsing dello schema
• Guasti ai sistemi a monte

Da una prospettiva operativa, questo è esattamente il tipo di anomalia che vogliamo segnalare immediatamente, consentendo una risposta rapida prima che i sistemi a valle ne risentano.

Una volta stabilita la logica di validazione, passiamo alla configurazione dello streaming che consente l'esecuzione in frazioni di secondo.

Real-Time Mode: configurazione essenziale

La Real-Time Mode si abilita tramite il trigger in tempo reale e opera in modalità di aggiornamento. In PySpark, si specifica un parametro di intervallo (ad es. "5 minutes").

Due requisiti di configurazione fondamentali:

1. Configurazione del cluster: la documentazione di Databricks specifica le impostazioni del cluster di job richieste e il flag di abilitazione RTM
2. Modalità di output: deve utilizzare la modalità update con i trigger RTM

Configurazione utilizzata in questa demo:

• Runtime: Databricks Runtime 16.4 LTS o versioni successive
• Compute: cluster dedicato (singolo utente) con numero fisso di worker (autoscaling disabilitato)
• Photon: disabilitato (non supportato con RTM)
• Worker: 4 worker per questo carico di lavoro
• Modalità di output: update (richiesta per RTM)

Consulta il repository associato per il file cluster_config.template.json completo.

Implementazione: pipeline di guardrail in tempo reale

Questa pipeline a passaggio singolo mostra l'integrazione fluida tra Kafka e Spark Real-Time Mode:

1. Connessione alla sorgente Kafka
2. Parsing dei payload JSON in entrata
3. Calcolo della decisione e delle motivazioni
4. Scrittura del JSON arricchito su Kafka

Puoi leggere il codice qui.

Risultati: prestazioni su scala

Abbiamo validato le prestazioni di Real-Time Mode elaborando i dati della blockchain di Ethereum—~23 milioni di messaggi distribuiti su 4 partizioni Kafka in modalità streaming continuo.

Prestazioni di throughput

La pipeline ha dimostrato caratteristiche di throughput notevoli:

• Frequenza di input: 65.592 righe/secondo
• Frequenza di elaborazione: 69.713 righe/secondo (sostenuta)
• Record totali elaborati: ~23.213.628 messaggi
• Configurazione del cluster: DBR 16.4 LTS, 4 worker (i3.xlarge), modalità singolo utente dedicata, Photon disabilitato

Metriche di latenza

I log del driver acquisiscono metriche di latenza dettagliate tramite l'evento spark.streaming.madeProgress. Real-Time Mode riporta processingLatencyMs, che misura il tempo che intercorre tra la lettura di un record da parte della query e la sua scrittura nel sink a valle.

Per questa pipeline di validazione stateless che elabora ~23 milioni di record, abbiamo osservato:

• P0, P50, P90, P95: 1 millisecondo (arrotondato a 0 nelle metriche) P99: 1 millisecondo Frequenza di elaborazione: 69.713 righe/secondo sostenuta
• Informazioni su processingLatencyMs: questa metrica misura il tempo che intercorre tra la lettura di un record da parte di RTM e la sua scrittura nel sink a valle. Viene misurata per task e riportata nella sezione rtmMetrics.processingLatencyMs di StreamingQueryProgress con percentili (P0, P50, P90, P95, P99). Per una pipeline Kafka-to-Kafka a singolo stadio come questa, rappresenta di fatto la latenza end-to-end per singolo record.
• Cosa significa: la stragrande maggioranza dei record (95° percentile) è stata elaborata in meno di mezzo millisecondo, e anche l'1% più lento ha completato l'elaborazione entro 1 millisecondo. I valori che mostrano "0" per P0-P95 indicano latenze inferiori a 0,5 ms (arrotondate per difetto dal sistema di metriche).

Nota: questi risultati rappresentano le prestazioni su una pipeline di validazione stateless. Operazioni stateful più complesse (aggregazioni, windowing) potrebbero registrare latenze più elevate all'interno dell'intervallo RTM da ~5 ms a ~300 ms, a seconda della complessità del carico di lavoro.

Principali considerazioni sulle prestazioni

• Interpretazione: il 99% di tutti i record è stato elaborato in meno di 1 millisecondo, con solo l'1% più lento che ha raggiunto 1 millisecondo. Ciò dimostra prestazioni a bassa latenza eccezionalmente costanti.
• Throughput: la pipeline ha sostenuto 69.713 righe/secondo durante l'elaborazione di ~23 milioni di record, dimostrando prestazioni stabili sotto carico continuo.

Queste metriche confermano che Real-Time Mode offre una latenza inferiore al millisecondo di livello enterprise (P95 0,5 ms, P99 = 1 ms) durante l'elaborazione di flussi ad alto volume a quasi 70.000 righe/secondo, eliminando il tradizionale compromesso tra piattaforme unificate e requisiti di bassa latenza.

Conclusione

Real-Time Mode estende Apache Spark™ Structured Streaming a una nuova classe di carichi di lavoro: applicazioni operative e sensibili alla latenza che richiedono una risposta immediata ai dati in streaming. Portando la latenza inferiore al secondo sulle API Spark che il tuo team già utilizza, elimina la necessità di gestire un motore specializzato separato per le pipeline più critiche in termini di tempo.

La proposta di valore è interessante:

• Un'unica piattaforma unificata gestisce carichi di lavoro sia analitici (nell'ordine dei secondi) che operativi (nell'ordine dei millisecondi)
• Le competenze Spark esistenti si trasferiscono direttamente, senza richiedere una specializzazione separata
• Rischio di migrazione minimo — una singola modifica alla configurazione del trigger sblocca le funzionalità in tempo reale
• Prestazioni validate in produzione — latenze P99 di 1 ms con P95 inferiore a 0,5 ms durante l'elaborazione di milioni di eventi

Che tu stia creando pipeline di rilevamento delle frodi, motori di personalizzazione o sistemi di calcolo delle feature ML, Real-Time Mode offre la latenza richiesta dalla tua applicazione, mantenendo al contempo la semplicità e l'ampiezza dell'ecosistema di Spark.

Inizia subito

Clona il repository associato per eseguire questa pipeline di guardrail end-to-end; include l'implementazione completa, la configurazione del cluster e la guida alla distribuzione.

Per approfondire, consulta la documentazione di Real-Time Mode per le opzioni di configurazione e le sorgenti/sink supportati, oppure guarda il video Real-Time Mode Technical Deep Dive per una panoramica completa dell'architettura.

Risorse

• Approfondimento tecnico sulla modalità Real-Time: come abbiamo integrato lo streaming inferiore a 300 millisecondi in Apache Spark™
• Garantire una latenza inferiore al secondo per i carichi di lavoro operativi su Databricks
• Documentazione della modalità Real-Time di Databricks
• Repository di codice associato - Demo dei guardrail per la modalità Real-Time - Implementazione pronta per la produzione con suite di test completa, configurazione del cluster e guida alla distribuzione

(Questo post sul blog è stato tradotto utilizzando strumenti basati sull'intelligenza artificiale) Post originale