Arricchire i dati in streaming

Apache Beam semplifica il flusso di lavoro di arricchimento dei dati fornendo una trasformazione di arricchimento chiavi in mano che puoi aggiungere alla tua pipeline. Questa pagina spiega come utilizzare la trasformazione di arricchimento di Apache Beam per arricchire i dati di streaming.

Quando arricchisci i dati, aumenti i dati non elaborati di un'origine aggiungendo dati correlati di una seconda origine. I dati aggiuntivi possono provenire da una serie di origini, come Bigtable o BigQuery. La trasformazione di arricchimento Apache Beam utilizza una ricerca chiave-valore per collegare i dati aggiuntivi ai dati non elaborati.

I seguenti esempi forniscono alcuni casi in cui l'arricchimento dei dati è utile:

  • Vuoi creare una pipeline di e-commerce che acquisisca le attività degli utenti da un sito web o un'app e fornisca consigli personalizzati. La trasformazione incorpora le attività nei dati della pipeline in modo da poter fornire i consigli personalizzati.
  • Hai dati utente che vuoi unire a dati geografici per eseguire analisi basate sulla geografia.
  • Vuoi creare una pipeline che raccolga i dati dai dispositivi IoT (Internet of Things) che inviano eventi di telemetria.

Vantaggi

La trasformazione dell'arricchimento presenta i seguenti vantaggi:

  • Trasforma i dati senza richiedere la scrittura di codice complesso o la gestione delle librerie sottostanti.
  • Fornisce gestori di origine integrati.
  • Utilizza la limitazione lato client per gestire la limitazione di frequenza delle richieste. Le richieste vengono ritardate in modo esponenziale con una strategia di nuovi tentativi predefinita. Puoi configurare limitazione di frequenza in base al tuo caso d'uso.

Supporto e limitazioni

La trasformazione dell'arricchimento presenta i seguenti requisiti:

  • Disponibile per pipeline batch e di streaming.
  • Il gestore BigTableEnrichmentHandler è disponibile nelle versioni 2.54.0 e successive dell'SDK Apache Beam Python.
  • Il gestore BigQueryEnrichmentHandler è disponibile nelle versioni 2.57.0 e successive dell'SDK Apache Beam Python.
  • Il gestore VertexAIFeatureStoreEnrichmentHandler è disponibile nelle versioni 2.55.0 e successive dell'SDK Apache Beam Python.
  • Quando utilizzi le versioni 2.55.0 e successive dell'SDK Apache Beam per Python, devi anche installare il client Python per Redis.
  • I job Dataflow devono utilizzare Portable Runner.

Utilizzare la trasformazione di arricchimento

Per utilizzare la trasformazione di arricchimento, includi il seguente codice nella pipeline:

import apache_beam as beam
from apache_beam.transforms.enrichment import Enrichment
from apache_beam.transforms.enrichment_handlers.bigtable import BigTableEnrichmentHandler

bigtable_handler = BigTableEnrichmentHandler(...)

with beam.Pipeline() as p:
  output = (p
            ...
            | "Create" >> beam.Create(data)
            | "Enrich with Bigtable" >> Enrichment(bigtable_handler)
            ...
            )

Poiché la trasformazione di arricchimento esegue un cross join per impostazione predefinita, progetta il join personalizzato per arricchire i dati di input. Questo design garantisce che l'unione includa solo i campi specificati.

Nell'esempio seguente, left è l'elemento di input della trasformazione di arricchimento e right sono i dati recuperati da un servizio esterno per questo elemento di input.

def custom_join(left: Dict[str, Any], right: Dict[str, Any]):
  enriched = {}
  enriched['FIELD_NAME'] = left['FIELD_NAME']
  ...
  return beam.Row(**enriched)

Parametri

Per utilizzare la trasformazione di arricchimento, è necessario il parametro EnrichmentHandler.

Puoi anche utilizzare un parametro di configurazione per specificare una funzione lambda per una funzione di join, un timeout, un limitatore o un ripetitore (strategia di ripetizione). Sono disponibili i seguenti parametri di configurazione:

  • join_fn: una funzione lambda che accetta dizionari come input e restituisce una riga arricchita (Callable[[Dict[str, Any], Dict[str, Any]], beam.Row]). La riga arricchita specifica come unire i dati recuperati dall'API. Il valore predefinito è un cross join.
  • timeout: il numero di secondi da attendere prima che la richiesta venga completata dall'API prima del timeout. Il valore predefinito è 30 secondi.
  • throttler: specifica il meccanismo di limitazione. L'unica opzione supportata è la limitazione adattiva lato client predefinita.
  • repeater: specifica la strategia di nuovi tentativi quando si verificano errori come TooManyRequests e TimeoutException. Il valore predefinito è ExponentialBackOffRepeater.

Passaggi successivi