AWS Serverless Weather Data Pipeline

Building a Serverless Weather Pipeline on AWS: A Step-by-Step Walkthrough

This is a build log for someone who's used AWS a bit — deployed a Lambda from the console, poked around S3 — but hasn't touched CDK, Step Functions, EventBridge Scheduler, or GitHub's OIDC setup before. I'll explain each concept the first time it comes up, and show the actual code behind every piece, roughly in the order I built it.

Here's what it ends up doing: every 10 minutes, EventBridge Scheduler kicks off a Step Functions workflow that pulls current weather for five cities in parallel from a free public API, reshapes the results into JSON Lines, drops them into S3 in a partitioned layout, and makes them queryable in Athena with plain SQL. No crawler, and no AWS credentials sitting anywhere in the GitHub repo that deploys it.

kasukur / serverless-weather-pipeline

AWS Serverless Weather Pipeline

Serverless Weather Data Pipeline

A small but complete serverless data pipeline on AWS walkthrough: EventBridge Scheduler → Step Functions → Lambda → S3 → Glue/Athena, deployed by GitHub Actions with no AWS access keys stored anywhere (authentication is via GitHub's OIDC provider).

flowchart TD
    A["EventBridge Scheduler (every 10 min)"] --> B["Step Functions state machine"]
    B --> C["PrepareCities (Pass)"]
    C --> D["ForEachCity (Map, concurrency 4)"]
    D --> E["FetchWeather (Lambda -> Open-Meteo public API)"]
    E -.-> F["retries transient errors (up to 2 attempts)"]
    E -.-> G["FetchFailed (Pass): per-city failure absorbed here, other cities continue"]
    E --> H["TransformWeatherData (Lambda, pure function, no AWS calls)"]
    H -.-> I["splits successes vs failures"]
    H -.-> J["builds JSON-Lines body + partitioned S3 key"]
    H --> K["LoadToS3 (Lambda, writes to S3 via boto3)"]
    K --> L["S3 (processed/dt=YYYY-MM-DD/hour=HH/*.jsonl)"]
    L --> M["Glue Data Catalog table (partition projection -- no crawler)"]
    M --> N["Athena (query with plain SQL)"]
    D -.->

…

View on GitHub

Table of Contents

• What we're building, and why each piece is there
• Step 1: Project setup
• Step 2: Write the Lambdas first, before touching any infrastructure code
• Step 3: Unit test the logic before it ever touches AWS
• Step 4: Wire it together as a Step Functions state machine, in CDK
• Step 5: Trigger it on a schedule — EventBridge Scheduler
• Step 6: Land the data in S3, partitioned for querying
• Step 7: Make it queryable without a crawler — Glue partition projection
• Step 8: Wire up failure notifications
• Step 9: Deploy without a single AWS access key — GitHub OIDC
• Step 10: The GitHub Actions workflows themselves
• Step 11: Bootstrap and the first deploy
• Step 12: A bug that got past every green checkmark
• Step 13: Query the data
• Step 14: Cost and cleanup
• What this was actually a demo of

What we're building, and why each piece is there

Six services, each doing one job.

EventBridge Scheduler is the trigger: a managed cron that kicks off a Step Functions execution on a schedule, so nothing has to sit around just calling StartExecution on a timer.

Step Functions is the orchestrator. It describes a sequence of steps — branching, retries, parallelism — as a state machine (AWS calls the format Amazon States Language), instead of that logic living inside application code. I reached for it because the "fetch five cities" step genuinely needs to run in parallel, retry the ones that fail transiently, and let the rest keep going if one city's API call fails outright. In Step Functions that's a few lines of declarative config. Hand-rolled inside one Lambda, it's a surprising amount of bookkeeping for something that sounds simple.

Lambda does the actual work at each step — three small, single-purpose functions, each one calling the next.

S3 is where the data lands, laid out with dt=/hour= prefixes so it reads like a partitioned table without needing an actual database.

Glue Data Catalog and Athena turn that S3 layout into something you can run SQL against. Glue holds the schema, Athena runs the queries.

SNS catches anything that goes wrong anywhere in the workflow and emails an alert, so a broken run doesn't just fail quietly and go unnoticed.

Step 1: Project setup

The whole thing is one CDK app (Python), split into two independent stacks. One sets up secretless GitHub deploys; the other is the actual pipeline. The split matters — the deploy-credentials stack gets deployed once, manually, with your own AWS credentials, and the pipeline stack deploys itself from GitHub Actions after that, automatically, forever, without a human needing credentials again.

mkdir serverless-weather-pipeline && cd serverless-weather-pipeline
python3 -m venv .venv
source .venv/bin/activate
npm install -g aws-cdk

# requirements.txt
aws-cdk-lib>=2.150.0,<3.0.0
constructs>=10.3.0,<11.0.0

# requirements-dev.txt
pytest>=8.0.0
boto3>=1.34.0

pip install -r requirements.txt -r requirements-dev.txt

Here's how it's laid out in the end:

app.py                        CDK app entry point, wires up both stacks
stacks/
  pipeline_stack.py           The data pipeline itself
  github_oidc_stack.py        OIDC provider + scoped deploy role
lambdas/
  fetch_weather/handler.py    Calls the weather API for one city
  transform/handler.py        Pure function -> JSON Lines
  load/handler.py             Writes the JSON Lines body to S3
tests/
  test_fetch_weather.py
  test_transform.py
  test_load.py
  test_stack_synth.py         CDK assertions: resources exist, IAM is scoped right
athena/sample_queries.sql
.github/workflows/
  test.yml                    Every PR: pytest + cdk synth, no AWS access
  deploy.yml                  Every push to main: OIDC, no static keys

Step 2: Write the Lambdas first, before touching any infrastructure code

Infrastructure-as-code has a slow feedback loop: write code, synthesize, deploy, wait, check. Application logic doesn't have to. Every Lambda here is plain Python with a handler(event, context) entry point, written and unit-tested with pytest before any CDK code even references them.

Fetch calls the weather API for one city. It has zero third-party dependencies — just urllib from the standard library — so there's no dependency layer to build or keep in sync, and the deployment package stays tiny:

import datetime
import json
import urllib.error
import urllib.request

OPEN_METEO_URL = "https://api.open-meteo.com/v1/forecast"
REQUEST_TIMEOUT_SECONDS = 10


def handler(event, context):
    city = event["city"]
    latitude = event["latitude"]
    longitude = event["longitude"]

    query = (
        f"?latitude={latitude}&longitude={longitude}"
        "&current_weather=true&timezone=UTC"
    )
    url = OPEN_METEO_URL + query

    try:
        with urllib.request.urlopen(url, timeout=REQUEST_TIMEOUT_SECONDS) as response:
            payload = json.loads(response.read().decode("utf-8"))
    except (urllib.error.URLError, TimeoutError) as exc:
        raise RuntimeError(f"Failed to fetch weather for {city}: {exc}") from exc

    current = payload.get("current_weather")
    if not current:
        raise RuntimeError(f"No current_weather field in Open-Meteo response for {city}: {payload}")

    return {
        "city": city,
        "latitude": latitude,
        "longitude": longitude,
        "temperature_c": current["temperature"],
        "windspeed_kmh": current["windspeed"],
        "winddirection_deg": current["winddirection"],
        "weathercode": current["weathercode"],
        "observation_time": current["time"],
        "fetched_at": datetime.datetime.now(datetime.timezone.utc).isoformat(),
    }

Notice it raises on failure instead of catching and returning some error object. That's on purpose. This Lambda gets invoked once per city inside a Step Functions Map state, and Step Functions' own retry/catch handles a raised exception natively. Swallow the error here instead, and you're just writing that same retry logic by hand, badly.

Transform takes the batch of per-city results (some of which may be failures, if a city never came back even after retries) and turns the successful ones into a JSON Lines body plus a partitioned S3 key. I wrote it as a pure function with the AWS-facing handler as a thin wrapper, specifically so it's trivial to test without mocking anything:

import json
import re
from datetime import datetime


def _partition_parts(run_started_at: str):
    dt = datetime.fromisoformat(run_started_at.replace("Z", "+00:00"))
    return dt.strftime("%Y-%m-%d"), dt.strftime("%H")


def _sanitize(run_id: str) -> str:
    return re.sub(r"[^A-Za-z0-9_-]", "-", run_id)


def build_records(fetch_results, run_id: str, run_started_at: str):
    records = []
    failures = []
    for item in fetch_results:
        if isinstance(item, dict) and "error" in item:
            failures.append({"city": item.get("city", "unknown"), "error": item["error"]})
        else:
            records.append(item)

    date_part, hour_part = _partition_parts(run_started_at)
    key = f"processed/dt={date_part}/hour={hour_part}/{_sanitize(run_id)}.jsonl"
    body = "\n".join(json.dumps(record, sort_keys=True) for record in records)

    return {
        "key": key,
        "body": body,
        "record_count": len(records),
        "failure_count": len(failures),
        "failures": failures,
    }


def handler(event, context):
    fetch_results = event["fetch_results"]
    run_id = event["run_id"]
    run_started_at = event["run_started_at"]

    result = build_records(fetch_results, run_id, run_started_at)

    if result["record_count"] == 0:
        raise RuntimeError(
            f"All {result['failure_count']} weather fetches failed for run {run_id}: {result['failures']}"
        )

    return {
        "key": result["key"],
        "body": result["body"],
        "record_count": result["record_count"],
        "failure_count": result["failure_count"],
    }

run_started_at comes from Step Functions' own execution context, not datetime.now() inside the function. Small detail, but it keeps the partition deterministic and testable instead of depending on whatever the wall clock happens to say the moment the Lambda runs.

Step 3: Unit test the logic before it ever touches AWS

SUCCESS_ITEM = {
    "city": "Sydney", "latitude": -33.8688, "longitude": 151.2093,
    "temperature_c": 18.4, "windspeed_kmh": 12.1, "winddirection_deg": 200,
    "weathercode": 1, "observation_time": "2026-08-23T10:00",
    "fetched_at": "2026-08-23T10:00:03+00:00",
}
FAILURE_ITEM = {
    "city": "Atlantis", "latitude": 0, "longitude": 0,
    "error": {"Error": "RuntimeError", "Cause": "Failed to fetch weather for Atlantis: timed out"},
}

def test_build_records_splits_success_and_failure():
    result = build_records([SUCCESS_ITEM, FAILURE_ITEM], run_id="test-run", run_started_at="2026-08-23T10:15:00Z")
    assert result["record_count"] == 1
    assert result["failure_count"] == 1

def test_build_records_partition_key_uses_run_started_at():
    result = build_records(
        [SUCCESS_ITEM], run_id="2026-08-23T10:00:00Z_abc123", run_started_at="2026-08-23T10:00:00.512Z",
    )
    assert result["key"] == "processed/dt=2026-08-23/hour=10/2026-08-23T10-00-00Z_abc123.jsonl"

The fetch test mocks urllib.request.urlopen directly instead of pulling in a heavier HTTP-mocking library. Stdlib in, stdlib mocked out:

from unittest.mock import patch

def test_handler_returns_normalized_weather():
    fake_payload = json.dumps({
        "current_weather": {
            "temperature": 21.3, "windspeed": 9.4, "winddirection": 180,
            "weathercode": 2, "time": "2026-08-23T10:00",
        }
    }).encode("utf-8")

    with patch("fetch_weather_handler.urllib.request.urlopen", return_value=_FakeResponse(fake_payload)):
        result = handler({"city": "Sydney", "latitude": -33.8688, "longitude": 151.2093}, None)

    assert result["temperature_c"] == 21.3

One thing that tripped me up briefly: every Lambda's entry file is named handler.py — that's just the Lambda convention, handler.handler as the configured entry point — which means plain Python import machinery can't tell lambdas/fetch_weather/handler.py apart from lambdas/transform/handler.py. Import both as a bare module called handler in the same test run and whichever loads first wins for every test file after it. The fix is loading each one by explicit file path under its own module name:

import importlib.util, pathlib, sys

_HANDLER_PATH = pathlib.Path(__file__).resolve().parents[1] / "lambdas" / "transform" / "handler.py"
_spec = importlib.util.spec_from_file_location("transform_handler", _HANDLER_PATH)
transform_handler = importlib.util.module_from_spec(_spec)
sys.modules[_spec.name] = transform_handler
_spec.loader.exec_module(transform_handler)

A little bit of ceremony, but it lets every Lambda file keep the same conventional name without the test suite getting confused about which one it's actually running.

Step 4: Wire it together as a Step Functions state machine, in CDK

This is the part where CDK actually saves real effort — instead of hand-writing Amazon States Language JSON, the workflow gets built out of Python objects that generate it for you. Chain a Pass (a no-op that just reshapes input) into a Map, into two more Lambda invocations:

prepare_cities = sfn.Pass(
    self, "PrepareCities",
    parameters={
        "cities": CITIES,
        "run_id.$": "$$.Execution.Name",
        "run_started_at.$": "$$.Execution.StartTime",
    },
)

That .$ suffix is Amazon States Language for "resolve this as a JSONPath expression against the input, not a literal string." $$.Execution.StartTime is one of a handful of fields Step Functions injects automatically about the execution itself — that's where run_started_at actually comes from in the transform Lambda above.

fetch_task = tasks.LambdaInvoke(
    self, "FetchWeather",
    lambda_function=fetch_fn,
    payload_response_only=True,
)
fetch_task.add_retry(errors=["States.ALL"], interval=Duration.seconds(2), max_attempts=2, backoff_rate=2.0)
fetch_task.add_catch(fetch_failed, errors=["States.ALL"], result_path="$.error")

for_each_city = sfn.Map(
    self, "ForEachCity",
    items_path="$.cities",
    max_concurrency=4,  # be polite to the free public API
    result_path="$.fetch_results",
)
for_each_city.item_processor(fetch_task)

payload_response_only=True unwraps the Lambda's raw return value directly into the state's output. Skip it, and you get the full invocation envelope — ExecutedVersion, StatusCode, Payload, and so on — and every downstream step has to reach into $.Payload just to get at what you actually care about. The retry policy gives each city two tries with a 2-second backoff before it gives up; the catch means a city that still fails after retries doesn't take the whole Map down with it — it gets routed to a Pass state that records the error and moves on, while everyone else keeps going.

transform_task = tasks.LambdaInvoke(
    self, "TransformWeatherData",
    lambda_function=transform_fn,
    payload_response_only=True,
    payload=sfn.TaskInput.from_object({
        "fetch_results.$": "$.fetch_results",
        "run_id.$": "$.run_id",
        "run_started_at.$": "$.run_started_at",
    }),
    result_path="$.transformed",
)

And chain it all together:

definition = prepare_cities.next(for_each_city).next(transform_task).next(load_to_s3)

state_machine = sfn.StateMachine(
    self, "WeatherPipelineStateMachine",
    state_machine_name="weather-ingestion-pipeline",
    definition_body=sfn.DefinitionBody.from_chainable(definition),
    timeout=Duration.minutes(5),
    tracing_enabled=True,
    logs=sfn.LogOptions(destination=state_machine_log_group, level=sfn.LogLevel.ALL),
)

load_to_s3, the last step, deserves its own section further down — the first version of it was wrong in a way that took a while to track down.

Step 5: Trigger it on a schedule — EventBridge Scheduler

Not the older EventBridge "rules" cron feature — this is a newer, dedicated scheduling service with its own resource type, built for "start this one thing on this cadence" rather than routing arbitrary events around.

scheduler_role = iam.Role(
    self, "SchedulerExecutionRole",
    assumed_by=iam.ServicePrincipal("scheduler.amazonaws.com"),
)
state_machine.grant_start_execution(scheduler_role)

scheduler.CfnSchedule(
    self, "HourlyWeatherSchedule",
    schedule_expression="rate(10 minutes)",
    flexible_time_window=scheduler.CfnSchedule.FlexibleTimeWindowProperty(mode="OFF"),
    target=scheduler.CfnSchedule.TargetProperty(
        arn=state_machine.state_machine_arn,
        role_arn=scheduler_role.role_arn,
        retry_policy=scheduler.CfnSchedule.RetryPolicyProperty(maximum_retry_attempts=1),
        input="{}",
    ),
    state="ENABLED",
)

grant_start_execution is CDK's IAM shorthand — it writes an IAM policy statement for exactly states:StartExecution on exactly this state machine's ARN, attached to exactly this role, so you never have to spell either ARN out by hand. flexible_time_window=OFF means "run at exactly this cadence" rather than letting AWS jitter the start time to spread load. rate(10 minutes) is a demo-friendly cadence for generating data fast; rate(1 hour) (or a cron(...) expression, if you want something less regular) makes more sense for anything left running unattended.

Step 6: Land the data in S3, partitioned for querying

data_bucket = s3.Bucket(
    self, "WeatherDataBucket",
    block_public_access=s3.BlockPublicAccess.BLOCK_ALL,
    enforce_ssl=True,
    removal_policy=RemovalPolicy.DESTROY,
    auto_delete_objects=True,  # demo convenience; use RETAIN in production
    lifecycle_rules=[
        s3.LifecycleRule(id="ExpireAthenaQueryResults", prefix="athena-results/", expiration=Duration.days(7)),
    ],
)

removal_policy=DESTROY plus auto_delete_objects=True means cdk destroy actually empties and deletes this bucket instead of orphaning it. CDK's default, sensibly, is to leave S3 buckets behind on stack deletion — deleting someone's real data by accident is a much worse failure mode than an orphaned bucket costing a few cents a month. That default is worth flipping for a throwaway demo and very much not worth flipping anywhere it holds data you'd miss, which is exactly why the comment next to it says so.

Objects land at something like processed/dt=2026-08-23/hour=14/<run-id>.jsonl. That layout is what makes the next part work.

Step 7: Make it queryable without a crawler — Glue partition projection

The conventional way to make an S3 prefix queryable in Athena is a Glue Crawler — a job you schedule that scans the bucket, infers the schema, and registers each partition it finds in the Glue Data Catalog. It works fine, but it costs money to run, and there's always a lag between a new partition landing in S3 and the crawler getting around to noticing it.

Partition projection skips the crawler part entirely. Instead of discovering partitions by scanning S3, you tell Glue the pattern your partitions follow, and Athena computes valid values straight from the query's WHERE clause at query time:

table_parameters = {
    "classification": "json",
    "projection.enabled": "true",
    "projection.dt.type": "date",
    "projection.dt.format": "yyyy-MM-dd",
    "projection.dt.range": "2024-01-01,NOW",
    "projection.dt.interval": "1",
    "projection.dt.interval.unit": "DAYS",
    "projection.hour.type": "integer",
    "projection.hour.range": "0,23",
    "projection.hour.digits": "2",
    "storage.location.template": f"s3://{data_bucket.bucket_name}/processed/dt=${{dt}}/hour=${{hour}}/",
}

glue_table = glue.CfnTable(
    self, "WeatherGlueTable",
    catalog_id=self.account,
    database_name=GLUE_DATABASE_NAME,
    table_input=glue.CfnTable.TableInputProperty(
        name=GLUE_TABLE_NAME,
        table_type="EXTERNAL_TABLE",
        parameters=table_parameters,
        partition_keys=[
            glue.CfnTable.ColumnProperty(name="dt", type="string"),
            glue.CfnTable.ColumnProperty(name="hour", type="string"),
        ],
        storage_descriptor=glue.CfnTable.StorageDescriptorProperty(
            location=f"s3://{data_bucket.bucket_name}/processed/",
            input_format="org.apache.hadoop.mapred.TextInputFormat",
            output_format="org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat",
            serde_info=glue.CfnTable.SerdeInfoProperty(
                serialization_library="org.openx.data.jsonserde.JsonSerDe",
            ),
            columns=[glue.CfnTable.ColumnProperty(name=name, type=t) for name, t in GLUE_COLUMNS],
        ),
    ),
)

Walking through the parts that matter: dt is a date-typed partition running from a fixed start date to NOW, formatted yyyy-MM-dd. hour is a two-digit, zero-padded integer from 0 to 23. storage.location.template tells Athena exactly how to build the S3 path for any dt/hour combination it needs. The serde_info plus input_format/output_format trio is what tells Athena "each line of each file is one independent JSON object," as opposed to, say, one big JSON array per file — and that distinction turns out to matter a lot in a couple of sections.

athena_workgroup = athena.CfnWorkGroup(
    self, "WeatherAthenaWorkGroup",
    name="weather-pipeline-wg",
    work_group_configuration=athena.CfnWorkGroup.WorkGroupConfigurationProperty(
        result_configuration=athena.CfnWorkGroup.ResultConfigurationProperty(
            output_location=f"s3://{data_bucket.bucket_name}/athena-results/",
        ),
        enforce_work_group_configuration=True,
        publish_cloud_watch_metrics_enabled=True,
    ),
)

A dedicated workgroup keeps Athena's query-result scratch files in the same bucket, under athena-results/ (which the earlier lifecycle rule expires after 7 days), instead of scattering them into Athena's account-wide default location.

Step 8: Wire up failure notifications

alerts_topic = sns.Topic(self, "PipelineAlertsTopic", display_name="Weather pipeline failures")
if alert_email:
    alerts_topic.add_subscription(subs.EmailSubscription(alert_email))

notify_failure = tasks.SnsPublish(
    self, "NotifyFailure",
    topic=alerts_topic,
    subject="Weather pipeline execution failed",
    message=sfn.TaskInput.from_text("The weather pipeline failed. Check the Step Functions execution history for details."),
)

for state in (for_each_city, transform_task, load_to_s3):
    state.add_catch(notify_failure, errors=["States.ALL"], result_path="$.stateMachineError")

Any of those three states failing outright (as opposed to a single city inside the Map, which has its own narrower catch from Step 4) routes to an SNS publish, instead of leaving a red FAILED execution in the console that nobody happens to check. A CloudWatch alarm on the state machine's failure metric does the same job as a second layer, tied to the same topic.

Step 9: Deploy without a single AWS access key — GitHub OIDC

Here's the constraint that shaped basically every decision about how this deploys: no AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY pair, stored as a GitHub secret, anywhere, ever. The usual argument against static keys applies even more once a repo is public: a leaked long-lived key is a standing liability until someone notices and rotates it, and "nobody got around to rotating it" is a very common way this actually goes wrong.

OIDC federation replaces that with something that expires by design. On every run, GitHub's own identity provider issues the workflow a short-lived, signed token — proof of exactly which repo, branch, and workflow run this is, valid for that one run and nothing else. AWS is set up to trust tokens signed by GitHub's provider, and an IAM role's trust policy spells out exactly which token claims are allowed to assume it. No secret ever gets generated, stored, or rotated — the whole handshake happens fresh, every time.

provider = iam.OpenIdConnectProvider(
    self, "GitHubOidcProvider",
    url="https://token.actions.githubusercontent.com",
    client_ids=["sts.amazonaws.com"],
)

sub_condition = f"repo:{github_owner}/{github_repo}:ref:refs/heads/{github_branch}"

deploy_role = iam.Role(
    self, "GitHubActionsDeployRole",
    assumed_by=iam.FederatedPrincipal(
        provider.open_id_connect_provider_arn,
        conditions={
            "StringEquals": {"token.actions.githubusercontent.com:aud": "sts.amazonaws.com"},
            "StringLike": {"token.actions.githubusercontent.com:sub": sub_condition},
        },
        assume_role_action="sts:AssumeRoleWithWebIdentity",
    ),
    max_session_duration=Duration.hours(1),
)

That sub_condition line is doing all the real work — it's the only thing stopping some other GitHub repo from assuming this role, even though the OIDC provider itself trusts all of GitHub equally. It's also almost exactly the line that turned into the hardest bug in the whole project, so it's worth showing what actually happened.

The bug: "Not authorized to perform sts:AssumeRoleWithWebIdentity"

With the stack above deployed and the role ARN saved as a repo variable, the deploy workflow could get its OIDC token just fine — but the AssumeRoleWithWebIdentity call on the AWS side came back flatly denied. I went through every documented cause one at a time. The OIDC provider existed, with the right thumbprint and the right audience in its client ID list. The role's trust policy had the right principal and the right action. I even dropped two temporary debug steps into deploy.yml just to rule out anything weird about the token itself:

- name: Debug OIDC context
  run: |
    echo "repository: ${{ github.repository }}"
    echo "ref: ${{ github.ref }}"
    echo "event_name: ${{ github.event_name }}"

- name: Debug role ARN
  run: echo "Assuming role: ${{ vars.AWS_DEPLOY_ROLE_ARN }}"

Both printed exactly what I expected — right repo, right ref, right role ARN. The trust policy, read straight back out of IAM, matched sub_condition above character for character. Every piece checked out on its own. It still failed.

The actual answer wasn't in the trust policy or the workflow logs at all. It was sitting in CloudTrail, which records the literal API call AWS evaluated and denied:

aws cloudtrail lookup-events \
  --lookup-attributes AttributeKey=EventName,AttributeValue=AssumeRoleWithWebIdentity

The denied event showed the real sub claim GitHub was actually sending: repo:kasukur@1291877/serverless-weather-pipeline@1343631754:ref:refs/heads/main. GitHub now appends an immutable numeric ID to both the account name and the repo name in that claim, presumably so a trust policy can't get quietly hijacked later if an account or repo gets renamed and the old name is picked up by someone else. That detail isn't in most OIDC-with-GitHub-Actions walkthroughs, which still document the plain repo:OWNER/REPO:ref:refs/heads/BRANCH format — and an exact match against that older format can never succeed against what GitHub actually sends now.

The fix: switch from an exact match to a StringLike wildcard, but only on the @<id> suffix — the owner and repo names themselves stay pinned exactly as before:

sub_condition = f"repo:{github_owner}@*/{github_repo}@*:ref:refs/heads/{github_branch}"

Same workflow run, no other changes, and it went straight through. The two debug steps came out of deploy.yml right after — they'd done their job, and there's no reason to leave a workflow permanently echoing role ARNs into a log. If a trust policy looks right on paper and still gets denied, cloudtrail lookup-events against the exact failing call is worth reaching for early. It shows the literal claim value AWS actually evaluated, not what a two-year-old tutorial says it should be.

Step 10: The GitHub Actions workflows themselves

Two workflows, split by trust level. Pull requests only get tests, no AWS access at all:

name: Test
on:
  pull_request:
    branches: [main]
  push:
    branches: [main]
permissions:
  contents: read
jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: '3.12' }
      - uses: actions/setup-node@v4
        with: { node-version: '20' }
      - run: pip install -r requirements.txt -r requirements-dev.txt
      - run: npm install -g aws-cdk
      - run: pytest -v
      - run: cdk synth --quiet --all

