Java

Construct aPipelineOptions object using the methodPipelineOptionsFactory.fromArgs.

For an example, see theLaunch on Dataflow samplesection on this page.

Python

Create aPipelineOptions object.

For an example, see theLaunch on Dataflow samplesection on this page.

Go

Setting pipeline options programmatically usingPipelineOptions is notsupported in the Apache Beam SDK for Go. Use Go command-line arguments.

For an example, see theLaunch on Dataflow samplesection on this page.

Java

The following example syntax is from theWordCount pipeline in theJava tutorial.

mvn-Pdataflow-runnercompileexec:java\-Dexec.mainClass=org.apache.beam.examples.WordCount\-Dexec.args="--project=PROJECT_ID \  --gcpTempLocation=gs://BUCKET_NAME/temp/ \  --output=gs://BUCKET_NAME/output \  --runner=DataflowRunner \  --region=REGION"

Replace the following:

• PROJECT_ID: your Google Cloud project ID
• BUCKET_NAME: the name of your Cloud Storage bucket
• REGION: aDataflow region,us-central1

Python

The following example syntax is from theWordCount pipeline in thePython tutorial.

python-mapache_beam.examples.wordcount\--regionDATAFLOW_REGION\--inputgs://dataflow-samples/shakespeare/kinglear.txt\--outputgs://STORAGE_BUCKET/results/outputs\--runnerDataflowRunner\--projectPROJECT_ID\--temp_locationgs://STORAGE_BUCKET/tmp/

Replace the following:

• DATAFLOW_REGION: theregionwhere you want to deploy the Dataflow job—for example,europe-west1

  The--region flag overrides the default region that isset in the metadata server, your local client, or environmentvariables.

• STORAGE_BUCKET: theCloud Storage bucket name

• PROJECT_ID: the Google Cloud project ID

Go

The following example syntax is from theWordCount pipeline in theGo tutorial.

gorunwordcount.go--inputgs://dataflow-samples/shakespeare/kinglear.txt\--outputgs://BUCKET_NAME/results/outputs\--runnerdataflow\--projectPROJECT_ID\--regionDATAFLOW_REGION\--staging_locationgs://BUCKET_NAME/binaries/

Replace the following:

• BUCKET_NAME: the Cloud Storage bucket name

• PROJECT_ID: the Google Cloud project ID

• DATAFLOW_REGION: Theregion where youwant to deploy the Dataflow job. For example,europe-west1.The--region flag overrides the default region that is set in the metadataserver, your local client, or environment variables.

Java

In yourPipelineOptions object, include theexperiments option by using the following syntax.This example sets the boot disk size to 80 GB with the experiment flag.

options.setExperiments("streaming_boot_disk_size_gb=80")

For an example that shows how to create thePipelineOptions object, see theLaunch on Dataflow samplesection on this page.

Python

In yourPipelineOptions object, include theexperiments option by using the following syntax.This example sets the boot disk size to 80 GB with the experiment flag.

beam_options=PipelineOptions(beam_args,experiments=['streaming_boot_disk_size_gb=80'])

For an example that shows how to create thePipelineOptions object, see theLaunch on Dataflow samplesection on this page.

Go

Setting pipeline options programmatically usingPipelineOptions is notsupported in the Apache Beam SDK for Go. Use Go command-line arguments.

Java

This example sets the boot disk size to 80 GB with the experiment flag.

--experiments=streaming_boot_disk_size_gb=80

Python

This example sets the boot disk size to 80 GB with the experiment flag.

--experiments=streaming_boot_disk_size_gb=80

Go

This example sets the boot disk size to 80 GB with the experiment flag.

--experiments=streaming_boot_disk_size_gb=80

Classic template

gclouddataflowjobsrunJOB_NAME--additional-experiments=EXPERIMENT[,...]

Flex template

gclouddataflowflex-templaterunJOB_NAME--additional-experiments=EXPERIMENT[,...]

Java

AccessPipelineOptions inside anyParDo transform'sDoFn instance by usingthe methodProcessContext.getPipelineOptions.

Python

This feature is not supported in the Apache Beam SDK for Python.

Go

Access pipeline options usingbeam.PipelineOptions.

Java

• project: the ID of your Google Cloud project.
• runner: the pipeline runner that executes your pipeline. ForGoogle Cloud execution, this must beDataflowRunner.

