Best practice per le pipeline Dataflow

Questa pagina descrive le best practice da utilizzare durante lo sviluppo delle pipeline Dataflow. L'utilizzo di queste best practice offre i seguenti vantaggi:

• Migliora l'osservabilità e il rendimento della pipeline
• Maggiore produttività degli sviluppatori
• Migliorare la testabilità della pipeline

Gli esempi di codice Apache Beam in questa pagina utilizzano Java, ma i contenuti si applicano agli SDK Apache Beam Java, Python e Go.

Domande da considerare

Quando progetti la pipeline, considera le seguenti domande:

• Dove vengono archiviati i dati di input della pipeline? Quanti set di dati di input hai?
• Come si presentano i tuoi dati?
• Cosa vuoi fare con i tuoi dati?
• Dove devono essere inviati i dati di output della pipeline?
• Il tuo job Dataflow utilizza Assured Workloads?

Utilizzare i modelli

Per accelerare lo sviluppo della pipeline, anziché crearla scrivendo codice Apache Beam, utilizza un modello Dataflow, se possibile. I modelli presentano i seguenti vantaggi:

• I modelli sono riutilizzabili.
• I modelli ti consentono di personalizzare ogni job modificando parametri specifici della pipeline.
• Chiunque a cui concedi le autorizzazioni può utilizzare il modello per eseguire il deployment della pipeline. Ad esempio, uno sviluppatore può creare un job da un modello e un data scientist dell'organizzazione può eseguire il deployment del modello in un secondo momento.

Puoi utilizzare un modello fornito da Google oppure crearne uno personalizzato. Alcuni modelli forniti da Google ti consentono di aggiungere una logica personalizzata come passaggio della pipeline. Ad esempio, il modello Pub/Sub a BigQuery fornisce un parametro per eseguire una funzione JavaScript definita dall'utente (UDF) memorizzata in Cloud Storage.

Poiché i modelli forniti da Google sono open source ai sensi della licenza Apache 2.0, puoi utilizzarli come base per nuove pipeline. I modelli sono utili anche come esempi di codice. Visualizza il codice del modello nel repository GitHub.

Assured Workloads

Assured Workloads contribuisce a far rispettare i requisiti di sicurezza e conformità per i clientiGoogle Cloud . Ad esempio, Regioni e assistenza nell'UE con controlli di sovranità contribuisce a garantire la residenza dei dati e la sovranità dei dati per i clienti con sede nell'UE. Per fornire queste funzionalità, alcune funzionalità di Dataflow sono limitate o soggette a restrizioni. Se utilizzi Assured Workloads con Dataflow, tutte le risorse a cui accede la pipeline devono trovarsi nel progetto o nella cartella Assured Workloads della tua organizzazione. Le risorse includono:

• Bucket Cloud Storage
• Set di dati di BigQuery
• Argomenti e sottoscrizioni Pub/Sub
• Set di dati Firestore
• Connettori I/O

In Dataflow, per i job di streaming creati dopo il 7 marzo 2024, tutti i dati utente vengono criptati con CMEK.

Per i job di streaming creati prima del 7 marzo 2024, le chiavi dei dati utilizzate nelle operazioni basate sulle chiavi, come la creazione di finestre, il raggruppamento e l'unione, non sono protette dalla crittografia CMEK. Per attivare questa crittografia per i tuoi job, svuota o annulla il job e poi riavvialo. Per saperne di più, consulta Crittografia degli artefatti di stato della pipeline.

Condividere i dati tra le pipeline

Non esiste un meccanismo di comunicazione tra pipeline specifico di Dataflow per la condivisione di dati o del contesto di elaborazione tra le pipeline. Puoi utilizzare un archivio durevole come Cloud Storage o una cache in memoria come App Engine per condividere i dati tra le istanze della pipeline.

Pianifica job

Puoi automatizzare l'esecuzione della pipeline nei seguenti modi:

• Utilizza le pipeline di dati Dataflow per creare pianificazioni di job ricorrenti e capire dove vengono utilizzate le risorse su più esecuzioni di job.
• Utilizza Cloud Scheduler.
• Utilizza l' operatore Dataflow di Apache Airflow, uno dei numerosi Google Cloud operatori in un workflow di Managed Service for Apache Airflow.
• Esegui processi di job personalizzati (cron) su Compute Engine.

Best practice per la scrittura del codice della pipeline

Le sezioni seguenti forniscono le best practice da utilizzare quando crei pipeline scrivendo codice Apache Beam.

Strutturare il codice Apache Beam

Per creare pipeline, è comune utilizzare la trasformazione ParDo di elaborazione parallela Apache Beam generica. Quando applichi una trasformazione ParDo, fornisci il codice sotto forma di oggetto DoFn. DoFn è una classe SDK Apache Beam che definisce una funzione di elaborazione distribuita.