Pushes to main get the real deploy. permissions: id-token: write is the one line that actually lets this job request an OIDC token in the first place — it's opt-in per job, not something implied by repo settings elsewhere:

name: Deploy
on:
  push:
    branches: [main]
  workflow_dispatch: {}
permissions:
  id-token: write
  contents: read
jobs:
  deploy:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: '3.12' }
      - uses: actions/setup-node@v4
        with: { node-version: '20' }
      - run: pip install -r requirements.txt -r requirements-dev.txt
      - run: npm install -g aws-cdk
      - run: pytest -v
      - name: Configure AWS credentials via GitHub OIDC
        uses: aws-actions/configure-aws-credentials@v4
        with:
          role-to-assume: ${{ vars.AWS_DEPLOY_ROLE_ARN }}
          aws-region: ${{ vars.AWS_REGION }}
      - run: >
          cdk deploy WeatherPipeline --require-approval never
          --context github_owner=${{ github.repository_owner }}
          --context github_repo=${{ github.event.repository.name }}
          --context github_branch=main

AWS_DEPLOY_ROLE_ARN and AWS_REGION live as GitHub repo Variables, not Secrets. That's deliberate, not an oversight — an IAM role ARN isn't sensitive on its own. The actual security boundary is the trust policy's sub condition from the last section, which means knowing the ARN doesn't get anyone anywhere unless the request also comes from this exact repo and branch.

