Create a Dataflow pipeline using Go

Set up your development environment

The Apache Beam SDK is an open source programming model for data pipelines. You define a pipeline with an Apache Beam program and then choose a runner, such as Dataflow, to run your pipeline.

We recommend that you use the latest version of Go when working with the Apache Beam SDK for Go. If you don't have the latest version of Go installed, use Go's Download and install guide to download and install Go for your specific operating system.

To verify the version of Go that you have installed, run the following command in your local terminal:

 go  
version 

Run the Beam wordcount example

The Apache Beam SDK for Go includes a wordcount pipeline example . The wordcount example does the following:

1. Reads a text file as input. By default, it reads a text file located in a Cloud Storage bucket with the resource name gs://dataflow-samples/shakespeare/kinglear.txt .
2. Parses each line into words.
3. Performs a frequency count on the tokenized words.

To run the latest version of the Beam wordcount example on your local machine, perform the following steps:

1. Use the git clone command to clone the apache/beam GitHub repository:

    git  
   clone  
   https://github.com/apache/beam.git 
   

2. Switch to the beam/sdks/go directory:

     cd 
     
   beam/sdks/go 
   

3. Use the following command to run the pipeline:

    go  
   run  
   examples/wordcount/wordcount.go  
    \ 
     
   --input  
   gs://dataflow-samples/shakespeare/kinglear.txt  
    \ 
     
   --output  
   outputs 
   

   The input flag specifies the file to read, and the output flag specifies the filename for the frequency count output.

After the pipeline completes, view the output results:

 more  
outputs* 

To exit, press q .

Modify the pipeline code

The Beam wordcount pipeline distinguishes between uppercase and lowercase words. The following steps show how to create your own Go module, modify the wordcount pipeline so that the pipeline is not case-sensitive, and run it on Dataflow.

Create a Go module

To make changes to the pipeline code, follow these steps.

1. Create a directory for your Go module in a location of your choice:

    mkdir  
   wordcount 
     cd 
     
   wordcount 
   

2. Create a Go module. For this example, use example/dataflow as the module path.

    go  
   mod  
   init  
   example/dataflow 
   

3. Download the latest copy of the wordcount code from the Apache Beam GitHub repository. Put this file into the wordcount directory you created.

4. If you are using a non-Linux operating system, you must get the Go unix package . This package is required to run pipelines on the Dataflow service.

    go  
   get  
   -u  
   golang.org/x/sys/unix 
   

5. Ensure that the go.mod file matches the module's source code:

    go  
   mod  
   tidy 
   

Run the unmodified pipeline

Verify the unmodified wordcount pipeline runs locally.

1. From the terminal, build and run the pipeline locally:

      
   go  
   run  
   wordcount.go  
   --input  
   gs://dataflow-samples/shakespeare/kinglear.txt  
    \ 
     
   --output  
   outputs 
   

2. View the output results:

      
   more  
   outputs* 
   

3. To exit, press q .

Change the pipeline code

To change the pipeline so that it is not case-sensitive, modify the code to apply the strings.ToLower function to all words.

1. In an editor of your choice, open the wordcount.go file.

2. Examine the init block (comments have been removed for clarity):

      
    func 
     
    init 
    () 
     
    { 
     
    register 
    . 
    DoFn3x0 
    [ 
    context 
    . 
    Context 
    , 
     
    string 
    , 
     
    func 
    ( 
    string 
    )]( 
   & extractFn 
    {}) 
     
    register 
    . 
    Function2x1 
    ( 
    formatFn 
    ) 
     
    register 
    . 
    Emitter1 
    [ 
    string 
    ]() 
     
    } 
    
   

3. Add a new line to register the strings.ToLower function:

      
    func 
     
    init 
    () 
     
    { 
     
    register 
    . 
    DoFn3x0 
    [ 
    context 
    . 
    Context 
    , 
     
    string 
    , 
     
    func 
    ( 
    string 
    )]( 
   & extractFn 
    {}) 
     
    register 
    . 
    Function2x1 
    ( 
    formatFn 
    ) 
     
    register 
    . 
    Emitter1 
    [ 
    string 
    ]() 
     
     register 
    . 
    Function1x1 
    ( 
    strings 
    . 
    ToLower 
    ) 
     
    } 
    
   

