Write from Dataflow to Cloud Storage

This document describes how to write text data from Dataflow to Cloud Storage by using the Apache Beam TextIO I/O connector .

Include the Google Cloud library dependency

To use the TextIO connector with Cloud Storage, include the following dependency. 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, see Install the Apache Beam SDK .

Enable gRPC on Apache Beam I/O connector on Dataflow

You can connect to Cloud Storage using gRPC through the Apache Beam I/O connector on Dataflow. gRPC is a high performance open-source remote procedure call (RPC) framework developed by Google that you can use to interact with Cloud Storage.

To speed up your Dataflow job's write requests to Cloud Storage, you can 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, the runner automatically sets this value. For most pipelines, using the default behavior is recommended. In this document, see Best practices .

Performance

The following table shows performance metrics for writing to Cloud Storage. The workloads were run on one e2-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, see Beam IO Performance .

Best practices

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

• Cloud Storage can scale to a very large number of requests per second. However, if your pipeline has large spikes in write volume, consider writing to multiple buckets, to avoid temporarily overloading any single Cloud Storage bucket.

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

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

• When naming files, don't use the at sign ('@') followed by a number or an asterisk ('*'). 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 GZIP compression:

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