Step 11: Bootstrap and the first deploy

Three one-time steps, run locally with your own AWS credentials. None of this ever runs in CI.

cdk bootstrap aws://ACCOUNT_ID/ap-southeast-2

This sets up the S3 bucket and IAM roles CDK itself uses to stage and deploy assets. If it fails with the bootstrap stack stuck in UPDATE_FAILED, complaining about a missing staging bucket — this happens if that bucket ever gets deleted manually, outside CloudFormation — recover it with aws cloudformation continue-update-rollback --stack-name CDKToolkit --resources-to-skip StagingBucket and re-run bootstrap. If that doesn't fully clear it, deleting the CDKToolkit stack outright and bootstrapping fresh works too.

cdk deploy WeatherPipeline-GitHubOidc \
  --context github_owner=<your-github-username-or-org> \
  --context github_repo=<your-repo-name> \
  --context github_branch=main

Grab the DeployRoleArn from the output and add it as a repo variable as described above. Worth double-checking CDK_DEFAULT_REGION/AWS_REGION are actually set in your shell before running this — an unset region variable makes CDK quietly fall back to us-east-1, and the next time you check the region you actually meant, it looks like nothing deployed at all.

From here on, every change ships by pushing to main — deploy.yml runs the tests, then cdk deploy WeatherPipeline, using the role from the stack above.

