Developing Data Processing Pipeline With Apache Beam

We will create BigQuery dataset and table with the appropriate schema as a data sink where our output from the dataflow job will reside in. The Dataset region will be your nearest location. It is Asia-south1 (Mumbai) in our case. You need to provide the output schema (already given in batch.py) while creating the table in BigQuery.

Next open cloud shell editor and set your project property if it is not already set and will clone the GitHub repository which has all supported files and data.

git clone https://github.com/aniket-g/batch-pipeline-using-apache-beam-python

Once it is done, change into the directory where all files reside.

Now copy the beer.csv file into our bucket using the command given below.

gsutil cp beers.csv gs://ag-pipeline/batch/

Alternatively, you can upload that CSV file by going to the Storage Bucket.

To run the pipeline, you need to have Apache Beam library installed on Virtual Machine.

sudo pip3 install apache_beam[gcp]

That’s all.

Now we will walk through the pipeline code to know how it works. Mostly we will look at the Ptransforms in the pipeline.

def discard_incomplete(data):
    """Filters out records that don't have an information."""
    return len(data['abv']) > 0 and len(data['id']) > 0 and len(data['name']) > 0 and len(data['style']) > 0

We have filtered out the data which does not have information or null values in it.

def convert_types(data):
    """Converts string values to their appropriate type."""
    data['abv'] = float(data['abv']) if 'abv' in data else None
    data['id'] = int(data['id']) if 'id' in data else None
    data['name'] = str(data['name']) if 'name' in data else None
    data['style'] = str(data['style']) if 'style' in data else None
    data['ounces'] = float(data['ounces']) if 'ounces' in data else None
    return data

The above function will convert the string values to their appropriate data type.

def del_unwanted_cols(data):
    """Deleting unwanted columns"""
    del data['ibu']
    del data['brewery_id']
    return data

In the above function, we deleted unwanted columns which ended up in cleaned data.

p = beam.Pipeline(options=PipelineOptions())(p | 'ReadData' >> beam.io.ReadFromText('gs://purchases-3/beers.csv', skip_header_lines =1)
       | 'Split' >> beam.Map(lambda x: x.split(','))
       | 'format to dict' >> beam.Map(lambda x: {"sr": x[0], "abv": x[1], "id": x[2], "name": x[3], "style": x[4], "ounces": x[5]}) 
       | 'DelIncompleteData' >> beam.Filter(discard_incomplete)
       | 'Convertypes' >> beam.Map(convert_types)
       | 'DelUnwantedData' >> beam.Map(del_unwanted_cols)
       | 'WriteToBigQuery' >> beam.io.WriteToBigQuery(
           '{0}:beer.beer_data'.format(PROJECT_ID),
           schema=SCHEMA,
           write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND))
    result = p.run()

beam.io.ReadFromText — reads the data from external sources into the PCollection

beam.map — works like ParDo, applied Map in multiple ways to transform every element in PCollection. The Map accepts a function that returns a single element for every input element in the PCollection.

beam.Filter — accepts a function that keeps elements that return True, and filters out the remaining elements.

beam.io.WriteToBigQuery — Write transform to a BigQuerySink accepts PCollections of dictionaries. It requires the following arguments

• TableReference can be a PROJECT:DATASET.TABLE or DATASET.TABLE string.
• TableSchema can be a NAME:TYPE{,NAME:TYPE}* string (e.g. ‘month:STRING,event_count:INTEGER’).

Now we run pipeline using dataflow runner using the following syntax.

python3 batch.py --runner DataFlowRunner --project aniket-g --temp_location gs://ag-pipeline/batch/temp --staging_location gs://ag-pipeline/batch/stag --region asia-east1 --job_name drinkbeer

Currently, Dataflow provides regional endpoints for some regions which do not include Asia-south1 hence I chose Asia-east1 in Region.

• project — The ID of your Google Cloud project.
• runner — The pipeline runner that will parse your program and construct your pipeline. It can be direct runner also if you want to debug your pipeline. Here we are using Dataflow runner.
• staging_location — A Cloud Storage path for Dataflow to stage code packages needed by workers executing the job.
• temp_location — A Cloud Storage path for Dataflow to stage temporary job files created during the execution of the pipeline.
• region — You can specify region where you want to run your dataflow runner.
• job_name (Optional) — Give any name to the dataflow pipeline.

Now go to Dataflow, you can see your job is running of batch type.