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 Apache Beam.

Includi la dipendenza della libreria Google Cloud

Per utilizzare il connettore TextIO 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 scrittura del job Dataflow in 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

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 la sezione Best practice.

Prestazioni

La tabella seguente mostra le metriche di rendimento 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 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

• In generale, evita di impostare un numero specifico di shard. In questo modo, il runner può selezionare un valore appropriato per la tua scala. Per attivare l'autosharding, 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 più). 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 non sequenziali, in modo da distribuire il carico. Per saperne di più, consulta la sezione Utilizzare una convenzione di denominazione che distribuisca 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 da un asterisco (*). Per saperne di più, consulta la sezione "@*" e "@N" sono specifiche di partizionamento 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:

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();
  }
}

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 in shard, utilizzando una convenzione di denominazione come 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 dove vengono archiviati questi file temporanei, utilizza il metodo withTempDirectory.

Passaggi successivi

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