Step 12: A bug that got past every green checkmark

With deploys working, the pipeline ran on schedule, every Step Functions execution showed SUCCEEDED, and Athena returned rows for every partition — except every data column came back NULL.

That combination is a pretty specific clue: nothing failed, so the problem has to be in the data itself, not the workflow logic. Downloading one of the raw S3 objects and running wc -l against it came back 0 — zero newlines, in a file that was supposed to hold five JSON Lines records. Looking at the actual bytes explained why:

The LoadToS3 step, as I'd originally written it, used a Step Functions native SDK integration — a way to call an AWS API directly from a state machine definition, no Lambda required:

load_to_s3 = tasks.CallAwsService(
    self, "LoadToS3",
    service="s3",
    action="putObject",
    parameters={
        "Bucket": data_bucket.bucket_name,
        "Key.$": "$.transformed.key",
        "Body.$": "$.transformed.body",
    },
    iam_resources=[data_bucket.arn_for_objects("processed/*")],
    result_path=sfn.JsonPath.DISCARD,
)

It's a neat trick when it works, and honestly, it looked correct — Body.$ resolves to the transform step's JSON Lines string, same as every other .$ reference in this workflow. The problem is that S3's Body parameter is fundamentally a raw byte blob, and Amazon States Language has no concept of "raw bytes" — everything moving through a state machine is JSON. When a multi-line string passed through as Body.$, what actually landed in S3 was the JSON-string-encoded version of it: the outer quotes stayed on, and every embedded newline and quote turned into the literal two-character escape sequence you'd see inside a JSON string — \n, \" — instead of the real bytes they were supposed to represent. Athena's JSON SerDe, reading the file line by line, hit one line that wasn't valid standalone JSON and gave up quietly. That's why it came back as NULL everywhere rather than an outright error — the SerDe's failure mode for a bad line is nulls for that row, not a blown-up query.

