Data pipeline patterns in GCP - Commands & Configuration

Introduction

Data pipelines move and transform data from one place to another automatically. They help organize data flow so you can analyze or use it easily without manual work.

When you want to collect data from multiple sources like databases and files into one place.

When you need to clean or change data format before using it in reports or apps.

When you want to move data regularly, like every hour or day, without doing it by hand.

When you want to combine data from different systems to get a full picture.

When you want to automate data tasks so your team can focus on analysis, not moving data.

Config File - dataflow_pipeline.py

dataflow_pipeline.py

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions

class FormatData(beam.DoFn):
    def process(self, element):
        # Example: convert all text to lowercase
        yield element.lower()

pipeline_options = PipelineOptions(
    project='example-project',
    region='us-central1',
    runner='DataflowRunner',
    temp_location='gs://example-bucket/temp',
    staging_location='gs://example-bucket/staging'
)

with beam.Pipeline(options=pipeline_options) as p:
    (p
     | 'ReadFromText' >> beam.io.ReadFromText('gs://example-bucket/input/data.txt')
     | 'FormatData' >> beam.ParDo(FormatData())
     | 'WriteToText' >> beam.io.WriteToText('gs://example-bucket/output/formatted_data')
    )

This Python file defines a simple data pipeline using Apache Beam on Google Cloud Dataflow.

FormatData class changes data to lowercase as an example transformation.

pipeline_options sets project info, region, and storage locations needed to run the pipeline on Dataflow.

The pipeline reads text data from a Cloud Storage bucket, processes it, and writes the output back to another bucket.

Commands

This command runs the data pipeline script. It starts the pipeline on Google Cloud Dataflow, which reads, processes, and writes data automatically.

Terminal

python3 dataflow_pipeline.py

Expected Output

INFO:apache_beam.runners.dataflow.dataflow_runner:Job 2024-0000000000000000 starting on project example-project

INFO:apache_beam.runners.dataflow.dataflow_runner:Job 2024-0000000000000000 is in state JOB_STATE_RUNNING

INFO:apache_beam.runners.dataflow.dataflow_runner:Job 2024-0000000000000000 finished successfully

This command lists the files in the output Cloud Storage bucket to verify the pipeline wrote the processed data.

Terminal

gsutil ls gs://example-bucket/output/

Expected Output

gs://example-bucket/output/formatted_data-00000-of-00001

This command shows the content of the output file to check the data was transformed as expected.

Terminal

gsutil cat gs://example-bucket/output/formatted_data-00000-of-00001

Expected Output

example line one

another example line

more data here

Key Concept

If you remember nothing else from this pattern, remember: data pipelines automate moving and changing data so you don’t have to do it manually.

Common Mistakes

Not setting the correct project or bucket names in the pipeline options

The pipeline will fail to start or cannot read/write data if these values are wrong.

Always double-check your Google Cloud project ID and Cloud Storage bucket names before running the pipeline.

Running the pipeline without having the input data file in the specified bucket

The pipeline will error because it cannot find the data to process.

Upload your input data file to the Cloud Storage bucket path before running the pipeline.

Not waiting for the pipeline job to finish before checking output files

Output files may not exist yet or be incomplete if the job is still running.

Check the job status and wait until it finishes successfully before verifying output.

Summary

Write a pipeline script that reads, processes, and writes data using Apache Beam on Google Cloud Dataflow.

Run the pipeline script to start the automated data processing job in the cloud.

Verify the output files in Cloud Storage to confirm the pipeline worked as expected.