Implementation:DataTalksClub Data engineering zoomcamp GCS Parquet Upload

Knowledge Sources	DataTalksClub_Data_engineering_zoomcamp
Domains	Data_Ingestion, GCS
Last Updated	2026-02-09 00:00 GMT

Overview

A Python script that downloads NYC yellow taxi Parquet files in parallel using ThreadPoolExecutor, creates or validates a GCS bucket, and uploads the files to GCS with retry logic and verification.

Description

This script provides a robust, production-oriented approach to loading NYC yellow taxi trip data into Google Cloud Storage. It defines four key functions:

• download_file(month): Downloads a Parquet file for a given month from the NYC TLC CloudFront CDN using urllib.request.urlretrieve. Returns the local file path on success or None on failure.
• create_bucket(bucket_name): Validates that the target GCS bucket exists and belongs to the current project. If the bucket does not exist, it creates it. If the bucket exists but is not accessible (owned by another project), the script exits with an error.
• verify_gcs_upload(blob_name): Checks whether a blob exists in the GCS bucket after upload, providing upload verification.
• upload_to_gcs(file_path, max_retries=3): Uploads a local file to GCS with retry logic. Uses a configurable chunk size of 8 MB (CHUNK_SIZE = 8 * 1024 * 1024). On each attempt, it uploads the file and verifies the upload. If verification fails, it retries up to max_retries times with a 5-second delay between attempts.

The main execution block uses ThreadPoolExecutor with 4 workers for both downloading and uploading, enabling parallel processing of the 6 months of data (January through June 2024). The script authenticates via a service account JSON credentials file (gcs.json), with a commented-out alternative for SDK-based authentication.

Usage

Use this implementation when you need to reliably upload Parquet files to GCS with verification and retry logic. It is particularly suited for batch loading of NYC taxi trip data into a data lake. The parallel execution pattern makes it efficient for processing multiple files. Prerequisites include a GCS service account credentials file and the google-cloud-storage Python package.

Code Reference

Source Location

• Repository: DataTalksClub_Data_engineering_zoomcamp
• File: cohorts/2025/03-data-warehouse/load_yellow_taxi_data.py
• Lines: 1-113

Signature

def download_file(month):
    ...

def create_bucket(bucket_name):
    ...

def verify_gcs_upload(blob_name):
    ...

def upload_to_gcs(file_path, max_retries=3):
    ...

Import

import os
import sys
import urllib.request
from concurrent.futures import ThreadPoolExecutor
from google.cloud import storage
from google.api_core.exceptions import NotFound, Forbidden
import time

I/O Contract

Inputs

download_file:

Name	Type	Required	Description
month	str	Yes	Two-digit month string (e.g., '01', '06') identifying which month of data to download

create_bucket:

Name	Type	Required	Description
bucket_name	str	Yes	Name of the GCS bucket to create or validate

verify_gcs_upload:

Name	Type	Required	Description
blob_name	str	Yes	Name of the blob to verify in the GCS bucket

upload_to_gcs:

Name	Type	Required	Description
file_path	str	Yes	Local path to the file to upload
max_retries	int	No	Maximum number of upload attempts (defaults to 3)

Configuration Constants:

Name	Type	Required	Description
BUCKET_NAME	str	Yes	Target GCS bucket name (hardcoded as dezoomcamp_hw3_2025)
CREDENTIALS_FILE	str	Yes	Path to the GCS service account JSON credentials file (gcs.json)
BASE_URL	str	Yes	Base URL for downloading Parquet files from the NYC TLC CDN
MONTHS	list	Yes	List of two-digit month strings to process (01 through 06)
CHUNK_SIZE	int	Yes	Upload chunk size in bytes (8 MB)

Outputs

Name	Type	Description
GCS Parquet files	GCS objects	Yellow taxi trip Parquet files uploaded to the GCS bucket with their original file names (e.g., yellow_tripdata_2024-01.parquet)
Local Parquet files	Local files	Downloaded Parquet files in the current working directory (not cleaned up after upload)
Verification result	bool	Each upload is verified by checking blob existence in GCS

Usage Examples

Basic Usage

# Run the script directly to download and upload all 6 months:
# python load_yellow_taxi_data.py

# The script's main block handles the full workflow:
if __name__ == "__main__":
    create_bucket(BUCKET_NAME)

    with ThreadPoolExecutor(max_workers=4) as executor:
        file_paths = list(executor.map(download_file, MONTHS))

    with ThreadPoolExecutor(max_workers=4) as executor:
        executor.map(upload_to_gcs, filter(None, file_paths))

    print("All files processed and verified.")

Using Individual Functions

# Download a single month's file
file_path = download_file("03")
# Returns: "./yellow_tripdata_2024-03.parquet"

# Upload with custom retry count
upload_to_gcs("./yellow_tripdata_2024-03.parquet", max_retries=5)

# Verify an upload
exists = verify_gcs_upload("yellow_tripdata_2024-03.parquet")
print(f"Upload verified: {exists}")

Related Pages

• Principle:DataTalksClub_Data_engineering_zoomcamp_GCS_Data_Upload