Leggi da Cloud Storage a Dataflow

Per leggere i dati da Cloud Storage a Dataflow, utilizza il connettore di I/O TextIO o AvroIO Apache Beam.

Includi la dipendenza della libreria Google Cloud

Per utilizzare il connettore TextIO o AvroIO con Cloud Storage, includi la seguente dipendenza. Questa libreria fornisce un gestore dello schema per i nomi dei file "gs://".

<dependency>
  <groupId>org.apache.beam</groupId>
  <artifactId>beam-sdks-java-io-google-cloud-platform</artifactId>
  <version>${beam.version}</version>
</dependency>

apache-beam[gcp]==VERSION

import _ "github.com/apache/beam/sdks/v2/go/pkg/beam/io/filesystem/gcs"

Per saperne di più, consulta Installa l'SDK Apache Beam.

Abilitare gRPC sul connettore I/O Apache Beam su Dataflow

Puoi connetterti a Cloud Storage utilizzando gRPC tramite il connettore I/O Apache Beam su Dataflow. gRPC è un framework open source per chiamata di procedura remota (RPC) ad alte prestazioni sviluppato da Google che puoi utilizzare per interagire con Cloud Storage.

Per velocizzare le richieste di lettura del job Dataflow a Cloud Storage, puoi attivare il connettore I/O Apache Beam su Dataflow per utilizzare gRPC.

Puoi configurare il connettore I/O Apache Beam su Dataflow per generare metriche correlate a gRPC in Cloud Monitoring. Le metriche correlate a gRPC possono aiutarti a:

• Monitora e ottimizza le prestazioni delle richieste gRPC a Cloud Storage.
• Risolvi i problemi ed esegui il debug.
• Ottieni informazioni sull'utilizzo e sul comportamento della tua applicazione.

Parallelismo

I connettori TextIO e AvroIO supportano due livelli di parallelismo:

• I singoli file vengono codificati separatamente, in modo che possano essere letti da più worker.
• Se i file non sono compressi, il connettore può leggere separatamente i sottointervalli di ogni file, il che porta a un livello di parallelismo molto elevato. Questa suddivisione è possibile solo se ogni riga del file è un record significativo. Ad esempio, non è disponibile per impostazione predefinita per i file JSON.

Prestazioni

La tabella seguente mostra le metriche di rendimento per la lettura da Cloud Storage. I carichi di lavoro sono stati eseguiti su un worker e2-standard2, utilizzando l'SDK Apache Beam 2.49.0 per Java. Non hanno utilizzato Runner v2.

Queste metriche si basano su semplici pipeline batch. Sono pensati per confrontare le prestazioni tra i connettori I/O e non sono necessariamente rappresentativi delle pipeline reali. Il rendimento della pipeline Dataflow è complesso e dipende dal tipo di VM, dai dati in fase di elaborazione, dal rendimento delle origini e dei sink esterni e dal codice utente. Le metriche si basano sull'esecuzione dell'SDK Java e non sono rappresentative delle caratteristiche di prestazioni di altri SDK di linguaggio. Per saperne di più, consulta Prestazioni di Beam IO.

Best practice

• Evita di utilizzare watchForNewFiles con Cloud Storage. Questo approccio non è scalabile per pipeline di produzione di grandi dimensioni, perché il connettore deve conservare un elenco dei file visualizzati in memoria. L'elenco non può essere svuotato dalla memoria, il che riduce la memoria di lavoro dei worker nel tempo. Ti consigliamo di utilizzare le notifiche Pub/Sub per Cloud Storage. Per saperne di più, consulta Pattern di elaborazione dei file.

• Se sia il nome file che i contenuti del file sono dati utili, utilizza la classe FileIO per leggere i nomi file. Ad esempio, un nome file potrebbe contenere metadati utili per l'elaborazione dei dati nel file. Per saperne di più, consulta la sezione Accesso ai nomi dei file. La documentazione di FileIO mostra anche un esempio di questo pattern.

Esempio

L'esempio seguente mostra come leggere da Cloud Storage.

import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.options.Description;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.values.TypeDescriptors;

public class ReadFromStorage {
  public static Pipeline createPipeline(Options options) {
    var pipeline = Pipeline.create(options);
    pipeline
        // Read from a text file.
        .apply(TextIO.read().from(
            "gs://" + options.getBucket() + "/*.txt"))
        .apply(
            MapElements.into(TypeDescriptors.strings())
                .via(
                    (x -> {
                      System.out.println(x);
                      return x;
                    })));
    return pipeline;
  }
}

Passaggi successivi

• Leggi la documentazione dell'API TextIO.
• Consulta l'elenco dei modelli forniti da Google.