Scrivere da Dataflow a Cloud Storage

Questo documento descrive come scrivere dati di testo da Dataflow a Cloud Storage utilizzando il TextIO connettore I/O di Apache Beam.

Includi la dipendenza della Google Cloud libreria

Per utilizzare il connettore TextIO con Cloud Storage, includi la seguente dipendenza. Questa libreria fornisce un gestore di schemi per i nomi 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 saperne di più, consulta Installare 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 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 scrittura del job Dataflow in Cloud Storage, puoi abilitare il connettore I/O Apache Beam su Dataflow per utilizzare gRPC.

Riga di comando

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

SDK Apache Beam

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

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:

  • Monitorare e ottimizzare le prestazioni delle richieste gRPC a Cloud Storage.
  • Risolvere i problemi ed eseguire il debug.
  • Ottenere informazioni dettagliate sull'utilizzo e sul comportamento della tua applicazione.

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

Parallelismo

Il parallelismo è determinato principalmente dal numero di shard. Per impostazione predefinita, il runner imposta automaticamente questo valore. Per la maggior parte delle pipeline, è consigliabile utilizzare il comportamento predefinito. In questo documento, consulta Best practice.

Prestazioni

La tabella seguente mostra le metriche delle prestazioni per la scrittura in 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 il runner portatile.

100 M record | 1 kB | 1 colonna Velocità effettiva (byte) Velocità effettiva (elementi)
Scrivi 130 MBps 130.000 elementi al secondo

Queste metriche si basano su semplici pipeline batch. Sono progettate per confrontare le prestazioni tra i connettori I/O e non sono necessariamente rappresentative delle pipeline reali. Le prestazioni della pipeline Dataflow sono complesse e dipendono dal tipo di VM, dai dati elaborati, dalle prestazioni 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 delle prestazioni di altri SDK di linguaggio. Per saperne di più, consulta Prestazioni I/O di Beam.

Best practice

  • In generale, evita di impostare un numero specifico di shard. In questo modo, il runner può selezionare un valore appropriato per la tua scalabilità. Per abilitare la suddivisione automatica, chiama .withAutoSharding(), non .withNumShards(0). Se regoli il numero di shard, ti consigliamo di scrivere tra 100 MB e 1 GB per shard. Tuttavia, il valore ottimale potrebbe dipendere dal carico di lavoro.

  • Cloud Storage può scalare fino a un numero molto elevato di richieste al secondo. Tuttavia, se la pipeline presenta picchi elevati nel volume di scrittura, valuta la possibilità di scrivere in più bucket per evitare di sovraccaricare temporaneamente un singolo bucket Cloud Storage.

  • In generale, la scrittura in Cloud Storage è più efficiente quando ogni scrittura è più grande (1 kB o superiore). La scrittura di piccoli record in un numero elevato di file può comportare prestazioni peggiori per byte.

  • Quando generi i nomi dei file, valuta la possibilità di utilizzare nomi di file non sequenziali per distribuire il carico. Per saperne di più, consulta Utilizzare una convenzione di denominazione che distribuisce il carico in modo uniforme tra gli intervalli di chiavi.

  • Quando assegni un nome ai file, non utilizzare il simbolo "@" seguito da un numero o un asterisco ("*"). Per saperne di più, consulta "@*" e "@N" sono specifiche di suddivisione riservate.

Esempio: scrivere file di testo in Cloud Storage

L'esempio seguente crea una pipeline batch che scrive file di testo utilizzando la compressione GZIP:

Java

Per eseguire l'autenticazione in Dataflow, configura le credenziali predefinite dell'applicazione. Per saperne di più, consulta Configurare l'autenticazione per un ambiente di sviluppo locale. 

import java.util.Arrays;
import java.util.List;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.Compression;
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.Create;

public class BatchWriteStorage {
  public interface Options extends PipelineOptions {
    @Description("The Cloud Storage bucket to write to")
    String getBucketName();

    void setBucketName(String value);
  }

  // Write text data to Cloud Storage
  public static void main(String[] args) {
    final List<String> wordsList = Arrays.asList("1", "2", "3", "4");

    var options = PipelineOptionsFactory.fromArgs(args).withValidation().as(Options.class);
    var pipeline = Pipeline.create(options);
    pipeline
        .apply(Create
            .of(wordsList))
        .apply(TextIO
            .write()
            .to(options.getBucketName())
            .withSuffix(".txt")
            .withCompression(Compression.GZIP)
        );
    pipeline.run().waitUntilFinish();
  }
}

Se l'input PCollection non è limitato, devi definire una finestra o un trigger nella raccolta, quindi specificare le scritture in finestra chiamando TextIO.Write.withWindowedWrites.

Python

Per eseguire l'autenticazione in Dataflow, configura le credenziali predefinite dell'applicazione. Per saperne di più, consulta Configurare l'autenticazione per un ambiente di sviluppo locale. 

import argparse
from typing import List

import apache_beam as beam
from apache_beam.io.textio import WriteToText
from apache_beam.options.pipeline_options import PipelineOptions

from typing_extensions import Self


def write_to_cloud_storage(argv: List[str] = None) -> None:
    # Parse the pipeline options passed into the application.
    class MyOptions(PipelineOptions):
        @classmethod
        # Define a custom pipeline option that specfies the Cloud Storage bucket.
        def _add_argparse_args(cls: Self, parser: argparse.ArgumentParser) -> None:
            parser.add_argument("--output", required=True)

    wordsList = ["1", "2", "3", "4"]
    options = MyOptions()

    with beam.Pipeline(options=options.view_as(PipelineOptions)) as pipeline:
        (
            pipeline
            | "Create elements" >> beam.Create(wordsList)
            | "Write Files" >> WriteToText(options.output, file_name_suffix=".txt")
        )

Per il percorso di output, specifica un percorso Cloud Storage che includa il nome del bucket e un prefisso del nome file. Ad esempio, se specifichi gs://my_bucket/output/file, il connettore TextIO scrive nel bucket Cloud Storage denominato my_bucket, e i file di output hanno il prefisso output/file*.

Per impostazione predefinita, il connettore TextIO suddivide i file di output utilizzando una convenzione di denominazione simile a questa: <file-prefix>-00000-of-00001. Se vuoi, puoi specificare un suffisso del nome file e uno schema di compressione, come mostrato nell'esempio.

Per garantire scritture idempotenti, Dataflow scrive in un file temporaneo e poi copia il file temporaneo completato nel file finale. Per controllare la posizione in cui vengono archiviati questi file temporanei, utilizza il withTempDirectory metodo.

Passaggi successivi