• gcpTempLocation: a Cloud Storage path forDataflow to stage most temporary files. The specified bucketmust already exist.

  If you don't specifygcpTempLocation, then Dataflow usesthe value of thetempLocation option. If you don't specify either ofthese options, then Dataflow creates a newCloud Storage bucket.

Python

• project: your Google Cloud project ID.
• region: the region for your Dataflow job.
• runner: the pipeline runner that executes your pipeline. ForGoogle Cloud execution, this must beDataflowRunner.
• temp_location: a Cloud Storage path forDataflow to stage temporary job files created during theexecution of the pipeline.

Go

• project: your Google Cloud project ID.
• region: the region for your Dataflow job.
• runner: the pipeline runner that executes your pipeline. ForGoogle Cloud execution, this must bedataflow.
• staging_location: a Cloud Storage path forDataflow to stage temporary job files created during theexecution of the pipeline.

Java

// Create and set your PipelineOptions.DataflowPipelineOptionsoptions=PipelineOptionsFactory.as(DataflowPipelineOptions.class);// For cloud execution, set the Google Cloud project, staging location,// and set DataflowRunner.options.setProject("my-project-id");options.setStagingLocation("gs://my-bucket/binaries");options.setRunner(DataflowRunner.class);// Create the Pipeline with the specified options.Pipelinep=Pipeline.create(options);

Python

importargparseimportapache_beamasbeamfromapache_beam.options.pipeline_optionsimportPipelineOptionsparser=argparse.ArgumentParser()# parser.add_argument('--my-arg', help='description')args,beam_args=parser.parse_known_args()# Create and set your PipelineOptions.# For Cloud execution, specify DataflowRunner and set the Cloud Platform# project, job name, temporary files location, and region.# For more information about regions, check:# https://cloud.google.com/dataflow/docs/concepts/regional-endpointsbeam_options=PipelineOptions(beam_args,runner='DataflowRunner',project='my-project-id',job_name='unique-job-name',temp_location='gs://my-bucket/temp',region='us-central1')# Note: Repeatable options like dataflow_service_options or experiments must# be specified as a list of string(s).# e.g. dataflow_service_options=['enable_prime']# Create the Pipeline with the specified options.withbeam.Pipeline(options=beam_options)aspipeline:pass# build your pipeline here.

Go

The Apache Beam SDK for Go uses Go command-line arguments. Useflag.Set() to set flag values.

// Use the Go flag package to parse custom options.flag.Parse()// Set the required options programmatically.// For Cloud execution, specify the Dataflow runner, Google Cloud// project ID, region, and staging location.// For more information about regions, see// https://cloud.google.com/dataflow/docs/concepts/regional-endpointsflag.Set("runner","dataflow")flag.Set("project","my-project-id")flag.Set("region","us-central1")flag.Set("staging_location","gs://my-bucket/binaries")beam.Init()// Create the Pipeline.p:=beam.NewPipeline()s:=p.Root()

Java

// Set your PipelineOptions to the specified command-line optionsMyOptionsoptions=PipelineOptionsFactory.fromArgs(args).withValidation();// Create the Pipeline with the specified options.Pipelinep=Pipeline.create(options);

Python

Use thePython argparse module to parse command-line options.

# Use Python argparse module to parse custom argumentsimportargparseimportapache_beamasbeamfromapache_beam.options.pipeline_optionsimportPipelineOptions# For more details on how to use argparse, take a look at:#   https://docs.python.org/3/library/argparse.htmlparser=argparse.ArgumentParser()parser.add_argument('--input-file',default='gs://dataflow-samples/shakespeare/kinglear.txt',help='The file path for the input text to process.')parser.add_argument('--output-path',required=True,help='The path prefix for output files.')args,beam_args=parser.parse_known_args()# Create the Pipeline with remaining arguments.beam_options=PipelineOptions(beam_args)withbeam.Pipeline(options=beam_options)aspipeline:lines=(pipeline|'Read files' >>beam.io.ReadFromText(args.input_file)|'Write files' >>beam.io.WriteToText(args.output_path))

Go

Use theGoflag package to parsecommand-line options. You must parse the options before you callbeam.Init(). In this example,output is a command-line option.

// Define configuration optionsvar(output=flag.String("output","","Output file (required)."))// Parse options before beam.Init()flag.Parse()beam.Init()// Input validation must be done after beam.Init()if*output==""{log.Fatal("No output provided!")}p:=beam.NewPipeline()

Java

When an Apache Beam Java program runs a pipeline on a service such asDataflow, it is typically executed asynchronously. To run apipeline and wait until the job completes, setDataflowRunner as thepipeline runner and explicitly callpipeline.run().waitUntilFinish().