The fix: stop asking a JSON-only state language to carry raw bytes, and just write the file with boto3 directly.

# lambdas/load/handler.py
import boto3

s3 = boto3.client("s3")

def handler(event, context):
    bucket = event["bucket"]
    key = event["key"]
    body = event["body"]

    s3.put_object(
        Bucket=bucket,
        Key=key,
        Body=body.encode("utf-8"),
        ContentType="application/json",
    )

    return {"bucket": bucket, "key": key}

body.encode("utf-8") hands boto3 actual bytes, with real newline characters exactly where they belong. Nothing in between to garble it, because nothing in between only speaks JSON. The state machine's last step became a plain LambdaInvoke, same shape as the other two:

load_to_s3 = tasks.LambdaInvoke(
    self, "LoadToS3",
    lambda_function=load_fn,
    payload_response_only=True,
    payload=sfn.TaskInput.from_object({
        "bucket": data_bucket.bucket_name,
        "key.$": "$.transformed.key",
        "body.$": "$.transformed.body",
    }),
    result_path=sfn.JsonPath.DISCARD,
)

And a regression test locks in exactly the distinction that mattered:

def test_handler_preserves_real_newlines_in_body_bytes():
    body = '{"city": "Sydney"}\n{"city": "Melbourne"}'
    module.handler({"bucket": "b", "key": "k", "body": body}, None)

    sent_body = module.s3.put_object.call_args.kwargs["Body"]
    assert sent_body.count(b"\n") == 1       # a real newline byte, not text
    assert b"\\n" not in sent_body            # not the two-character escape sequence
    assert not sent_body.startswith(b'"')     # not JSON-string-encoded

