Leggi da Cloud Storage a Dataflow

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

Includi la dipendenza dalla 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://".

Java 

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

Python 

apache-beam[gcp]==VERSION

Vai 

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

Per ulteriori informazioni, consulta Installa l'SDK Apache Beam.

Abilita 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 RPC (chiamata di procedura remota) open source 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 l'utilizzo di gRPC.

Riga di comando

  1. Assicurati di utilizzare la versione 2.55.0 o successiva dell' SDK Apache Beam.
  2. Per eseguire un job Dataflow, utilizza l'opzione di pipeline --additional-experiments=use_grpc_for_gcs. Per informazioni sulle diverse opzioni di pipeline, consulta Flag facoltativi.

SDK Apache Beam

  1. Assicurati di utilizzare la versione 2.55.0 o successiva dell' SDK Apache Beam.
  2. Per eseguire un job Dataflow, utilizza l'opzione --experiments=use_grpc_for_gcs pipeline. Per informazioni sulle diverse opzioni di pipeline, consulta Opzioni di base.

Puoi configurare il connettore I/O di Apache Beam su Dataflow per generare metriche relative a gRPC in Cloud Monitoring. Le metriche relative 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.

Per informazioni su come configurare il connettore I/O di Apache Beam su Dataflow per generare metriche relative a gRPC, consulta Utilizzare le metriche lato client. Se la raccolta delle metriche non è necessaria per il tuo caso d'uso, puoi scegliere di disattivarla. Per istruzioni, vedi Disattivare le metriche lato client.

Parallelismo

I connettori TextIO e AvroIO supportano due livelli di parallelismo:

  • I singoli file vengono codificati separatamente, in modo che più utenti possano leggerli.
  • Se i file non sono compressi, il connettore può leggere i sottointervalli di ciascun file distintamente, il che comporta 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 sul rendimento per la lettura da Cloud Storage. I carichi di lavoro sono stati eseguiti su un e2-standard2 worker utilizzando l'SDK Apache Beam 2.49.0 per Java. Non hanno utilizzato Runner v2.

100 milioni di record | 1 kB | 1 colonna Velocità effettiva (byte) Velocità effettiva (elementi)
Leggi 320 MB/s 320.000 elementi al secondo

Queste metriche si basano su semplici pipeline batch. Sono progettati per confrontare il rendimento tra i connettori I/O e non sono necessariamente rappresentativi delle pipeline reali. Le prestazioni della pipeline Dataflow sono complesse e dipendono dal tipo di VM, dai dati in fase di elaborazione, dalle prestazioni di origini e destinazioni esterne e dal codice utente. Le metriche si basano sull'esecuzione dell'SDK Java e non sono rappresentative delle caratteristiche di prestazioni di altri SDK per lingua. Per ulteriori informazioni, consulta Rendimento IO di Beam.

Best practice

  • Evita di utilizzare watchForNewFiles con Cloud Storage. Questo approccio non è scalabile per pipeline di produzione di grandi dimensioni, perché il connettore deve mantenere in memoria un elenco dei file visualizzati. L'elenco non può essere svuotato dalla memoria, il che riduce la memoria di lavoro dei worker nel tempo. Valuta la possibilità di utilizzare le notifiche Pub/Sub per Cloud Storage. Per ulteriori informazioni, consulta Modelli di elaborazione dei file.

  • Se sia il nome file sia 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 ulteriori informazioni, consulta la sezione Accedere ai nomi file. La documentazione di FileIO mostra anche un esempio di questo pattern.

Esempio

L'esempio riportato di seguito mostra come leggere da Cloud Storage.

Java

Per autenticarti a Dataflow, configura le Credenziali predefinite dell'applicazione. Per ulteriori informazioni, consulta Configurare l'autenticazione per un ambiente di sviluppo locale. 

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