4. Examine the CountWords function:

      
    func 
     
    CountWords 
    ( 
    s 
     
    beam 
    . 
    Scope 
    , 
     
    lines 
     
    beam 
    . 
    PCollection 
    ) 
     
    beam 
    . 
    PCollection 
     
    { 
     
    s 
     
    = 
     
    s 
    . 
    Scope 
    ( 
    "CountWords" 
    ) 
     
    // Convert lines of text into individual words. 
     
    col 
     
    := 
     
    beam 
    . 
    ParDo 
    ( 
    s 
    , 
     
   & extractFn 
    { 
    SmallWordLength 
    : 
     
    * 
    smallWordLength 
    }, 
     
    lines 
    ) 
     
    // Count the number of times each word occurs. 
     
    return 
     
    stats 
    . 
    Count 
    ( 
    s 
    , 
     
    col 
    ) 
     
    } 
    
   

5. To lowercase the words, add a ParDo that applies strings.ToLower to every word:

      
    func 
     
    CountWords 
    ( 
    s 
     
    beam 
    . 
    Scope 
    , 
     
    lines 
     
    beam 
    . 
    PCollection 
    ) 
     
    beam 
    . 
    PCollection 
     
    { 
     
    s 
     
    = 
     
    s 
    . 
    Scope 
    ( 
    "CountWords" 
    ) 
     
    // Convert lines of text into individual words. 
     
    col 
     
    := 
     
    beam 
    . 
    ParDo 
    ( 
    s 
    , 
     
   & extractFn 
    { 
    SmallWordLength 
    : 
     
    * 
    smallWordLength 
    }, 
     
    lines 
    ) 
     
     // Map all letters to lowercase. 
     
    lowercaseWords 
     
    := 
     
    beam 
    . 
    ParDo 
    ( 
    s 
    , 
     
    strings 
    . 
    ToLower 
    , 
     
    col 
    ) 
     
    // Count the number of times each word occurs. 
     
    return 
     
    stats 
    . 
    Count 
    ( 
    s 
    , 
     
     lowercaseWords 
    ) 
     
    } 
    
   

6. Save the file.

Run the updated pipeline locally

Run your updated wordcount pipeline locally and verify the output has changed.

1. Build and run the modified wordcount pipeline:

      
   go  
   run  
   wordcount.go  
   --input  
   gs://dataflow-samples/shakespeare/kinglear.txt  
    \ 
     
   --output  
   outputs 
   

2. View the output results of the modified pipeline. All words should be lowercase.

      
   more  
   outputs* 
   

3. To exit, press q .

Run the pipeline on the Dataflow service

To run the updated wordcount example on the Dataflow service, use the following command:

 go  
run  
wordcount.go  
--input  
gs://dataflow-samples/shakespeare/kinglear.txt  
 \ 
  
--output  
gs:// BUCKET_NAME 
/results/outputs  
 \ 
  
--runner  
dataflow  
 \ 
  
--project  
 PROJECT_ID 
  
 \ 
  
--region  
 DATAFLOW_REGION 
  
 \ 
  
--staging_location  
gs:// BUCKET_NAME 
/binaries/ 

Replace the following:

• BUCKET_NAME : the Cloud Storage bucket name.

• PROJECT_ID : the Google Cloud project ID.

• DATAFLOW_REGION : the region where you want to deploy the Dataflow job. For example, europe-west1 . For a list of available locations, see Dataflow locations . The --region flag overrides the default region that is set in the metadata server, your local client, or environment variables.

View your results

You can see a list of your Dataflow jobs in Google Cloud console. In the Google Cloud console, go to the Dataflow Jobspage.

Go to Jobs

The Jobspage displays details of your wordcount job, including a status of Runningat first, and then Succeeded.

When you run a pipeline using Dataflow, your results are stored in a Cloud Storage bucket. View the output results by using either the Google Cloud console or the local terminal.