When you useDataflowRunner and callwaitUntilFinish() on thePipelineResult object returned frompipeline.run(), the pipeline executeson Google Cloud but the local code waits for the cloud job to finish andreturn the finalDataflowPipelineJob object. While the job runs, theDataflow service prints job status updates and console messageswhile it waits.

Python

When an Apache Beam Python program runs a pipeline on a service such asDataflow, it is typically executed asynchronously. To blockuntil pipeline completion, use thewait_until_finish() method of thePipelineResult object, returned from therun() method of the runner.

Go

When an Apache Beam Go program runs a pipeline on Dataflow,it is synchronous by default and blocks until pipeline completion. If youdon't want to block, there are two options:

1. Start the job in a Go routine.

   gofunc(){pr,err:=beamx.Run(ctx,p)iferr!=nil{// Handle the error.}// Send beam.PipelineResult into a channel.results<-pr}()// Do other operations while the pipeline runs.

2. Use the--async command-line flag, which is in thejobopts package.

Java

If your pipeline reads from an unbounded data source, such asPub/Sub, the pipeline automatically executes in streaming mode.

Python

If your pipeline uses an unbounded data source, such as Pub/Sub, youmust set thestreaming option to true.

Go

If your pipeline reads from an unbounded data source, such asPub/Sub, the pipeline automatically executes in streaming mode.

Java

// Create and set our Pipeline Options.PipelineOptionsoptions=PipelineOptionsFactory.create();// Create the Pipeline with the specified options.Pipelinep=Pipeline.create(options);

Python

importargparseimportapache_beamasbeamfromapache_beam.options.pipeline_optionsimportPipelineOptionsparser=argparse.ArgumentParser()# parser.add_argument('--my-arg')args,beam_args=parser.parse_known_args()# Create and set your Pipeline Options.beam_options=PipelineOptions(beam_args)args=beam_options.view_as(MyOptions)withbeam.Pipeline(options=beam_options)aspipeline:lines=(pipeline|beam.io.ReadFromText(args.input)|beam.io.WriteToText(args.output))

Go

// Parse options before beam.Init()flag.Parse()beam.Init()p:=beam.NewPipeline()

Java

To add your own options, define an interface with getter and setter methodsfor each option, as in the following example:

publicinterfaceMyOptionsextendsPipelineOptions{StringgetMyCustomOption();voidsetMyCustomOption(StringmyCustomOption);}

Python

To add your own options, use theadd_argument() method (which behavesexactly like Python's standardargparse module),as in the following example:

fromapache_beam.options.pipeline_optionsimportPipelineOptionsclassMyOptions(PipelineOptions):@classmethoddef_add_argparse_args(cls,parser):parser.add_argument('--input')parser.add_argument('--output')

Go

To add your own options, use theGo flag package as shown in thefollowing example:

var(input=flag.String("input","","")output=flag.String("output","",""))

Java

You set the description and default value using annotations, as follows:

publicinterfaceMyOptionsextendsPipelineOptions{@Description("My custom command line argument.")@Default.String("DEFAULT")StringgetMyCustomOption();voidsetMyCustomOption(StringmyCustomOption);}

We recommend that you register your interface withPipelineOptionsFactoryand then pass the interface when creating thePipelineOptions object. Whenyou register your interface withPipelineOptionsFactory, the--help canfind your custom options interface and add it to the output of the--helpcommand.PipelineOptionsFactory validates that your custom options arecompatible with all other registered options.

The following example code shows how to register your custom options interfacewithPipelineOptionsFactory:

PipelineOptionsFactory.register(MyOptions.class);MyOptionsoptions=PipelineOptionsFactory.fromArgs(args).withValidation().as(MyOptions.class);

Now your pipeline can accept--myCustomOption=value as a command-lineargument.

Python

You set the description and default value as follows:

fromapache_beam.options.pipeline_optionsimportPipelineOptionsclassMyOptions(PipelineOptions):@classmethoddef_add_argparse_args(cls,parser):parser.add_argument('--input',default='gs://dataflow-samples/shakespeare/kinglear.txt',help='The file path for the input text to process.')parser.add_argument('--output',required=True,help='The path prefix for output files.')

Go

You set the description and default value as follows:

var(input=flag.String("input","gs://MY_STORAGE_BUCKET/input","Input for the pipeline")output=flag.String("output","gs://MY_STORAGE_BUCKET/output","Output for the pipeline"))