Building a Simple ETL Pipeline with Python and Google Cloud Platform

This article is based on a tutorial originally published on Towards Data Science. The explanation and code walkthrough below are an independent technical summary of the concepts and steps described there.

What You Will Build

By the end of this guide, you will have a working serverless ETL pipeline that:

• Connects to an FTP server and retrieves the most recent CSV file (Extract)
• Cleans and modifies the CSV content (Transform)
• Loads the result into a Google BigQuery table (Load)
• Runs automatically on a schedule via Google Cloud Scheduler

What Is ETL?

Extract, Transform, Load (ETL) is the standard pattern for moving data from one or more source systems into a destination — typically a data warehouse — where it is stored in a format suited for analysis. The three stages are:

• Extract: Read raw data from the source (a database, flat file, API, FTP server, etc.).
• Transform: Clean, normalise, filter, or enrich the data to match the target schema or business rules.
• Load: Write the processed data into the destination system (e.g., BigQuery, Redshift, Snowflake).

Many commercial ETL tools exist, but they can be overkill when you simply need to move a file from point A to point B on a schedule. A lightweight Python script running on a serverless platform is often a faster, cheaper alternative.

Why Google Cloud Functions?

Google Cloud Functions (GCF) is Google Cloud’s serverless compute platform. You deploy a function — a single unit of logic — and GCF handles provisioning, scaling, and teardown automatically. You pay only for actual execution time. Comparable offerings from other cloud providers include AWS Lambda and Azure Functions.

For an ETL workload that runs once per day and processes a single CSV file, Cloud Functions is a cost-effective choice: no virtual machines to maintain and no containers to manage.

Real-World Scenario

A logistics company’s infrastructure team exports one CSV file of daily transaction data from their internal database and uploads it to an FTP server operated by a third-party service provider. The analytics team needs that data in their BigQuery data warehouse so it can be joined with other sources and surfaced to stakeholders in dashboards.

The ETL pipeline described in this article was built to solve exactly this problem:

• Triggered daily by Cloud Scheduler via an HTTP request to a Cloud Function
• The function logs into the FTP server, identifies the latest file, and downloads it
• The CSV is cleaned in memory
• The cleaned data is streamed into a BigQuery table

Step 1 — Set Up a Google Cloud Function

Follow these steps in the Cloud Functions Overview page in the Google Cloud Console:

1. Make sure the correct GCP project is selected.
2. Click Create Function and give your function a descriptive name (e.g., ftp-to-bigquery).
3. Set the Trigger to HTTP.
4. Under Source code, select Inline editor (you will paste your code here).
5. Set the Runtime to Python 3.7 (or a later supported version).
6. Under Advanced options, choose the Region closest to your users or data sources. Note that Cloud Functions is not available in every Google Cloud region — check the supported regions list before selecting.
7. Click Create.

Step 2 — Declare Your Dependencies

A Cloud Function project contains two files: main.py (your logic) and requirements.txt (your Python dependencies). Click the requirements.txt tab in the inline editor and add the following libraries:

google-cloud-bigquery
python-csv
requests
wget
pytest-shutil

Here is what each dependency does:

• google-cloud-bigquery — The official Google client library for interacting with BigQuery (creating tables, streaming rows, running queries).
• python-csv — Provides additional CSV manipulation utilities on top of Python’s built-in csv module.
• requests — A widely used HTTP library; used here to construct the FTP URL request.
• wget — Simplifies downloading files from a URL directly to the Cloud Function’s temporary filesystem (/tmp).
• pytest-shutil — Provides filesystem and SSH utilities used when working with remote file servers.

Step 3 — Extract: Connect to FTP and Download the Latest File

In the main.py tab, define your entry-point function. Cloud Functions calls this function when the HTTP trigger fires. The extract stage logs into the FTP server, lists the directory contents to identify the most recently uploaded CSV, and downloads it to the function’s temporary storage.

import ftplib
import os
import wget

def ftp_function(request):
    """
    Entry point for the Cloud Function.
    Called via HTTP trigger (e.g., from Cloud Scheduler).
    """
    ftp_host = os.environ.get('FTP_HOST')
    ftp_user = os.environ.get('FTP_USER')
    ftp_pass = os.environ.get('FTP_PASS')
    remote_dir = '/path/to/csv/files'

    # Connect and log in
    ftp = ftplib.FTP(ftp_host)
    ftp.login(user=ftp_user, passwd=ftp_pass)
    ftp.cwd(remote_dir)

    # Get the most recent file in the directory
    entries = list(ftp.mlsd())
    entries.sort(key=lambda e: e[1].get('modify', ''), reverse=True)
    latest_file = entries[0][0]  # filename of the most recently modified entry

    # Download the file to /tmp (Cloud Functions writable directory)
    ftp_url = f'ftp://{ftp_user}:{ftp_pass}@{ftp_host}{remote_dir}/{latest_file}'
    local_path = f'/tmp/{latest_file}'
    wget.download(ftp_url, local_path)

    ftp.quit()
    return f'Downloaded: {latest_file}'
Code language: PHP (php)

A few important implementation notes:

• Store credentials (FTP host, username, password, BigQuery dataset IDs) in Cloud Function environment variables rather than hard-coding them in source. You set these in the GCP Console under the function’s configuration.
• Cloud Functions provides a writable temporary directory at /tmp with a maximum size of 512 MB. Files written there are not persisted between invocations.
• The mlsd() method returns directory entries with metadata including modification timestamps, making it straightforward to identify the newest file. If the FTP server does not support mlsd(), fall back to parsing LIST or NLST output.

Step 4 — Transform: Clean the CSV Data