The lesson generalizes past this one integration: native SDK integrations are a genuinely nice way to skip a Lambda, but only for parameters that map cleanly onto JSON's type system. Anything that's conceptually a byte blob — a file body, binary data, anything with meaningful embedded whitespace — is worth testing with a real multi-line payload before you trust it, because a short single-line string will round-trip through the exact same bug without ever revealing it.

Step 13: Query the data

No crawler run, no MSCK REPAIR TABLE — partition projection means Athena computes valid dt=/hour= partitions directly from the query, so a partition is queryable the moment its file lands in S3.

SELECT city, temperature_c, windspeed_kmh, observation_time
FROM weather_pipeline_db.observations
WHERE dt = '2026-08-23' AND hour = '12'
ORDER BY city;

A couple more, to show partition projection actually working across the WHERE clause rather than one fixed partition:

-- Average temperature per city over the last 7 days
SELECT city, round(avg(temperature_c), 1) AS avg_temp_c, count(*) AS observations
FROM weather_pipeline_db.observations
WHERE dt >= date_format(current_date - INTERVAL '7' DAY, '%Y-%m-%d')
GROUP BY city
ORDER BY avg_temp_c DESC;

-- Latest observation per city, across all partitions
SELECT city, temperature_c, windspeed_kmh, observation_time
FROM (
    SELECT city, temperature_c, windspeed_kmh, observation_time,
           row_number() OVER (PARTITION BY city ORDER BY fetched_at DESC) AS rn
    FROM weather_pipeline_db.observations
)
WHERE rn = 1;

