World Football Insights: Cloud-native serverless ELT
on Google Cloud Platform

A serverless topology with 14 orchestrated tasks

The whole pipeline runs on managed GCP services. Cloud Composer (which is managed Apache Airflow on GKE) hosts the DAG. Dataproc Serverless runs the PySpark transformations without me creating or managing a single cluster. Google Cloud Storage stores raw NDJSON, Silver Parquet, and Gold Parquet. BigQuery materializes Silver and Gold as external tables. No servers. No clusters. No infrastructure I have to maintain.

The DAG has a fan-out / merge / validate / fan-out / merge / fan-out shape. Three entity streams (teams, standings, matches) flow through in parallel where they can, and converge at the two Dataproc jobs and the quality gate.

The Bronze layer is deliberately absent from BigQuery. It stays in GCS as raw NDJSON. This is the honest lakehouse pattern: raw data lives in object storage as the single source of truth, only cleaned and enriched layers live in the warehouse. Analysts who need to audit raw responses can query GCS Parquet through BigQuery External Tables, but the warehouse itself does not carry duplicate storage cost.

Async extraction, quality gates, and explicit schemas

The extract task is an aiohttp async loop that fetches three endpoints from TheSportsDB: all teams in the competition, the league table (standings), and all fixtures for the season. Each response is normalized into newline-delimited JSON and uploaded to GCS via GCSHook. The loop has built-in retry with exponential backoff on HTTP 429 rate limits and non-200 responses. Between requests, it sleeps for a configurable delay to avoid burst-triggering the API.

async def fetch_json(url):
    for attempt in range(RETRIES):
        try:
            async with aiohttp.ClientSession(
                timeout=aiohttp.ClientTimeout(total=30)
            ) as session:
                async with session.get(url) as response:
                    if response.status == 200:
                        data = await response.json()
                        if any(data.values()):
                            return data
                    elif response.status == 429:
                        wait_time = DELAY * (2 ** attempt)
                        logging.warning(f"Rate limited, wait {wait_time}s")
                        await asyncio.sleep(wait_time)
                    else:
                        logging.warning(f"HTTP {response.status}, retry {attempt+1}")
                        await asyncio.sleep(DELAY)
        except Exception as error:
            wait_time = DELAY * (2 ** attempt)
            logging.error(f"Error: {error}, retry {attempt+1}")
            await asyncio.sleep(wait_time)
    return None

Three GCSObjectExistenceSensor tasks then wait in parallel for each raw file to land. Sensor mode is reschedule, which releases the worker slot between pokes so a slow API upstream does not tie up Composer resources. Poke interval is 15 seconds, timeout is 5 minutes. If any file fails to arrive, the downstream Bronze-to-Silver job never starts.

Bronze-to-Silver runs as a Dataproc Serverless PySpark batch. It reads NDJSON, explodes the nested API arrays, renames fields to snake_case, casts data types where appropriate, and writes partitioned Parquet with snappy compression to gs://.../silver/. Silver-to-Gold does the same, plus adds derived metrics through Spark SQL functions and window aggregations for group standings.

Between Silver writes and Silver-to-Gold, a Python validation task reads each Silver Parquet from GCS, checks row count is non-zero, and confirms required columns exist. If any check fails, the whole DAG fails before Gold is touched. This is the "validate before load" principle applied to a cloud pipeline: if Silver is broken, do not let broken data propagate to the analytics layer.

def validate_silver(**context):
    gcs = GCSHook()
    errors = []
    for entity in ["teams", "standings", "matches"]:
        folder_prefix = f"{PATHS['silver']}/{entity}/"
        all_files = gcs.list(BUCKET, prefix=folder_prefix)
        parquet_files = [p for p in all_files if p.endswith(".parquet")]
        if not parquet_files:
            errors.append(f"{entity}: no Parquet files found")
            continue
        file_bytes = gcs.download(BUCKET, parquet_files[0])
        df = pq.read_table(BytesIO(file_bytes)).to_pandas()
        if len(df) == 0:
            errors.append(f"{entity}: zero rows")
        required = REQUIRED_COLUMNS[entity]
        for col in required:
            if col not in df.columns:
                errors.append(f"{entity}: missing column {col}")
    if errors:
        raise ValueError(f"Validation failed: {errors}")

The final loads to BigQuery use GCSToBigQueryOperator with external_table=True and schema_fields=... pointing at pre-declared dictionaries derived from SchemaField objects. Every column has its type and mode declared before the load runs. BigQuery does not infer anything.

GOLD_SCHEMAS = {
    "standings": [
        SchemaField("standing_id", "STRING", mode="NULLABLE"),
        SchemaField("position", "STRING", mode="NULLABLE"),
        SchemaField("team_id", "STRING", mode="NULLABLE"),
        SchemaField("team_name", "STRING", mode="NULLABLE"),
        # ... 12 more fields ...
        SchemaField("points_per_game", "FLOAT", mode="NULLABLE"),
        SchemaField("win_percentage", "FLOAT", mode="NULLABLE"),
        SchemaField("goals_scored_per_game", "FLOAT", mode="NULLABLE"),
        SchemaField("goals_conceded_per_game", "FLOAT", mode="NULLABLE"),
        SchemaField("running_total_points", "FLOAT", mode="NULLABLE"),
        SchemaField("gold_loaded_at", "TIMESTAMP", mode="NULLABLE"),
    ],
    # teams and matches schemas omitted for brevity
}

# Convert SchemaField objects to JSON-safe dicts for the operator
GOLD_SCHEMA_DICTS = {
    name: [field.to_api_repr() for field in schema_list]
    for name, schema_list in GOLD_SCHEMAS.items()
}

Three runs, one debugging story, honest numbers

The pipeline shipped end-to-end on Cloud Composer. Below are three real runs I captured, in the order I made them. Each row is a real Composer run ID; each duration is what Airflow measured.

The pattern is real and worth naming honestly. First run: 1h 13m. Fix a two-line config bug: 11m 53s. Add production-grade schemas: 24m 53s. Each move was a genuine engineering decision with a genuine tradeoff. The final production shape (Run 3) is slower than the peak-optimized run, and that is the correct answer for what the pipeline needs to do.

What this project does not do (yet)