This document describes how to write text data from Dataflow toCloud Storage by using the Apache BeamTextIOI/O connector.

Include the Google Cloud Platform library dependency

To use theTextIO connector with Cloud Storage, include the followingdependency. This library provides a schema handler for"gs://" filenames.

<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"

For more information, seeInstall the Apache Beam SDK.

Enable gRPC on Apache Beam I/O connector on Dataflow

You canconnect to Cloud Storage using gRPC through theApache Beam I/O connector on Dataflow.gRPC is ahigh performance open-source remote procedure call (RPC) framework developed byGoogle that you can use to interact withCloud Storage.

To speed up your Dataflow job's write requests to Cloud Storage, youcan enable the Apache Beam I/O connector on Dataflow to use gRPC.

You can configure Apache Beam I/O connector on Dataflow to generate gRPC related metrics in Cloud Monitoring. The gRPC related metrics can help you to do the following:

• Monitor and optimize the performance of gRPC requests to Cloud Storage.
• Troubleshoot and debug issues.
• Gain insights into your application's usage and behavior.

Parallelism

Parallelism is determined primarily by the number of shards. By default, therunner automatically sets this value. For most pipelines, using the defaultbehavior is recommended. In this document, seeBest practices.

Performance

The following table shows performance metrics for writing toCloud Storage. The workloads were run on onee2-standard2 worker,using the Apache Beam SDK 2.49.0 for Java. They did not use Runner v2.

These metrics are based on simple batch pipelines. They are intended to compare performance between I/O connectors, and are not necessarily representative of real-world pipelines. Dataflow pipeline performance is complex, and is a function of VM type, the data being processed, the performance of external sources and sinks, and user code. Metrics are based on running the Java SDK, and aren't representative of the performance characteristics of other language SDKs. For more information, seeBeam IO Performance.

Best practices

• In general, avoid setting a specific number of shards. This allows the runnerto select an appropriate value for your scale. To enable autosharding, call.withAutoSharding(), not.withNumShards(0). If you tune the number ofshards, we recommend writing between 100MB and 1GB per shard. However, theoptimum value might depend on the workload.

• Cloud Storage can scale to a very large number of requests persecond. However, if your pipeline has large spikes in write volume, considerwriting to multiple buckets, to avoid temporarily overloading any singleCloud Storage bucket.

• In general, writing to Cloud Storage is more efficient when eachwrite is larger (1 kb or greater). Writing small records to a large number offiles can result in worse performance per byte.

• When generating file names, consider using non-sequential file names, in orderto distribute load. For more information, seeUse a naming convention that distributes load evenly across key ranges.

• When naming files, don't use the at sign ('@') followed by a number or anasterisk ('*'). For more information, see"@*" and "@N" are reserved sharding specs.

Example: Write text files to Cloud Storage

The following example creates a batch pipeline that writes text files using GZIPcompression:

importjava.util.Arrays;importjava.util.List;importorg.apache.beam.sdk.Pipeline;importorg.apache.beam.sdk.io.Compression;importorg.apache.beam.sdk.io.TextIO;importorg.apache.beam.sdk.options.Description;importorg.apache.beam.sdk.options.PipelineOptions;importorg.apache.beam.sdk.options.PipelineOptionsFactory;importorg.apache.beam.sdk.transforms.Create;publicclassBatchWriteStorage{publicinterfaceOptionsextendsPipelineOptions{@Description("The Cloud Storage bucket to write to")StringgetBucketName();voidsetBucketName(Stringvalue);}// Write text data to Cloud Storagepublicstaticvoidmain(String[]args){finalList<String>wordsList=Arrays.asList("1","2","3","4");varoptions=PipelineOptionsFactory.fromArgs(args).withValidation().as(Options.class);varpipeline=Pipeline.create(options);pipeline.apply(Create.of(wordsList)).apply(TextIO.write().to(options.getBucketName()).withSuffix(".txt").withCompression(Compression.GZIP));pipeline.run().waitUntilFinish();}}

importargparsefromtypingimportListimportapache_beamasbeamfromapache_beam.io.textioimportWriteToTextfromapache_beam.options.pipeline_optionsimportPipelineOptionsfromtyping_extensionsimportSelfdefwrite_to_cloud_storage(argv:List[str]=None)->None:# Parse the pipeline options passed into the application.classMyOptions(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()withbeam.Pipeline(options=options.view_as(PipelineOptions))aspipeline:(pipeline|"Create elements" >>beam.Create(wordsList)|"Write Files" >>WriteToText(options.output,file_name_suffix=".txt"))

For the output path, specify a Cloud Storage path that includes thebucket name and a filename prefix. For example, if you specifygs://my_bucket/output/file, theTextIO connector writes to theCloud Storage bucket namedmy_bucket, and the output files have the prefixoutput/file*.

By default, theTextIO connector shards the output files, using a namingconvention like this:<file-prefix>-00000-of-00001. Optionally, you canspecify a filename suffix and a compression scheme, as shown in the example.

To ensure idempotent writes, Dataflow writes to a temporary fileand then copies the completed temporary file to the final file.To control where these temporary files are stored,use thewithTempDirectorymethod.

What's next

• Read theTextIO API documentation.
• See the list ofGoogle-provided templates.