Puoi considerare il codice DoFn come piccole entità indipendenti: possono potenzialmente essere molte istanze in esecuzione su macchine diverse, ognuna senza conoscenza delle altre. Pertanto, ti consigliamo di creare funzioni pure, che sono ideali per la natura parallela e distribuita degli elementi DoFn. Le funzioni pure hanno le seguenti caratteristiche:

• Le funzioni pure non dipendono da uno stato nascosto o esterno.
• Non hanno effetti collaterali osservabili.
• Sono deterministici.

Il modello di funzione pura non è strettamente rigido. Quando il codice non dipende da elementi non garantiti dal servizio Dataflow, le informazioni sullo stato o i dati di inizializzazione esterni possono essere validi per DoFn e altri oggetti funzione.

Quando strutturi le trasformazioni ParDo e crei gli elementi DoFn, considera le seguenti linee guida:

• Quando utilizzi l'elaborazione exactly-once, il servizio Dataflow elabora ogni elemento dell'PCollection di input da un'istanza DoFn esattamente una volta.
• Il servizio Dataflow non garantisce il numero di volte in cui viene richiamato un DoFn.
• Il servizio Dataflow non garantisce la modalità esatta di raggruppamento degli elementi distribuiti. Non garantisce quali elementi, se presenti, vengono elaborati insieme.
• Il servizio Dataflow non garantisce il numero esatto di istanze DoFn create nel corso di una pipeline.
• Il servizio Dataflow è a tolleranza di errore e potrebbe riprovare a eseguire il codice più volte se i worker riscontrano problemi.
• Il servizio Dataflow potrebbe creare copie di backup del tuo codice. Potrebbero verificarsi problemi con gli effetti collaterali manuali, ad esempio se il codice si basa su file temporanei con nomi non univoci o li crea.
• Il servizio Dataflow serializza l'elaborazione degli elementi per istanza DoFn. Il codice non deve essere rigorosamente thread-safe, ma qualsiasi stato condiviso tra più istanze di DoFn deve essere thread-safe.

Crea librerie di trasformazioni riutilizzabili

Il modello di programmazione Apache Beam consente di riutilizzare le trasformazioni. Creando una libreria condivisa di trasformazioni comuni, puoi migliorare la riutilizzabilità, la testabilità e la proprietà del codice da parte di team diversi.

Considera i due esempi di codice Java riportati di seguito, che leggono entrambi gli eventi di pagamento. Supponendo che entrambe le pipeline eseguano la stessa elaborazione, possono utilizzare le stesse trasformazioni tramite una libreria condivisa per i passaggi di elaborazione rimanenti.

Il primo esempio proviene da un'origine Pub/Sub senza limiti:

PipelineOptions options = PipelineOptionsFactory.create();

Pipeline p = Pipeline.create(options)

// Initial read transform
PCollection<PaymentEvent> payments =
    p.apply("Read from topic",
        PubSubIO.readStrings().withTimestampAttribute(...).fromTopic(...))
        .apply("Parse strings into payment events",
            ParDo.of(new ParsePaymentEventFn()));

Il secondo esempio proviene da un'origine database relazionale delimitata:

PipelineOptions options = PipelineOptionsFactory.create();

Pipeline p = Pipeline.create(options);

PCollection<PaymentEvent> payments =
    p.apply(
        "Read from database table",
        JdbcIO.<PaymentEvent>read()
            .withDataSourceConfiguration(...)
            .withQuery(...)
            .withRowMapper(new RowMapper<PaymentEvent>() {
              ...
            }));

Il modo in cui implementi le best practice per il riutilizzo del codice varia in base al linguaggio di programmazione e allo strumento di compilazione. Ad esempio, se utilizzi Maven, puoi separare il codice di trasformazione in un modulo separato. Puoi quindi includere il modulo come sottomodulo in progetti multimodulo più grandi per pipeline diverse, come mostrato nel seguente esempio di codice:

// Reuse transforms across both pipelines
payments
    .apply("ValidatePayments", new PaymentTransforms.ValidatePayments(...))
    .apply("ProcessPayments", new PaymentTransforms.ProcessPayments(...))
  ...

Per saperne di più, consulta le seguenti pagine della documentazione di Apache Beam:

• Requisiti per la scrittura del codice utente per le trasformazioni Apache Beam
• Guida di stile per PTransform: una guida di stile per gli autori di nuove raccolte riutilizzabili di PTransform

Utilizzare le code dead letter per la gestione degli errori

A volte la pipeline non riesce a elaborare gli elementi. I problemi relativi ai dati sono una causa comune. Ad esempio, un elemento che contiene dati JSON formattati in modo errato può causare errori di analisi.