In this scenario, the transform step makes a specific business rule change: wherever the source data contains the vendor code FBA, it should be replaced with the display name AMAZON. This is a simple find-and-replace on CSV content, but the same pattern scales to more complex normalisation logic.

import csv

def transform_csv(input_path, output_path):
    """
    Read the downloaded CSV, apply transformations, and write a cleaned version.
    """
    with open(input_path, 'r', newline='', encoding='utf-8') as infile, \
         open(output_path, 'w', newline='', encoding='utf-8') as outfile:

        reader = csv.reader(infile)
        writer = csv.writer(outfile)

        for row in reader:
            # Replace every occurrence of 'FBA' with 'AMAZON' across all columns
            cleaned_row = ['AMAZON' if cell.strip() == 'FBA' else cell for cell in row]
            writer.writerow(cleaned_row)

    return output_path
Code language: PHP (php)

This function reads the raw file row by row, applies the substitution, and writes a new cleaned file to /tmp. Processing row by row keeps memory usage low even for large files.

Step 5 — Load: Insert Data into BigQuery

The final stage reads the cleaned CSV and loads it into a BigQuery table using the google-cloud-bigquery client library. The simplest approach for a daily batch is to use the BigQuery load job, which reads a file from Cloud Storage or a local path and appends (or overwrites) a table.

from google.cloud import bigquery
import os

def load_to_bigquery(cleaned_path):
    """
    Load the cleaned CSV into a BigQuery table.
    """
    client = bigquery.Client()

    dataset_id = os.environ.get('BQ_DATASET')
    table_id = os.environ.get('BQ_TABLE')
    table_ref = client.dataset(dataset_id).table(table_id)

    job_config = bigquery.LoadJobConfig(
        source_format=bigquery.SourceFormat.CSV,
        skip_leading_rows=1,   # skip the header row
        autodetect=True,       # infer schema from data
        write_disposition=bigquery.WriteDisposition.WRITE_APPEND,
    )

    with open(cleaned_path, 'rb') as source_file:
        load_job = client.load_table_from_file(
            source_file,
            table_ref,
            job_config=job_config
        )

    load_job.result()  # wait for the job to complete
    print(f'Loaded {load_job.output_rows} rows into {dataset_id}.{table_id}')
Code language: PHP (php)

Key configuration choices:

• autodetect=True lets BigQuery infer column names and types from the CSV header and first rows. For production use, define an explicit schema to avoid type inference surprises.
• WRITE_APPEND adds new rows to the existing table each day. Use WRITE_TRUNCATE if you want to replace the table contents on each run.
• The Cloud Function’s service account must have the BigQuery Data Editor and BigQuery Job User IAM roles on the target project.

Step 6 — Putting It All Together in main.py

Wire the three stages into a single entry-point function that Cloud Functions will invoke:

def ftp_function(request):
    raw_path = '/tmp/raw_data.csv'
    cleaned_path = '/tmp/cleaned_data.csv'

    # 1. Extract
    download_latest_ftp_file(raw_path)   # calls the FTP logic shown above

    # 2. Transform
    transform_csv(raw_path, cleaned_path)

    # 3. Load
    load_to_bigquery(cleaned_path)

    return 'ETL pipeline completed successfully.', 200
Code language: PHP (php)

Step 7 — Automate the Pipeline with Cloud Scheduler

To run the pipeline daily without manual intervention, use Google Cloud Scheduler to send an HTTP POST request to your Cloud Function’s trigger URL on a cron schedule:

1. Open Cloud Scheduler in the GCP Console and click Create Job.
2. Set the Frequency using a cron expression, e.g., 0 6 * * * to run at 06:00 UTC every day.
3. Set the Target to HTTP and paste in the trigger URL of your Cloud Function.
4. Set the HTTP method to POST.
5. Save and enable the job.

Cloud Scheduler will now call the function automatically each day, triggering the full Extract → Transform → Load sequence without any manual steps.

Architecture Summary

The complete pipeline looks like this:

• Cloud Scheduler fires an HTTP request on a daily cron schedule
• Cloud Function receives the request and runs ftp_function
• The function connects to the FTP Server, downloads the latest CSV to /tmp
• The CSV is transformed in memory and written to a cleaned file
• The cleaned file is loaded into BigQuery via a load job
• Stakeholders query BigQuery directly or via a BI tool (e.g., Looker, Data Studio)

Production Considerations

Before running this pattern in production, consider the following:

• Error handling: Wrap each stage in try/except blocks and return meaningful HTTP status codes. Log errors to Cloud Logging so failures are visible.
• Idempotency: If the scheduler retries a failed run, you may load duplicate rows. Track which files have been processed (e.g., in Cloud Firestore or a BigQuery audit table) and skip already-loaded files.
• Credentials security: Use Secret Manager to store FTP credentials instead of plain environment variables for stronger security controls and audit logging.
• Schema management: Replace autodetect=True with an explicit BigQuery schema definition to prevent silent type changes when the upstream CSV format changes.
• Timeout limits: Cloud Functions has a maximum execution timeout (up to 9 minutes for first-generation functions, up to 60 minutes for second-generation). For large files, consider Cloud Run instead.
• FTP over SFTP: Plain FTP transmits credentials in clear text. If possible, negotiate with the data provider to switch to SFTP (SSH File Transfer Protocol), and use paramiko or pysftp in place of ftplib.

Conclusion

This walkthrough demonstrates how a handful of Python standard library modules, combined with the Google Cloud client library for BigQuery and the serverless execution environment of Cloud Functions, is enough to build a reliable daily ETL pipeline — no dedicated ETL tool required. The same pattern extends naturally to other sources (REST APIs, Cloud Storage, databases) and more complex transformation logic as your data needs grow.

For the original full tutorial with additional context and screenshots, see the article on Towards Data Science.