Step 14: Cost and cleanup

Everything here is pay-per-use and small at this scale. Lambda invocations and Step Functions state transitions sit comfortably inside the free tier, S3 storage for a handful of JSON files is fractions of a cent, and Athena charges per byte scanned — a few cents at most for a demo this size. The one ongoing cost actually worth watching is the schedule itself: it keeps invoking the pipeline on whatever cadence it's set to, forever, until something disables or destroys it.

cdk destroy WeatherPipeline

Leave WeatherPipeline-GitHubOidc deployed if you'll reuse the same deploy role later — it costs nothing sitting idle, and keeping it means never redoing the OIDC setup. Destroy that one too, with cdk destroy WeatherPipeline-GitHubOidc, once you're actually done with the project.

What this was actually a demo of

The weather data itself was never really the point — it's just a convenient, keyless public API that produces enough structure to be worth transforming and querying. The real subject is everything around it: orchestrating parallel work with per-item retry and partial-failure handling instead of one monolithic function, deploying with zero long-lived credentials via OIDC (and the real gap between how that's documented and how it actually behaves today), querying partitioned data in Athena without a crawler, and — probably the one lesson from all three bugs above that travels furthest — the fact that "the execution succeeded" and "the data is correct" are two different claims, and only one of them ever shows up as a red X in a console.