Sebbene sia possibile rilevare le eccezioni all'interno del metodo DoFn.ProcessElement, registrare l'errore ed eliminare l'elemento, questo approccio comporta la perdita dei dati e impedisce di esaminarli in un secondo momento per la gestione manuale o la risoluzione dei problemi.

Utilizza invece un pattern chiamato coda di messaggi non recapitabili (coda di messaggi non elaborati). Rileva le eccezioni nel metodo DoFn.ProcessElement e registra gli errori. Anziché eliminare l'elemento non riuscito, utilizza gli output di ramificazione per scrivere gli elementi non riusciti in un oggetto PCollection separato. Questi elementi vengono quindi scritti in un data sink per essere ispezionati e gestiti in un secondo momento con una trasformazione separata.

Il seguente esempio di codice Java mostra come implementare il pattern della coda di messaggi non recapitabili.

TupleTag<Output> successTag = new TupleTag<>() {};
TupleTag<Input> deadLetterTag = new TupleTag<>() {};

PCollection<Input> input = /* ... */;

PCollectionTuple outputTuple =
    input.apply(ParDo.of(new DoFn<Input, Output>() {
      @Override
      void processElement(ProcessContext c) {
        try {
          c.output(process(c.element()));
        } catch (Exception e) {
          LOG.severe("Failed to process input {} -- adding to dead-letter file",
              c.element(), e);
          c.sideOutput(deadLetterTag, c.element());
        }
      }).withOutputTags(successTag, TupleTagList.of(deadLetterTag)));

// Write the dead-letter inputs to a BigQuery table for later analysis
outputTuple.get(deadLetterTag)
    .apply(BigQueryIO.write(...));

// Retrieve the successful elements...
PCollection<Output> success = outputTuple.get(successTag);
// and continue processing ...

Utilizza Cloud Monitoring per applicare diverse politiche di monitoraggio e avviso per la coda di messaggi non recapitabili della pipeline. Ad esempio, puoi visualizzare il numero e le dimensioni degli elementi elaborati dalla trasformazione dead letter e configurare gli avvisi da attivare se vengono soddisfatte determinate condizioni di soglia.

Gestire le mutazioni dello schema

Puoi gestire i dati con schemi imprevisti ma validi utilizzando un pattern di coda dei messaggi non recapitabili, che scrive gli elementi non riusciti in un oggetto PCollection separato. In alcuni casi, vuoi gestire automaticamente gli elementi che riflettono uno schema modificato come elementi validi. Ad esempio, se lo schema di un elemento riflette una mutazione come l'aggiunta di nuovi campi, puoi adattare lo schema del data sink per accogliere le mutazioni.

La mutazione automatica dello schema si basa sull'approccio di output ramificato utilizzato dal pattern dead letter. Tuttavia, in questo caso attiva una trasformazione che modifica lo schema di destinazione ogni volta che vengono rilevati schemi additivi. Per un esempio di questo approccio, consulta Come gestire gli schemi JSON mutanti in una pipeline di streaming, con Square Enix sul blog Google Cloud .

Decidere come unire i set di dati

L'unione dei set di dati è un caso d'uso comune per le pipeline di dati. Puoi utilizzare input laterali o la trasformazione CoGroupByKey per eseguire join nella pipeline. Ognuno presenta vantaggi e svantaggi.

Gli input secondari offrono un modo flessibile per risolvere problemi comuni di elaborazione dei dati, come l'arricchimento dei dati e le ricerche basate su chiavi. A differenza degli oggetti PCollection, gli input secondari sono mutabili e possono essere determinati in fase di runtime. Ad esempio, i valori di un input laterale potrebbero essere calcolati da un altro ramo della pipeline o determinati chiamando un servizio remoto.

Dataflow supporta gli input secondari persistendo i dati in un'archiviazione permanente, in modo simile a un disco condiviso. Questa configurazione rende disponibile l'input laterale completo a tutti i worker.

Tuttavia, le dimensioni degli input secondari possono essere molto grandi e potrebbero non rientrare nella memoria del worker. La lettura da un input secondario di grandi dimensioni può causare problemi di prestazioni se i worker devono leggere costantemente dallo spazio di archiviazione permanente.

La trasformazione CoGroupByKey è una trasformazione Apache Beam di base che unisce (appiattisce) più oggetti PCollection e raggruppa gli elementi che hanno una chiave comune. A differenza di un input secondario, che rende disponibili tutti i dati dell'input secondario a ogni worker, CoGroupByKey esegue un'operazione di rimescolamento (raggruppamento) per distribuire i dati tra i worker. CoGroupByKey è quindi ideale quando gli oggetti PCollection che vuoi unire sono molto grandi e non rientrano nella memoria del worker.

Segui queste linee guida per decidere se utilizzare gli input laterali o CoGroupByKey:

• Utilizza gli input secondari quando uno degli oggetti PCollection che stai unendo è sproporzionatamente più piccolo degli altri e l'oggetto PCollection più piccolo rientra nella memoria del worker. Il caching dell'input secondario interamente in memoria rende il recupero degli elementi rapido ed efficiente.
• Utilizza gli input aggiuntivi quando hai un oggetto PCollection che deve essere unito più volte nella pipeline. Anziché utilizzare più trasformazioni CoGroupByKey, crea un unico input laterale che possa essere riutilizzato da più trasformazioni ParDo.
• Utilizza CoGroupByKey se devi recuperare una parte consistente di un oggetto PCollection che supera significativamente la memoria del worker.

Per saperne di più, vedi Risolvere gli errori di Dataflow per esaurimento della memoria.

Ridurre al minimo le operazioni costose per elemento

Un'istanza DoFn elabora batch di elementi chiamati bundle, che sono unità di lavoro atomiche composte da zero o più elementi. I singoli elementi vengono poi elaborati dal metodo DoFn.ProcessElement, che viene eseguito per ogni elemento. Poiché il metodo DoFn.ProcessElement viene chiamato per ogni elemento, qualsiasi operazione dispendiosa in termini di tempo o di calcolo richiamata da questo metodo viene eseguita per ogni singolo elemento elaborato dal metodo.

Se devi eseguire operazioni costose una sola volta per un batch di elementi, includi queste operazioni nel metodo DoFn.Setup o nel metodo DoFn.StartBundle anziché nell'elemento DoFn.ProcessElement. Alcuni esempi includono le seguenti operazioni:

• Analisi di un file di configurazione che controlla alcuni aspetti del comportamento dell'istanza DoFn. Richiama questa azione una sola volta, quando l'istanza di DoFn viene inizializzata, utilizzando il metodo DoFn.Setup.

• Creazione di un client di breve durata riutilizzato in tutti gli elementi di un bundle, ad esempio quando tutti gli elementi del bundle vengono inviati tramite una singola connessione di rete. Richiama questa azione una volta per bundle utilizzando il metodo DoFn.StartBundle.

Limitare le dimensioni dei batch e le chiamate simultanee ai servizi esterni

Quando chiami servizi esterni, puoi ridurre i sovraccarichi per chiamata utilizzando la trasformazione GroupIntoBatches. Questa trasformazione crea batch di elementi di una dimensione specificata. Il batching invia gli elementi a un servizio esterno come un unico payload anziché singolarmente.

In combinazione con il batching, limita il numero massimo di chiamate parallele (simultanee) al servizio esterno scegliendo chiavi appropriate per partizionare i dati in entrata. Il numero di partizioni determina il parallelismo massimo. Ad esempio, se a ogni elemento viene assegnata la stessa chiave, una trasformazione downstream per chiamare il servizio esterno non viene eseguita in parallelo.

Prendi in considerazione uno dei seguenti approcci per generare le chiavi per gli elementi:

• Scegli un attributo del set di dati da utilizzare come chiavi dei dati, ad esempio gli ID utente.
• Genera chiavi dei dati per dividere gli elementi in modo casuale in un numero fisso di partizioni, dove il numero di valori chiave possibili determina il numero di partizioni. Devi creare un numero sufficiente di partizioni per il parallelismo. Ogni partizione deve contenere un numero sufficiente di elementi per rendere utile la trasformazione GroupIntoBatches.

Il seguente esempio di codice Java mostra come dividere in modo casuale gli elementi in dieci partizioni:

// PII or classified data which needs redaction.
PCollection<String> sensitiveData = ...;

int numPartitions = 10; // Number of parallel batches to create.
PCollection<KV<Long, Iterable<String>>> batchedData =
    sensitiveData
        .apply("Assign data into partitions",
            ParDo.of(new DoFn<String, KV<Long, String>>() {
              Random random = new Random();

              @ProcessElement
              public void assignRandomPartition(ProcessContext context) {
                context.output(
                  KV.of(randomPartitionNumber(), context.element()));
              }
              private static int randomPartitionNumber() {
                return random.nextInt(numPartitions);
              }
            }))
        .apply("Create batches of sensitive data",
            GroupIntoBatches.<Long, String>ofSize(100L));

// Use batched sensitive data to fully utilize Redaction API,
// which has a rate limit but allows large payloads.
batchedData
    .apply("Call Redaction API in batches", callRedactionApiOnBatch());

Elabora contemporaneamente le operazioni lente per elemento

Quando il tuo DoFn esegue operazioni lente per elemento (ad esempio chiamando un modello di machine learning esterno o un'API web), l'esecuzione sincrona all'interno di DoFn.ProcessElement blocca i thread di lavoro finché ogni operazione non viene completata. Questo blocco limita la velocità effettiva e può causare timeout dei bundle o tassi di tentativi elevati.

Anche con il partizionamento ottimale delle chiavi, la serializzazione per chiave garantisce che venga elaborato un solo elemento alla volta per chiave. Tuttavia, il wrapping della classe DoFn con AsyncWrapper viola questo vincolo elaborando gli elementi contemporaneamente in un pool di thread in background e restituendo il controllo al runner principale prima del termine dell'elaborazione. L'utilizzo dello stato e dei timer di Beam AsyncWrapper preserva l'elaborazione exactly-once.

Quando progetti una pipeline che utilizza AsyncWrapper, segui queste linee guida:

• Partiziona i dati in entrata con le chiavi: AsyncWrapper richiede l'input di coppie chiave-valore KV<K, InputT>. Per distribuire il lavoro tra più worker e prevenire colli di bottiglia di serializzazione, partiziona il set di dati in entrata in un numero appropriato di chiavi distinte.
• Limita la concorrenza dei worker: utilizza il parametro parallelism per impostare il numero massimo di operazioni simultanee consentite per nodo worker.
• Garantisci la sicurezza dei thread: se imposti parallelism maggiore di 1 e DoFn è stateful, anche DoFn deve essere thread-safe.
• Evita l'aggregazione a livello di bundle: non utilizzare la logica di batch o aggregazione all'interno di StartBundle o FinishBundle, perché AsyncWrapper richiama questi metodi del ciclo di vita per elemento.
• Non utilizzare output taggati: gli output multipli taggati (tag di output aggiuntivi) non sono supportati.
• Esecuzione dell'ottimizzazione (Java): per impostazione predefinita, le attività vengono eseguite sulla JVM condivisa ForkJoinPool. Imposta useThreadPool=true per isolare l'esecuzione in un pool di thread fisso dedicato dimensionato in base al parallelismo configurato. Questo approccio evita la contesa con altri consumer ForkJoinPool nella stessa JVM.
• Esecuzione della messa a punto (Python): nell'SDK Python, tuttavia, AsyncWrapper supporta una modalità use_asyncio=True aggiuntiva che esegue le coroutine all'interno di un ciclo di eventi asyncio anziché di un pool di thread in background.

I seguenti esempi di codice Java mostrano come eseguire il wrapping di un DoFn con AsyncWrapper:

import org.apache.beam.sdk.transforms.AsyncWrapper;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.WithKeys;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollection;
import org.joda.time.Duration;

// Define a synchronous DoFn that makes slow external service calls.
public class CallSlowExternalServiceFn extends DoFn<String, String> {
  private transient ExternalServiceClient externalServiceClient;

  @Setup
  public void setup() {
    externalServiceClient = new ExternalServiceClient();
  }

  @ProcessElement
  public void processElement(@Element String element, OutputReceiver<String> receiver) {
    String result = externalServiceClient.call(element);
    receiver.output(result);
  }
}

// Partition incoming data across distinct keys to distribute worker load.
PCollection<KV<String, String>> keyedData = inputData.apply(
    WithKeys.of(element -> element.substring(0, 1))); // or your actual partition key

// Wrap the DoFn with AsyncWrapper.
PCollection<String> results = keyedData.apply(
    "Process Asynchronously",
    ParDo.of(
        new AsyncWrapper<String, String, String>(
            new CallSlowExternalServiceFn(),
            /* parallelism= */ 20,
            /* timerFrequency= */ Duration.standardSeconds(5),
            /* maxItemsToBuffer= */ 50,
            /* timeout= */ Duration.standardSeconds(1),
            /* maxWaitTime= */ Duration.millis(500),
            /* idFn= */ null, // defaults to element identity for deduplication
            /* useThreadPool= */ true)));

Identificare i problemi di rendimento causati dai passaggi uniti

Dataflow crea un grafico di passaggi che rappresenta la pipeline in base alle trasformazioni e ai dati che hai utilizzato per costruirla. Questo grafico è chiamato grafico di esecuzione della pipeline.

Quando esegui il deployment della pipeline, Dataflow potrebbe modificare il grafico di esecuzione della pipeline per migliorare le prestazioni. Ad esempio, Dataflow potrebbe unire alcune operazioni, un processo noto come ottimizzazione della fusione, per evitare l'impatto su prestazioni e costi della scrittura di ogni oggetto PCollection intermedio nella pipeline.

In alcuni casi, Dataflow potrebbe determinare in modo errato il modo ottimale per unire le operazioni nella pipeline, il che può limitare la capacità del job di utilizzare tutti i worker disponibili. In questi casi, puoi impedire la fusione delle operazioni.

Considera il seguente esempio di codice Apache Beam. Una trasformazione GenerateSequence crea un piccolo oggetto PCollection delimitato, che viene poi ulteriormente elaborato da due trasformazioni ParDo downstream.

La trasformazione Find Primes Less-than-N potrebbe essere costosa dal punto di vista computazionale e probabilmente verrà eseguita lentamente per numeri elevati. Al contrario, la trasformazione Increment Number viene completata rapidamente.

import com.google.common.math.LongMath;
...

public class FusedStepsPipeline {

  final class FindLowerPrimesFn extends DoFn<Long, String> {
    @ProcessElement
    public void processElement(ProcessContext c) {
      Long n = c.element();
      if (n > 1) {
        for (long i = 2; i < n; i++) {
          if (LongMath.isPrime(i)) {
            c.output(Long.toString(i));
          }
        }
      }
    }
  }

  public static void main(String[] args) {
    Pipeline p = Pipeline.create(options);

    PCollection<Long> sequence = p.apply("Generate Sequence",
        GenerateSequence
            .from(0)
            .to(1000000));

    // Pipeline branch 1
    sequence.apply("Find Primes Less-than-N",
        ParDo.of(new FindLowerPrimesFn()));

    // Pipeline branch 2
    sequence.apply("Increment Number",
        MapElements.via(new SimpleFunction<Long, Long>() {
          public Long apply(Long n) {
            return ++n;
          }
        }));

    p.run().waitUntilFinish();
  }
}

Il seguente diagramma mostra una rappresentazione grafica della pipeline nell'interfaccia di monitoraggio di Dataflow.

L'interfaccia di monitoraggio di Dataflow mostra che la stessa velocità di elaborazione lenta si verifica per entrambe le trasformazioni, in particolare 13 elementi al secondo. Potresti aspettarti che la trasformazione Increment Number elabori gli elementi rapidamente, ma sembra invece essere legata alla stessa velocità di elaborazione di Find Primes Less-than-N.

Il motivo è che Dataflow ha unito i passaggi in un'unica fase, impedendone l'esecuzione indipendente. Puoi utilizzare il comando gcloud dataflow jobs describe per trovare ulteriori informazioni:

gcloud dataflow jobs describe --full job-id --format json

Nell'output risultante, i passaggi uniti sono descritti nell'oggetto ExecutionStageSummary nell'array ComponentTransform:

...

    "executionPipelineStage": [
      {
        "componentSource": [
          ...
        ],
        "componentTransform": [
          {
            "name": "s1",
            "originalTransform": "Generate Sequence/Read(BoundedCountingSource)",
            "userName": "Generate Sequence/Read(BoundedCountingSource)"
          },
          {
            "name": "s2",
            "originalTransform": "Find Primes Less-than-N",
            "userName": "Find Primes Less-than-N"
          },
          {
            "name": "s3",
            "originalTransform": "Increment Number/Map",
            "userName": "Increment Number/Map"
          }
        ],
        "id": "S01",
        "kind": "PAR_DO_KIND",
        "name": "F0"
      }

...

In questo scenario, poiché la trasformazione Find Primes Less-than-N è il passaggio lento, interrompere la fusione prima di questo passaggio è una strategia appropriata. Un metodo per separare i passaggi consiste nell'inserire una trasformazione GroupByKey e raggruppare prima del passaggio, come mostrato nel seguente esempio di codice Java.

sequence
    .apply("Map Elements", MapElements.via(new SimpleFunction<Long, KV<Long, Void>>() {
      public KV<Long, Void> apply(Long n) {
        return KV.of(n, null);
      }
    }))
    .apply("Group By Key", GroupByKey.<Long, Void>create())
    .apply("Emit Keys", Keys.<Long>create())
    .apply("Find Primes Less-than-N", ParDo.of(new FindLowerPrimesFn()));

Puoi anche combinare questi passaggi di separazione in una trasformazione composita riutilizzabile.

Dopo aver separato i passaggi, quando esegui la pipeline, Increment Number viene completato in pochi secondi e la trasformazione Find Primes Less-than-N, che richiede molto più tempo, viene eseguita in una fase separata.

Questo esempio applica un'operazione di raggruppamento e separazione per separare i passaggi. Puoi utilizzare altri approcci per altre circostanze. In questo caso, la gestione dell'output duplicato non è un problema, dato l'output consecutivo della trasformazione GenerateSequence. Gli oggetti KV con chiavi duplicate vengono deduplicati in una singola chiave nella trasformazione di raggruppamento (GroupByKey) e nella trasformazione di separazione (Keys). Per conservare i duplicati dopo le operazioni di raggruppamento e separazione, crea coppie chiave-valore seguendo questi passaggi:

1. Utilizza una chiave casuale e l'input originale come valore.
2. Raggruppa utilizzando la chiave casuale.
3. Emetti i valori per ogni chiave come output.

Puoi anche utilizzare una trasformazione Reshuffle per impedire la fusione delle trasformazioni circostanti. Tuttavia, gli effetti collaterali della trasformazione Reshuffle non sono trasferibili tra diversi runner Apache Beam.

Per ulteriori informazioni sull'ottimizzazione del parallelismo e della fusione, consulta Ciclo di vita della pipeline.

Utilizzare le metriche Apache Beam per raccogliere informazioni sulle pipeline

Le metriche Apache Beam sono una classe di utilità che produce metriche per la generazione di report sulle proprietà di una pipeline in esecuzione. Quando utilizzi Cloud Monitoring, le metriche di Apache Beam sono disponibili come metriche personalizzate di Cloud Monitoring.

L'esempio seguente mostra le metriche Counter di Apache Beam utilizzate in una sottoclasse DoFn.

Il codice di esempio utilizza due contatori. Un contatore tiene traccia degli errori di analisi JSON (malformedCounter), mentre l'altro contatore tiene traccia se il messaggio JSON è valido, ma contiene un payload vuoto (emptyCounter). In Cloud Monitoring, i nomi delle metrica personalizzata sono custom.googleapis.com/dataflow/malformedJson e custom.googleapis.com/dataflow/emptyPayload. Puoi utilizzare le metriche personalizzate per creare visualizzazioni e criteri di avviso in Cloud Monitoring.

final TupleTag<String> errorTag = new TupleTag<String>(){};
final TupleTag<MockObject> successTag = new TupleTag<MockObject>(){};

final class ParseEventFn extends DoFn<String, MyObject> {

  private final Counter malformedCounter = Metrics.counter(ParseEventFn.class, "malformedJson");
  private final Counter emptyCounter = Metrics.counter(ParseEventFn.class, "emptyPayload");
  private Gson gsonParser;

  @Setup
  public setup() {
    gsonParser = new Gson();
  }

  @ProcessElement
  public void processElement(ProcessContext c) {
    try {
      MyObject myObj = gsonParser.fromJson(c.element(), MyObject.class);
      if (myObj.getPayload() != null) {
        //  Output the element if non-empty payload
        c.output(successTag, myObj);
      }
      else {
        // Increment empty payload counter
        emptyCounter.inc();
      }
    }
    catch (JsonParseException e) {
      // Increment malformed JSON counter
      malformedCounter.inc();
      // Output the element to dead-letter queue
      c.output(errorTag, c.element());
    }
  }
}

Proteggere i bucket Cloud Storage dagli attacchi alle pipeline

Per proteggere i bucket Cloud Storage da un attacco della pipeline Dataflow, devi capire come può verificarsi una compromissione.

Un malintenzionato raramente prende di mira direttamente Cloud Storage. Sfruttano invece una vulnerabilità chiamata avvelenamento delle risorse shadow. In questo scenario, un malintenzionato compromette il bucket Cloud Storage contenente i modelli Dataflow, i metadati o le funzioni definite dall'utente (UDF) Python personalizzate per inserire codice dannoso. Quando Dataflow esegue lo scale automatico e avvia una nuova VM worker, estrae il codice compromesso da Cloud Storage, lo esegue, ruba il token dell'account di servizio del worker ed estrae i dati dai bucket di origine.

Per contribuire a bloccare questo punto di ingresso dell'attacco, devi proteggere i bucket Cloud Storage che alimentano Dataflow, proteggere le pipeline stesse e limitare rigorosamente le autorizzazioni sui bucket Cloud Storage.

Proteggere i bucket Cloud Storage

Per contribuire a proteggere il punto di iniezione, proteggi i bucket che utilizzi per i modelli Dataflow, i file di staging e le dipendenze del codice. Implementando controlli rigorosi dell'accesso in scrittura su questi bucket, impedisci l'inserimento di codice non autorizzato nell'ambiente di lavoro.

• Isola i bucket:separa i bucket di dati dai bucket di codice operativo. Dataflow utilizza i flag --stagingLocation e --tempLocation. Conserva questi dati in un bucket dedicato e con accesso estremamente limitato, completamente separato dai dati di input o output non elaborati.
• Applica l'accesso uniforme a livello di bucket (UBLA): abilita UBLA su tutti i bucket correlati a Dataflow. In questo modo vengono disattivati gli elenchi di controllo dell'accesso dell'accesso (ACL) legacy e soggetti a errori e viene garantito che solo le policy IAM (Identity and Access Management) centralizzate determinino chi può accedere ai file.
• Blocca le autorizzazioni di scrittura:concedi al account di servizio CI/CD roles/storage.objectAdmin (scrittura/eliminazione) nei bucket di staging e dei modelli. Concedi agli sviluppatori e ad altri service account al massimo roles/storage.objectViewer (sola lettura).
• Abilita il controllo delle versioni degli oggetti e la conservazione degli oggetti:attiva il controllo delle versioni degli oggetti nei bucket dei modelli. Se un malintenzionato o un account compromesso sovrascrive un modello, puoi ricevere immediatamente un avviso relativo alla nuova versione e annullare le modifiche. In combinazione con un criterio di conservazione degli oggetti, questo aiuta a impedire agli utenti malintenzionati di eliminare le dipendenze di staging.

Rafforzare l'ambiente di esecuzione Dataflow

Per contribuire a mantenere un ambiente di esecuzione sicuro e proteggere dal furto di dati, limita la comunicazione dei worker.

• Blocca il traffico internet in uscita (networking privato rigoroso): quando avvii Dataflow, passa --usePublicIps=false. Inoltre, configura le regole firewall VPC per bloccare tutto il traffico in uscita verso internet pubblico dalle subnet Dataflow. Se un worker viene compromesso, non può aprire una reverse shell o esfiltrare dati a un indirizzo IP esterno controllato da un attaccante.
• Applica i Controlli di servizio VPC (VPC-SC): posiziona i bucket di dati Cloud Storage e il progetto Dataflow all'interno di un perimetro di sicurezza VPC-SC rigoroso. VPC-SC funge da firewall di rete che esegue l'override delle autorizzazioni IAM. Anche se un malintenzionato ruba un token di account di servizio worker valido, VPC-SC blocca qualsiasi richiesta di copia dei dati dal bucket Cloud Storage a un bucket al di fuori del perimetro dell'organizzazione.

Implementare il principio del privilegio minimo rigoroso nei bucket di dati

Se un worker Dataflow viene compromesso, la sua area interessata è definita dalle autorizzazioni IAM del suoaccount di serviziot.

• Evita di utilizzare il account di servizio Compute Engine predefinito:questo account ha per impostazione predefinita ampie autorizzazioni Editor sull'intero progetto. Utilizza un account di servizio worker dedicato e gestito dall'utente utilizzando il flag --serviceAccount.
• Ruoli IAM specifici per il bucket:non concedere al account di servizio worker Dataflow ruoli di archiviazione a livello di progetto, come roles/storage.admin. Concedi invece i diritti direttamente sui bucket specifici:
  • Bucket di input: concedi il ruolo roles/storage.objectViewer (sola lettura).
  • Bucket di output:concedi il ruolo roles/storage.objectCreator (sola scrittura, ovvero può creare oggetti, ma non può leggere o sovrascrivere i dati esistenti).
  • Bucket di staging/temporaneo: concedi i ruoli roles/storage.objectAdmin (lettura, scrittura ed eliminazione).

Proteggere i container personalizzati e la catena di fornitura del software

Se utilizzi immagini Docker personalizzate per le dipendenze Dataflow utilizzando --sdkContainerImage, un malintenzionato potrebbe tentare di compromettere il registro delle immagini anziché un bucket Cloud Storage.

• Esegui la scansione di Artifact Registry:archivia le immagini dei worker personalizzati in Artifact Registry e attiva la scansione delle vulnerabilità.

• Utilizza digest immagine immutabili:anziché fare riferimento a un'immagine tramite un tag modificabile come us-docker.pkg.dev/my-project/dataflow-worker:latest, specifica il digest SHA crittografico esatto e immutabile nelle opzioni della pipeline:

  --sdkContainerImage=us-docker.pkg.dev/my-project/dataflow-worker@sha256:7b9c...`
  

  In questo modo, il codice esatto che hai scansionato e approvato in CI/CD è quello che viene eseguito sui worker, bloccando completamente gli attacchi di sostituzione delle immagini.

Utilizzare la registrazione per rilevare le anomalie

Le configurazioni sicure contribuiscono a bloccare l'attacco, ma devi anche configurare gli avvisi di Cloud Logging per i seguenti trigger:

1. Eventi nei bucket Cloud Storage di staging o nel modello Dataflow che hanno origine da un'identità non CI/CD, ad esempio richieste storage.objects.update o storage.objects.create.
2. Volumi elevati di richieste storage.objects.get da indirizzi IP imprevisti o picchi insoliti nel volume di dati che si spostano tramite Dataflow, che potrebbero indicare un tentativo di esfiltrazione attivo.

Esegui la migrazione di una pipeline tra progetti Google Cloud

I job Dataflow sono associati al progetto Google Cloud in cui li hai creati. Non puoi spostare un job direttamente in un altro progetto. Per spostare una pipeline in un altro progetto, arresta il job esistente e ricrealo nel nuovo progetto. Per una guida dettagliata su questo processo, consulta Eseguire la migrazione dei job della pipeline a un altro progetto. Google Cloud

Scopri di più

Le pagine seguenti forniscono ulteriori informazioni su come strutturare la pipeline, su quali trasformazioni applicare ai dati e su cosa considerare quando scegli i metodi di input e output della pipeline.

• Sviluppa e testa le pipeline Dataflow.
• Progetta la pipeline.
• Crea la pipeline.

Per saperne di più sulla creazione del codice utente, consulta i requisiti per le funzioni fornite dall'utente.