app.dataengineerhub.blog

Tag: etl

  • Snowflake Time Travel vs. Fail-safe: What Gets Recovered and When

    Snowflake Time Travel vs. Fail-safe: What Gets Recovered and When

    3:14 a.m., and a migration script hands off to DROP TABLE orders_staging; against what everyone on the team swore was a permanent table. It wasn’t. Somewhere in the last quarter it got recreated as TRANSIENT to shave storage costs, and nobody updated the runbook. The on-call engineer isn’t worried — Snowflake has Time Travel, Snowflake has Fail-safe, this is a solved problem. Except Fail-safe doesn’t apply to transient tables. Zero days. The table is gone the moment its one-day Time Travel window closes, and by the time anyone notices, it already has. Six hours of ingestion, rebuilt by hand from source, on a Saturday.

    That’s the gap this article is about. Time Travel and Fail-safe get talked about together so often that people assume they’re one continuous safety net. They’re not the same feature, they don’t behave the same way, and the difference has real financial and recovery-time consequences that most teams only discover during an incident.

    TL;DR

    • → Time Travel lets you query, clone, or UNDROP historical data yourself for a retention window you configure — 0 to 1 day on Standard Edition, up to 90 days on Enterprise Edition and above, for permanent objects.
    • → Fail-safe is a separate, fixed 7-day recovery period that starts after Time Travel expires, and it is not self-service — only Snowflake Support can pull data back from it, and only for permanent tables.
    • → Transient and temporary tables carry zero Fail-safe days. They’re cheaper precisely because Snowflake gives up that protection.
    • → Both features bill as storage: Time Travel data accrues at the normal storage rate, and a table with heavy daily updates on a long retention window can multiply its effective storage several times over.
    • → Time Travel doesn’t create a second copy of your table — it retains the old micro-partitions that a write would otherwise discard, via Snowflake’s copy-on-write architecture.
    • → UNDROP TABLE, AT, and BEFORE only work inside the Time Travel window. Once an object crosses into Fail-safe, none of those commands work anymore.

    What Time Travel Actually Stores

    Snowflake tables are stored as immutable micro-partitions — compressed, columnar chunks of roughly 50–500 MB of uncompressed data each. When you run an UPDATE or DELETE, Snowflake doesn’t rewrite rows in place. It writes new micro-partitions reflecting the change and stops referencing the old ones from the table’s current state. That’s the whole trick: Time Travel is Snowflake choosing not to immediately throw those old partitions away.

    Snowflake doesn’t back up a table for Time Travel — it just delays discarding the partitions a write would otherwise drop.

    Every table, schema, and database has a DATA_RETENTION_TIME_IN_DAYS parameter that controls how long those superseded partitions stick around before they’re eligible for permanent deletion. Retention is inherited: set it on a database and every schema and table created under it picks up the value unless overridden lower down. There’s also an account-level MIN_DATA_RETENTION_TIME_IN_DAYS floor — if it’s set, the effective retention for any object becomes whichever is larger, its own setting or the floor. That parameter is easy to forget exists and even easier to be surprised by months later.

    Standard Edition accounts get 1 day of retention by default, and you can only turn it down to 0 — there’s no way to go longer without upgrading to Enterprise Edition. Enterprise and above allow up to 90 days for permanent databases, schemas, and tables, configurable per object.

    Retention by Table Type — the Comparison Nobody Reads Until It’s Too Late

    The gotcha in the opening story lives entirely in this table. Table type determines the retention ceiling independent of edition, and it determines whether Fail-safe exists at all.

    Table typeTime Travel rangeFail-safe periodNotes
    Permanent0–1 day (Standard) · 0–90 days (Enterprise+)7 days, fixedThe only table type with Fail-safe protection
    Transient0–1 day, on any edition0 daysCapped at 1 day even on Enterprise — cannot be extended
    Temporary0–1 day, session-scoped0 daysDropped automatically when the session ends

    Notice that transient tables don’t just lose Fail-safe — their Time Travel ceiling is capped at one day regardless of what edition you’re on or what the account default says. That’s the entire reason transient tables cost less to store: Snowflake is retaining less history for them, full stop.

    Querying and Restoring Inside the Window

    Time Travel is queryable directly in SQL, three ways: by timestamp, by relative offset, or by the query ID of the statement that changed the data.

    -- Query a table as it existed at a specific timestamp
SELECT * FROM orders
AT (TIMESTAMP => '2026-07-28 09:00:00'::timestamp);

-- Query as it existed immediately before a specific statement ran
SELECT * FROM orders
BEFORE (STATEMENT => '8e5d0c1d-0073-4f57-8263-6e6bb1a2b1d4');

-- Restore an accidentally dropped table, in place
UNDROP TABLE orders_staging;

-- Clone a table's state from 6 hours ago into a new object,
-- useful for diffing without touching the live table
CREATE TABLE orders_audit_clone
CLONE orders AT (OFFSET => -60*60*6);

    All four of those commands only work while the object — or the specific rows you’re targeting — is still inside its Time Travel window. Past that point, UNDROP returns an object-not-found error, not a graceful fallback into Fail-safe. This trips people up constantly: Fail-safe existing doesn’t mean these commands quietly keep working against it. They don’t.

    Fail-safe: What It’s For, and What It Isn’t

    Fail-safe is the part of this system most engineers get wrong, because the name suggests self-service safety and it’s the opposite. It’s a fixed, non-configurable 7-day period that begins the moment an object’s Time Travel retention expires, and it exists for Snowflake’s disaster-recovery purposes — not for routine “oops I dropped a table” moments. You cannot query it, clone from it, or run UNDROP against it. The only path back is a support ticket, and Snowflake is explicit that recovery through Fail-safe can take anywhere from hours to several days, positioning it as a last resort rather than a recovery SLA you can plan around.

    Only permanent tables get Fail-safe. It’s a fixed 7-day buffer after Time Travel expires, and it’s not something you access yourself.

    And critically, Fail-safe only exists for permanent tables. Look back at the comparison table above — transient and temporary objects get 0 days of it. If your team leans on transient tables for staging (a completely reasonable cost optimization, discussed in our zero-copy cloning guide), you’ve implicitly decided that a mistake on those tables gets exactly one day — the Time Travel window — before it’s unrecoverable at any price, support ticket included.

    The Cost Math

    Time Travel and Fail-safe both bill as storage, at your account’s normal per-terabyte rate. As of 2026, Snowflake’s published on-demand list price is around $23 per compressed terabyte per month for AWS US East, with regional variation — worth flagging because a lot of older guides still cite a $40/TB figure that Snowflake has since moved off of.

    The part that surprises people isn’t the rate, it’s the multiplier. Time Travel storage isn’t billed once — it’s billed for the entire retention window, for every version of every changed row, not just until the next write overwrites it.

    Illustrative numbers, not a live account screenshot — but the shape of the math holds: retention window × daily churn rate is the real cost driver, not table size alone.

    A 100 GB table with 10% of its rows modified daily, sitting on a 90-day retention setting, accrues on the order of 10 GB of Time Travel history per day. Over the full window that’s roughly 900 GB — close to a 9x storage multiplier over the table’s own size, from one retention setting on one heavily-churned table. Multiply that across every staging and fact table on a 90-day account default and it stops being a rounding error.

    This is also where the MIN_DATA_RETENTION_TIME_IN_DAYS parameter bites teams that think they’ve already optimized. Someone sets an individual table’s retention to 1 day to cut costs, but the account-level minimum is still 30 — the effective retention is the larger of the two, and the storage bill doesn’t move.

    The Gotchas Nobody Warns You About

    Transient tables have zero Fail-safe, by design, not by oversight. That’s the trade you’re making every time you choose transient for cost savings. It’s a good trade for genuinely disposable staging data. It’s a bad surprise for anything that turns out to matter more than you thought.

    Dropping and recreating a schema resets what its children inherit. If you drop a schema and recreate it with a different DATA_RETENTION_TIME_IN_DAYS, tables created afterward inherit the new value — but objects dropped under the old setting keep whatever retention was active at the time they were dropped, not the new one. It’s easy to assume a schema-level change is retroactive. It isn’t.

    The account-level minimum silently overrides a lower table-level setting. As covered above — if you’re trying to cut Time Travel storage costs by lowering retention on specific tables and the number on your bill doesn’t move, check MIN_DATA_RETENTION_TIME_IN_DAYS before assuming the change didn’t take.

    Cloning at a past timestamp quietly locks in stale data. CREATE TABLE ... CLONE x AT (...) is a zero-copy operation that materializes as a real object pointing at that historical state. It’s easy to leave one of these lying around after an investigation and forget it’s not tracking the live table anymore.

    Fail-safe recovery is not a routine operation, and Snowflake treats it that way. There’s no dashboard, no self-service button, and no fixed turnaround time — it’s a support ticket that gets prioritized as the disaster-recovery mechanism it was designed to be, not an extension of your undo history.

    The One Principle

    Time Travel is a tool you use; Fail-safe is a safety net Snowflake uses on your behalf — design your retention and table types as if Fail-safe doesn’t exist, because for anything outside a permanent table, it doesn’t.

    Related reading: Snowflake Time Travel architecture, deep dive · zero-copy cloning for storage and CI/CD · how Snowflake stores data internally · Snowflake docs: Understanding and using Time Travel · Snowflake docs: Understanding storage cost

  • Running Ollama Inside a Data Pipeline: What Actually Breaks

    Running Ollama Inside a Data Pipeline: What Actually Breaks

    Three weeks ago I watched a teammate open the OpenAI billing dashboard and go quiet. We’d built a PII-tagging job that ran a small classification prompt against every new row landing in a raw events table, about 500,000 rows a day, to flag anything that looked like an email, a phone number, or a government ID before it hit a shared schema. It worked. It also cost $340 a day once we accounted for retries, and legal wanted to know exactly which vendor now had a standing copy of our customer data. Neither number was going to survive the next budget review.

    The fix wasn’t a smarter prompt or a cheaper API tier. It was moving the model onto the same box that already ran the pipeline. Ollama had been sitting in my “toys, not tools” mental bucket for a year, something for chatting with a local Llama build on a Saturday. Turns out it’s a perfectly serviceable inference server for exactly this kind of narrow, high-volume, structured-output task, and it doesn’t send a single row anywhere.

    Flowchart showing PII Tagging Job: Airflow sends batches to Ollama API and a quantized model, outputs structured JSON, stored in Snowflake Stage, and merged into a Snowflake Target Table.

    The whole job runs on infrastructure you already control. Only the base URL in your HTTP client changes if you ever move to a hosted model.

    TL;DR

    • → Ollama exposes an OpenAI-compatible API on localhost:11434, so swapping a cloud LLM call for a local one in an Airflow task is usually a one-line change to the client’s base URL.
    • → Small quantized models (3B–8B parameters at 4-bit) handle narrow, structured tasks like PII tagging, log classification, or doc-string generation well; they are not a drop-in replacement for a frontier model on open-ended reasoning.
    • → A 4-bit quantized 3B model needs roughly 2–3 GB of memory instead of the 6+ GB full precision would require, which is why it fits on a shared pipeline host without a dedicated GPU budget.
    • → For a 500K-row daily classification job, local inference on existing hardware runs at effectively $0 marginal cost per run, versus real per-token cloud spend that scales with volume.
    • → Ollama is a single background daemon, not a per-task process, so the first request in an Airflow DAG can be slow while the model loads into memory, and this needs its own timeout handling.
    • → Running the model locally also means the compliance conversation changes: no row of customer data leaves the host you already control access to.

    Why This Belongs in the Pipeline, Not Just the Terminal

    Most Ollama content is written for a single, interactive session: install it, pull a model, chat with it, done. That’s a fine on-ramp, but it undersells what the tool is actually good for once you strip away the chat interface. Underneath the terminal experience is a plain HTTP server. It handles model loading, memory management, and hardware acceleration, and it exposes a REST API that speaks the same shape as OpenAI’s chat completions endpoint. That last part matters more than the local-vs-cloud framing usually gives it credit for: if your pipeline code already calls an LLM through an OpenAI-compatible client, pointing it at Ollama is a base-URL change, not a rewrite.

    That’s the same reasoning that made Snowflake’s own Cortex tooling worth covering here: the interesting engineering question is never “can the model do the task,” it’s “what does it cost to wire this into infrastructure we already run.” Ollama’s answer is: not much, provided the task is narrow enough for a small model to handle reliably.

    Installing Ollama on a Pipeline Host, Not a Laptop

    The install itself is unremarkable, which is the point. On the Linux boxes running our Airflow workers, it’s a single script:

    curl -fsSL https://ollama.com/install.sh | sh

# confirm the daemon is up and check the version
ollama --version
systemctl status ollama

    Ollama installs itself as a systemd service on Linux, listening on 127.0.0.1:11434 by default. If your Airflow workers and the model need to live on separate hosts, you’ll want to bind it to the internal network interface instead and lock that down with a security group, not expose it publicly:

    sudo systemctl edit ollama
# add under [Service]:
# Environment="OLLAMA_HOST=0.0.0.0:11434"
sudo systemctl restart ollama

    Picking a Model That Fits the Task, Not the Demo

    For structured tagging work, you don’t want the biggest model that fits on the box. You want the smallest one that hits your accuracy bar, because latency and memory headroom compound across half a million rows. We landed on a 3B-class model after testing three sizes against a hand-labeled validation set:

    ollama pull llama3.2
ollama pull qwen3:8b
ollama pull gemma4:e4b

    By default Ollama pulls a 4-bit quantized build (q4_K_M), which is why a 3B model downloads at roughly 2 GB instead of the 6 GB a full-precision (fp16) version would need. That quantization step compresses the model’s weights into 4-bit integers, and for a classification task with a fixed, narrow label set, the accuracy hit is negligible. It would be a different conversation for open-ended generation.

    Wiring It Into an Airflow DAG

    The integration point is a plain HTTP call inside a PythonOperator, using the OpenAI-compatible endpoint Ollama exposes at /v1/chat/completions:

    from openai import OpenAI
from airflow.decorators import task

client = OpenAI(base_url="http://localhost:11434/v1", api_key="not-needed")

@task
def tag_pii_batch(rows: list[dict]) -> list[dict]:
    results = []
    for row in rows:
        response = client.chat.completions.create(
            model="llama3.2",
            messages=[
                {"role": "system", "content": (
                    "Classify the input for PII. Respond with strict JSON: "
                    '{"has_pii": bool, "categories": [string]}'
                )},
                {"role": "user", "content": row["raw_text"]},
            ],
            temperature=0,
        )
        results.append({"row_id": row["id"], "tags": response.choices[0].message.content})
    return results

    Run output for a single row looks like this once it comes back through the DAG’s logging:

    {"row_id": "evt_88213", "tags": "{\"has_pii\": true, \"categories\": [\"email\", \"phone\"]}"}

    The output then lands in a staging table and gets merged into the target with a standard MERGE statement, the same pattern we used when writing about Time Travel for auditing exactly which run tagged which row.

    The Cost Math

    Here’s the comparison that actually mattered to our budget review. Assume 500,000 rows a day, roughly 120 input tokens and 40 output tokens per row for a classification prompt like the one above.

    ApproachDaily token volumeMarginal cost / dayData leaves the host?
    Cloud API, mid-tier model~80M tokens~$310–$360, scales with volumeYes
    Ollama, local 3B model, existing pipeline host~80M tokens~$0 marginal (existing hardware)No

    The local path isn’t free in an absolute sense, you’re spending CPU/GPU cycles you already own, and if the host is undersized you’ll eventually pay for a bigger instance. But that’s a fixed, predictable infrastructure cost instead of a bill that grows linearly with data volume, which is the same argument we made when comparing orchestrator licensing models: usage-based pricing is fine until usage is the thing you’re trying to grow.

    The Gotchas Nobody Warns You About

    The first request in a DAG run pays the cold-load tax. Ollama unloads a model from memory after a period of inactivity (five minutes by default). If your DAG runs hourly, every run’s first task can eat several extra seconds waiting for the model to load back in. Set keep_alive in the request payload to a longer duration, or ping the model with a lightweight warm-up call before the real batch starts.

    Concurrent Airflow tasks will silently queue, not fail. Ollama serializes requests to a given model by default unless you’ve explicitly configured parallel request handling. If you fan out ten parallel tasks expecting ten times the throughput, you’ll instead get one queue that’s ten times longer, with no error to tell you why it’s slow.

    Schema-constrained output still needs a parser, not blind trust. Even with an explicit “respond with strict JSON” instruction, small models occasionally wrap their answer in a stray sentence or a markdown code fence. Wrap every response in a real JSON parse with a fallback path, the same defensive habit you’d apply to any external API, local or not.

    A shared host means a shared blast radius. If the same box also runs other Airflow tasks, a model pinned in GPU memory can starve them of resources in ways that look like an unrelated flaky DAG. Give the inference workload its own resource ceiling, whether that’s a cgroup limit or a dedicated worker pool.

    Version drift is real and mostly undocumented per-task. Ollama ships frequent releases, and quantization defaults or model behavior can shift between versions. Pin the Ollama version and the exact model tag (not just latest) in whatever config or container image your pipeline deploys, the same way you’d pin a Python package version.

    The One Principle

    A local model is an infrastructure decision before it’s an AI decision — treat it like the database or the queue it’s sitting next to, with monitoring, resource limits, and version pins, and the “AI” part of the problem turns out to be the easy half.

    That framing is also why this pattern generalizes past PII tagging. The same DAG shape works for classifying malformed records before they hit a Delta Lake or Iceberg table, generating draft column descriptions during a schema change, or triaging alert text before it pages someone. None of those need a frontier model. They need a small one that’s fast, cheap, and doesn’t leave the building, which is precisely the gap Ollama fills once you stop thinking of it as a chat toy and start thinking of it as another service in the stack, not unlike how we’ve written about the tradeoffs inside Snowflake’s warehouse cache or the workflow shifts in Cortex Code Desktop.

    Related reading: Snowflake Cortex Code Desktop · Airflow vs. Prefect · Snowflake Time Travel · Ollama · Ollama OpenAI-Compatibility Docs

  • Dynamic Airflow DAGs via Snowflake Metadata: Eliminating Hardcoded Pipeline Tasks

    Dynamic Airflow DAGs via Snowflake Metadata: Eliminating Hardcoded Pipeline Tasks

    I once inherited an Airflow repo with 214 DAG files that were, functionally, the same DAG. Each one extracted a table from a source system, loaded it into Snowflake, and ran a transform. The only differences between them were the table name, the schedule, and which SQL file to run. Someone had copy-pasted the template 214 times, and every schema change meant a find-and-replace across 214 files and a prayer that nothing got missed. Onboarding a new table meant copy-pasting a 215th.

    That repo is the case against hardcoded pipeline tasks in a single painful sentence: if the only thing that changes between your DAGs is data, then your DAGs should be generated from data. The fix is to move the pipeline definitions out of Python files and into a Snowflake metadata table, then generate the DAGs from that table. Add a row, get a pipeline. Change a row, change a pipeline. No copy-paste, no 214-file find-and-replace.

    This is the guide to doing that properly — including the distinction that trips most people up (there are two completely different “dynamic” features in Airflow and they solve different problems), the metadata-driven generation pattern, and the parsing gotchas that will wreck your scheduler if you get them wrong.

    TL;DR

    → Two different features share the word “dynamic.” Dynamic DAG generation builds DAG structure at parse time from config/metadata — the task count is fixed for a given run. Dynamic task mapping (.expand()) creates N task instances at runtime from an upstream task’s output. They solve different problems; you’ll often use both.

    → The metadata-driven pattern: store pipeline definitions (DAG name, tasks, schedule, SQL file, parent/child dependencies) in a Snowflake table → feed each row into a Jinja template → render a dag.py file per pipeline. Add a row, get a DAG.

    → Add operational columns to the metadata table — created_at and last_updated_at — so you can track which pipelines exist and trigger regeneration when a definition changes.

    → Use environment variables, not Airflow Variables, in top-level DAG code. Airflow Variables hit the metadata DB on every parse and will slow your scheduler to a crawl.

    → Generate tasks in a stable, sorted order every time (ORDER BY in your query or sorted() in Python), or the Grid View reshuffles tasks on every refresh and your history becomes unreadable.

    → For large numbers of generated DAGs, use get_parsing_context() to skip building DAG objects you don’t need during task execution — one documented case cut parsing from 120s to 200ms.

    → Use dynamic task mapping when the count is unknown until runtime (e.g. “process however many files landed today”). Note trigger_rule=ALWAYS is not allowed on task-generated mapped tasks.

    The distinction that trips everyone up

    Before any code, get this straight, because conflating the two is the single most common source of confusion I see. Airflow has two features with “dynamic” in the name and they are not interchangeable.

    Dynamic DAG generation is about producing DAG files or objects programmatically. Instead of hand-writing 214 near-identical DAGs, you write one generator that reads definitions from somewhere (a config file, a metadata table) and emits the DAGs. The important property: the structure is decided at parse time, when Airflow loads the DAG file. For a given DAG run, the number of tasks is fixed. This is what you want when you have many similar pipelines that differ only by parameters.

    Dynamic task mapping (introduced in Airflow 2.3, via .expand() and .map()) is about creating task instances at runtime. A task returns a list, and Airflow creates one copy of a downstream task per element — and it doesn’t know how many until the upstream task actually runs. This is the MapReduce model: the scheduler creates N copies of the mapped task right before execution. This is what you want when the count is genuinely unknown until runtime — “process each file that landed in S3 today,” where “today” might be 3 files or 300.

    The rule of thumb: if you know the shape of the work when the DAG is parsed, use dynamic DAG generation. If the shape depends on data that only exists at runtime, use dynamic task mapping. A mature setup often uses both — generated DAGs whose internal tasks map over runtime data.

    Left: fixed structure known at parse time. Right: task instances fanned out at runtime. Same word, opposite problems.

    The metadata-driven pattern

    The pipeline that builds pipelines: a Snowflake metadata table feeds a Jinja template that renders one dag.py per row, which Airflow then parses like any other DAG.

    The architecture has four moving parts. First, a metadata table in Snowflake that holds pipeline definitions. At minimum it stores, per task: the DAG name it belongs to, the task name, the schedule, what the task runs (say, a SQL file path), and the parent/child dependency links. A row-per-task layout with a parent_task column lets you express arbitrary dependency graphs — a task names its parent, and the generator wires the edges.

    Here’s a minimal shape:

    CREATE TABLE pipeline_metadata (
  dag_name      STRING,
  task_name     STRING,
  parent_task   STRING,  -- NULL for a root task
  schedule      STRING,  -- e.g. '0 2 * * *'
  sql_file      STRING,  -- what the task executes
  is_active     BOOLEAN,
  created_at    TIMESTAMP,
  last_updated_at TIMESTAMP
);

    Second, a Jinja template — a .j2 file that looks like a DAG with placeholders where the metadata values go: the DAG id, the schedule, a loop that emits one operator per task, and the dependency wiring. Third, a generator that queries the metadata table, groups rows by dag_name, and renders the template once per DAG, writing out a dag.py file. Fourth, Airflow’s normal DAG File Processor, which parses those rendered files exactly as if you’d hand-written them.

    The payoff is the operational columns. Because each row carries created_at and last_updated_at, you can tell when a pipeline was first defined and when it last changed. When someone edits a definition, last_updated_at moves, and you can trigger regeneration for just the affected DAGs rather than rebuilding everything. Onboarding a new pipeline is now an INSERT, not a new file.

    Rendering the DAG from a row

    The generator itself is short. Conceptually: query the active metadata, group by DAG, and for each group render the template with that group’s tasks and dependencies. A sketch:

    from jinja2 import Environment, FileSystemLoader
import os

# env var, NOT an Airflow Variable — see the parsing note below
env = os.environ.get("DEPLOYMENT", "PROD")

rows = run_query("""
  SELECT dag_name, task_name, parent_task, schedule, sql_file
  FROM pipeline_metadata
  WHERE is_active = TRUE
  ORDER BY dag_name, task_name  -- stable order, always
""")

template = Environment(loader=FileSystemLoader("templates")) \
    .get_template("dag_template.j2")

for dag_name, tasks in group_by_dag(rows):
  rendered = template.render(dag_name=dag_name, tasks=tasks, env=env)
  with open(f"dags/{dag_name}.py", "w") as f:
    f.write(rendered)

    Notice the ORDER BY. That is not cosmetic — it’s load-bearing, and the next section explains why.

    The parsing gotchas that wreck schedulers

    Three parse-time mistakes and their fixes. Every one of these is invisible until your scheduler is under load, then very visible.

    Dynamic generation runs at parse time, and the DAG File Processor parses your files constantly. Anything expensive or unstable in that path multiplies across every parse. Three specific mistakes:

    Airflow Variables in top-level code. It’s tempting to configure your generator with Variable.get("something"). Don’t, not at the top level. Every Airflow Variable read in top-level code opens a connection to the metadata database, and top-level code runs on every parse. At scale this hammers your metadata DB and drags parsing. Use environment variables (os.environ.get(...)) for anything read during generation — they’re free to read and don’t touch the DB.

    Unstable task ordering. If your generator emits tasks in a different order on different parses — because the query has no ORDER BY, or you iterated a Python set — Airflow’s Grid View reshuffles the task rows every time it refreshes. Your run history becomes impossible to read, and it looks like the DAG is changing when it isn’t. Always impose a stable order: ORDER BY in the query, or sorted() in Python. Deterministic generation is not optional.

    Parsing every DAG on every task execution. The DAG File Processor loads the whole file to get metadata, but executing a single task only needs that one DAG object. If your generator builds hundreds of DAGs in one file, every task execution pays to construct all of them. The fix is get_parsing_context(): check which DAG is actually being parsed and skip generating the rest. The documented “Magic Loop” example cut parsing from 120 seconds to 200 milliseconds this way. It’s most valuable when the generated-DAG count is high — use it with care and test it, since it doesn’t apply if later DAGs depend on earlier ones.

    When to reach for dynamic task mapping instead

    Everything above generates structure from metadata known at parse time. But some workloads only reveal their shape at runtime, and that’s dynamic task mapping’s job. The canonical example is file processing: an unknown number of files land in cloud storage each day, and you want one task instance per file loaded into Snowflake.

    The pattern is a task that returns the list, and a downstream task that expands over it:

    @task
def list_new_files():
  return get_s3_keys(prefix=f"{{{{ ds_nodash }}}}/")  # however many landed

@task
def load_to_snowflake(key):
  copy_into_snowflake(key)

load_to_snowflake.expand(key=list_new_files())

    The scheduler creates one load_to_snowflake instance per key, right before execution, and the Grid View shows the mapped count in brackets. You can also map over task groups with the @task_group decorator and .expand() when each unit of work is several steps, using the map_indexes parameter to pull the right XCom per instance. One constraint to remember: trigger_rule=TriggerRule.ALWAYS is not allowed on a task-generated mapped task, because the expanded parameters are undefined at the moment of immediate execution — Airflow raises an error at parse time if you try.

    Cost and maintenance math

    The win here isn’t compute cost, it’s maintenance cost, and it compounds. Go back to the 214-DAG repo. A schema change that touched every pipeline meant editing 214 files — call it a day of careful, error-prone work, plus review, plus the near-certainty of missing one. With metadata-driven generation, the same change is either one UPDATE to the metadata table or one edit to the shared Jinja template, followed by regeneration. Minutes, not a day, and uniform by construction — you cannot miss one, because there’s only one definition.

    Onboarding scales the same way. In the file-per-pipeline world, each new table is a new hand-authored file and a new opportunity for drift. In the metadata world it’s an INSERT. Ten new tables is ten rows. The marginal cost of a pipeline drops toward zero, which changes what’s worth automating — pipelines that weren’t worth hand-writing become trivially worth a row.

    The gotchas nobody warns you about

    Generation failure takes down everything at once. The flip side of one definition is one point of failure. A bug in the template or generator doesn’t break one DAG, it breaks all of them. Validate rendered output (even a quick python -c "compile(...)" check) before writing files, and keep the last-good rendered files so a bad generation doesn’t wipe working DAGs.

    Metadata and reality drift. The metadata table says what pipelines should exist; the dags/ folder holds what does. If someone edits a rendered file by hand, or a row is deleted without removing the file, the two diverge. Treat rendered files as build artifacts, never edit them directly, and have regeneration remove files for DAGs no longer in the metadata.

    Secrets don’t belong in the metadata table. It’s tempting to store connection details per pipeline. Keep credentials in Airflow Connections or a secrets backend and reference them by name from the metadata — the table should hold pipeline structure, not secrets.

    Too much magic hurts debuggability. A generated DAG is one level removed from the code you read. When something breaks at 2 a.m., the on-call engineer is debugging rendered output, not the template. Keep the template readable, keep the rendered files on disk (don’t generate purely in memory), and make it obvious which metadata row produced which DAG.

    The one principle

    If the only thing that changes between your pipelines is data, define them with data, not code. Put pipeline definitions in a Snowflake metadata table, render them through one Jinja template, and let Airflow parse the result — but keep generation deterministic, keep Airflow Variables out of top-level code, and remember that one definition means one point of failure worth guarding. The goal isn’t cleverness; it’s that onboarding the 215th pipeline should be an INSERT, and a schema change should touch one place, not two hundred.

    Related reading: Airflow: Dynamic DAG Generation (official docs) · Airflow: Dynamic Task Mapping (official docs) · Orchestrating dbt With Airflow on Snowflake · dbt State on Snowflake: Skip Unchanged Models · Snowflake Query Execution: What Really Happens

  • Someone renamed a column.Your Pipeline Died.Here’s the fix

    Someone renamed a column.Your Pipeline Died.Here’s the fix

    Someone on the backend team renamed order_total to order_amount. Clean name. Makes total sense for their domain model. They shipped it on a Thursday afternoon. By Friday morning, your revenue dashboard was showing zero. Not wrong numbers. Zero. Because your Snowflake pipeline was still selecting order_total from the events table, and the column simply wasn’t there anymore.

    You found out from a Slack message. From a director. At 9 AM.

    This is the most common production incident in data engineering in 2026, and it’s almost never caused by bad code. It’s caused by the absence of a formal agreement between the team producing data and the team consuming it. That agreement has a name: a data contract. And most data teams still don’t have one.

    The excuse is usually some version of “we move too fast.” The reality is that the teams who move fastest are the ones with contracts, because they stop discovering breaking changes from directors on Friday mornings and start catching them in CI on Thursday afternoons, before anything ships.

    TL;DR

    → A data contract is a formal specification — schema, semantics, SLAs, ownership — between a data producer and its consumers. Not documentation. Enforcement.

    → Most data incidents don’t start with missing data or broken code. They start with a well-intentioned upstream change that silently invalidated an assumption someone downstream was relying on.

    → Contracts have three parts: schema (structure and types), semantics (what fields actually mean), and SLAs (freshness, completeness, availability). Schema-only contracts miss most real breakages.

    → The dual-write pattern is the only safe migration path for breaking changes: keep old field + add new field → both populated during transition → deprecation notice with a hard date → removal at v2. Each phase takes at minimum 30 days. Skipping phases causes incidents.

    → 90 days minimum notice for breaking changes. Data pipelines have long release cycles; consumers need time to update downstream logic, tests, and dashboards.

    → A contract not enforced in CI is just documentation. The ODCS (Open Data Contract Standard) YAML spec plus `datacontract-cli` gives you executable, version-controlled contracts in about 30 minutes per dataset.

    → dbt integration: map contract checks to dbt tests. Require a version bump plus consumer sign-off on breaking changes before merge. After one month of this, most teams report significantly fewer schema surprises.

    → The worst gotcha: contracts that only cover schema, not semantics. A field that changes meaning without changing type is undetectable to automated checks — and it’s how revenue figures silently drift for weeks.

    Why schemas break and who owns the blame

    Schema evolution sits between two teams that don’t talk to each other on the same cadence. The producer team — usually a backend or platform engineering team — is shipping product features, often weekly, and treats every field they emit as their own. The consumer team — your data engineering team — is running pipelines that depend on those fields staying stable, and finds out about breaking changes the same way archaeologists find ruins: by digging through wreckage.

    The producer isn’t wrong for evolving their schema. The consumer isn’t wrong for depending on it. The incident happens because there was no shared definition of what “a safe change” means, no process for communicating it, and no tooling to enforce the agreement. The blame falls on the process, not the person. Which means the fix is a process change, not a person change.

    Schema evolution is the load-bearing problem in data engineering in 2026, and it’s the problem most teams handle the worst. The good teams treat upstream schemas as contracts and run checks against those contracts on every pipeline run. The teams that lose stakeholder trust treat upstream schemas as suggestions and find out about every breaking change from a Slack message that starts “hey, the dashboard looks weird.”

    That Slack message is always sent on a Friday. It is always sent to a director.

    What a data contract actually contains

    The mistake most teams make when they start with data contracts is writing schema-only contracts. Field names, data types, nullability. It feels rigorous. It catches a specific class of errors — column removed, type changed — but misses most real incidents.

    Real breakages happen at the semantics layer. The producer changes order_total from gross to net revenue. Same field name. Same FLOAT type. No schema violation. But your revenue dashboard is now off by 23%, silently, because the number means something different than it did last week. A schema validator cannot catch this. Only a semantic contract can — one that documents what a field means, how it should be used, and what constitutes a valid business interpretation of its values.

    A complete data contract has three layers. Schema: field names, data types, nullability, constraints (no negative values in a price field, for example). Semantics: what each field means in business terms, how it maps to domain concepts, what transformations are applied before it reaches the consumer. SLAs: freshness guarantees (this dataset is refreshed within 15 minutes of source update), completeness thresholds (at least 99.5% of expected rows must be present), availability targets, and a named owner with actual contact information — not “data team.”

    The Open Data Contract Standard and the YAML spec

    The good news for teams starting in 2026 is that there’s a growing standard: ODCS (Open Data Contract Standard), a YAML-based specification that defines schema, quality rules, SLAs, and ownership in a single document. It’s human-readable, version-controllable in git, and machine-parseable by tools like `datacontract-cli`, which can validate contracts, run compatibility checks, and generate reports.

    A minimal ODCS contract for an orders dataset looks like:

    dataContractSpecification: 0.9.3
id: orders-v1
info:
title: Orders
version: 1.0.0
owner: [email protected]
servers:
production:
type: snowflake
database: PROD_DB
schema: PUBLIC
table: orders
models:
orders:
fields:
order_id:
type: string
required: true
description: Unique identifier for the order
order_amount:
type: number
required: true
description: Net revenue after discounts and returns, in USD
minimum: 0
created_at:
type: timestamp
required: true
servicelevels:
freshness:
description: Data refreshed within 15 minutes of source update
threshold: PT15M
completeness:
description: At least 99.5% of expected rows present
threshold: "99.5%"

    This is not documentation theater. This YAML file is executable. `datacontract-cli test` validates your actual Snowflake table against this contract. It checks types, required fields, minimum values, and can be wired into CI so that any schema change that would violate the contract fails the PR before it merges.

    The only safe migration path for breaking changes

    When a producer needs to make a breaking change — remove a field, rename it, change its type, change its semantics — the contract provides a coordination mechanism. There’s a specific pattern that works, and teams that skip steps in it pay for it.

    Day 0: Announce. The producer creates a deprecation notice in the contract YAML, updates the changelog, and notifies consumers via a designated channel. Critically, this notification includes a hard date for removal — not “eventually” or “when everyone has migrated.” Deprecated without a date is just a polite rumor. A field can sit in limbo for eighteen months while producers assume nobody uses it and consumers assume it will live forever.

    Days 0–60: Dual-write. The producer populates both the old field and the new field simultaneously. Consumers can migrate on their own schedule during this window. The producer monitors usage of the old field (this is easy with Snowflake’s QUERY_HISTORY and column-level access tracking) to know when all consumers have switched.

    Day 60: Deprecation notice with hard date. Consumers who haven’t migrated get a 30-day final warning. This is the reminder that actually motivates stragglers. The hard date is non-negotiable.

    Day 90+: Removal at v2. The old field is gone. The contract version bumps to 2.0.0. This is a semantic major version — it breaks backward compatibility — and that bump is what triggers automated alerts to any consumer still on v1.

    No drama. No guessing. No 2 AM rollback. Give consumers at least 90 days notice for breaking changes. This seems long, but data pipelines have long release cycles, and consumers need time to update downstream logic, tests, and dashboards.

    Making it executable: CI enforcement that actually works

    The critical architectural decision with data contracts is this: a contract not enforced in CI is just documentation, and documentation drifts. Within six months, the contract YAML and the actual schema diverge, nobody updates the contract when they ship features, and you’re back to tribal knowledge with extra steps.

    The enforcement pattern that works:

    1. Compatibility check on PR. Before any schema change merges, run `datacontract-cli diff` against the current production contract. Breaking changes fail the PR automatically. Non-breaking changes (adding a nullable field, loosening a constraint) pass. The definition of “breaking” is explicit in the contract spec, not up to whoever reviews the PR.

    2. Consumer sign-off for breaking changes. If a breaking change is intentional (the producer knows and has planned for it), the PR requires explicit approval from all registered consumers of that dataset. This is enforced via GitHub CODEOWNERS or equivalent. Producers can’t ship breaking changes unilaterally.

    3. dbt test integration. Map contract quality rules to dbt tests. Freshness SLAs become `dbt source freshness` checks. Completeness thresholds become row count assertions. Not-null requirements become `not_null` tests. These run on every dbt build, so violations are caught before models complete — not after reports are wrong.

    4. Runtime validation at ingestion. Before data loads into your Silver or Gold layers, validate incoming records against the contract. Rows that violate constraints get quarantined in a dead-letter queue, not silently loaded as nulls. This catches semantic drift that schema validation misses: an order_amount field that’s suddenly returning negative values because someone upstream changed the sign convention.

    The gotchas that sink most implementations

    Exposing raw transactional schemas as data products. This is the most common structural mistake. When your data contract directly mirrors your application’s OLTP schema, every application refactor becomes a consumer’s problem. The fix is a stable abstraction layer — expose only what consumers need, not the underlying operational detail. Schema changes to the application layer should be absorbed by your ingestion layer, not propagated downstream.

    Brittle contracts that break more than they prevent. Strict attribute lengths, tightly constrained enums, or hyper-specific format requirements feel like good quality controls. In practice, they make schemas so rigid that producers constantly need change approvals for minor operational updates that have no downstream impact. Design contracts around semantic guarantees and business invariants, not implementation details. amount > 0 is a semantic guarantee. DECIMAL(18,4) is an implementation detail that will change.

    Unclear ownership is the silent killer. Data contracts fail most often not because of tooling gaps, but because accountability is unclear. When something breaks, teams scramble to diagnose issues that fall between ownership boundaries. Every contract needs a named owner with actual incident-response obligations. Not a team. Not a Slack channel. A person whose name is in the contract and who gets paged when a contract violation is detected at runtime.

    Semantic changes that look like no-ops. Changing what a field means without changing its name, type, or schema is the hardest class of breakage to catch. order_amount switching from gross to net. A user_id changing from internal to external identifiers. These require semantic versioning (a major version bump) and human review, not just automated compatibility checks. Your CI can catch structural breakage; only your team can catch semantic breakage.

    Contracts that cover batch but ignore streaming. If you have a Kafka-based event pipeline feeding your Snowflake tables, the schema contract lives in the Kafka topic, not in the table. Changes to the Kafka Avro schema — registered in Confluent Schema Registry or AWS Glue — need the same versioning and deprecation discipline as your warehouse schemas. Most teams only contract the warehouse side and get burned by streaming schema changes that propagate silently into their pipeline.

    The real cost math

    Data engineering incidents from schema breakage are expensive in ways that don’t show up on warehouse bills. A typical schema incident at a mid-sized company looks like: 3–4 hours of two engineers debugging, 1 hour of a data analyst investigating wrong numbers, a director review, and a post-mortem. Call that 10 person-hours, at a blended rate of $150/hour. That’s $1,500 per incident.

    Teams that experience two schema incidents a month — which is conservative for a team without contracts — are burning $3,000/month, or $36,000/year, on incidents alone. That doesn’t count the cost of wrong decisions made from bad data before the incident was even discovered. One revenue calculation running off a silent semantic change for three weeks is often worth more than a year of incident cost.

    The tooling investment for data contracts — `datacontract-cli`, ODCS YAML per dataset, CI integration — is a few days of engineering time. The 90-day discipline is a process change, not a tooling cost. The math is not close.

    Where to start (not where everyone starts)

    Everyone says “start with your most critical datasets.” That’s correct but useless. More specifically: identify the three datasets that caused production incidents in the last 90 days. Start with those. Not your biggest datasets. Not your most complex. The ones that already broke something.

    For each: write the ODCS YAML (schema + semantics + SLAs + owner). Add `datacontract-cli` compatibility checks to the PR workflow for that dataset. Map the quality rules to dbt tests. That’s the first sprint. After one month of this on three datasets, you’ll have a template, a workflow, and enough muscle memory to expand to the rest of the catalog without it feeling like a governance initiative nobody asked for.

    The one principle

    Change is inevitable. Unmanaged change is expensive. A data contract is the agreement that makes change boring instead of dangerous. The goal isn’t to prevent schemas from evolving — schemas should evolve as the business evolves. The goal is to make every evolution visible, deliberate, and announced far enough in advance that nobody finds out about it from a director on a Friday morning.

    Related reading: Open Data Contract Standard (ODCS) · datacontract-cli on GitHub · dbt State: Skip Unchanged Nodes, Cut Runtime by 60% · dbt Fusion: 30x Faster Parsing · Snowflake Query Execution: What Really Happens

  • Mastering Real-Time ETL with Google Cloud Dataflow: A Comprehensive Tutorial

    Mastering Real-Time ETL with Google Cloud Dataflow: A Comprehensive Tutorial

    In the fast-paced world of data engineering, mastering real-time ETL with Google Cloud Dataflow is a game-changer for businesses needing instant insights. Extract, Transform, Load (ETL) processes are evolving from batch to real-time, and Google Cloud Dataflow stands out as a powerful, serverless solution for building streaming data pipelines. This tutorial dives into how Dataflow enables efficient, scalable data processing, its integration with other Google Cloud Platform (GCP) services, and practical steps to get started in 2025.

    Whether you’re processing live IoT data, monitoring user activity, or analyzing financial transactions, Dataflow’s ability to handle real-time streams makes it a top choice. Let’s explore its benefits, setup process, and a hands-on example to help you master real-time ETL with Google Cloud Dataflow.

    Why Choose Google Cloud Dataflow for Real-Time ETL?

    Google Cloud Dataflow offers a unified platform for batch and streaming data processing, powered by the Apache Beam SDK. Its serverless nature eliminates the need to manage infrastructure, allowing you to focus on pipeline logic.

    Hand-drawn illustration depicting the serverless architecture of Google Cloud Dataflow for efficient real-time ETL processing.

    Key benefits include:

    • Serverless Architecture: Automatically scales resources based on workload, reducing operational overhead and costs.
    • Seamless GCP Integration: Works effortlessly with BigQuery, Pub/Sub, Cloud Storage, and Data Studio, creating an end-to-end data ecosystem.
    • Real-Time Processing: Handles continuous data streams with low latency, ideal for time-sensitive applications.
    • Flexibility: Supports multiple languages (Java, Python) and custom transformations via Apache Beam.

    For businesses in 2025, where real-time analytics drive decisions, Dataflow’s ability to process millions of events per second positions it as a leader in cloud-based ETL solutions.

    Setting Up Google Cloud Dataflow

    Before building pipelines, set up your GCP environment:

    1. Create a GCP Project: Go to the Google Cloud Console and create a new project.
    2. Enable Dataflow API: Navigate to APIs & Services > Library, search for “Dataflow API,” and enable it.
    3. Install SDK: Use the Cloud SDK or install the Apache Beam SDK:
    pip install apache-beam[gcp]

    4. Authenticate: Run gcloud auth login and set your project with gcloud config set project PROJECT_ID.

    This setup ensures you’re ready to deploy and manage real-time ETL with Google Cloud Dataflow pipelines.

    Building a Real-Time Streaming Pipeline

    Let’s create a simple pipeline to process real-time data from Google Cloud Pub/Sub, transform it, and load it into BigQuery. This example streams simulated sensor data and calculates average values.

    Hand-drawn diagram of a real-time ETL pipeline using Google Cloud Dataflow, from Pub/Sub to BigQuery
    Step-by-Step Code Example
    import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
import json

class DataflowOptions(PipelineOptions):
    @classmethod
    def _add_argparse_args(cls, parser):
        parser.add_argument('--input_subscription', default='projects/your-project/subscriptions/your-subscription')
        parser.add_argument('--output_table', default='your-project:dataset.table')

def run():
    options = DataflowOptions()
    with beam.Pipeline(options=options) as p:
        # Read from Pub/Sub
        data = (p
                | 'Read from Pub/Sub' >> beam.io.ReadFromPubSub(subscription=options.input_subscription)
                | 'Decode JSON' >> beam.Map(lambda x: json.loads(x.decode('utf-8')))
                )

        # Transform: Calculate average sensor value
        averages = (data
                    | 'Group by Sensor' >> beam.GroupByKey()
                    | 'Compute Average' >> beam.MapTuple(lambda k, v: (k, sum(v) / len(v) if v else 0))
                    )

        # Write to BigQuery
        averages | 'Write to BigQuery' >> beam.io.WriteToBigQuery(
            options.output_table,
            schema='sensor_id:STRING,average_value:FLOAT',
            write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND,
            create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED
        )

if __name__ == '__main__':
    run()
    How It Works
    • Input: Subscribes to a Pub/Sub topic streaming JSON data (e.g., {“sensor_id”: “S1”, “value”: 25.5}).
    • Transform: Groups data by sensor ID and computes the running average.
    • Output: Loads results into a BigQuery table for real-time analysis.

    Run this pipeline with:

    python your_script.py --project=your-project --job_name=real-time-etl --runner=DataflowRunner --region=us-central1 --setup_file=./setup.py

    This example showcases real-time ETL with Google Cloud Dataflow’s power to process and store data instantly.

    Integrating with Other GCP Services

    Dataflow shines with its ecosystem integration:

    Hand-drawn overview of Google Cloud Dataflow's integrations with GCP services like Pub/Sub and BigQuery for real-time ETL
    • Pub/Sub: Ideal for ingesting real-time event streams from IoT devices or web applications.
    • Cloud Storage: Use as a staging area for intermediate data or backups.
    • BigQuery: Enables SQL-based analytics on processed data.
    • Data Studio: Visualize results in dashboards for stakeholders.

    For instance, connect Pub/Sub to stream live user clicks, transform them with Dataflow, and visualize trends in Data Studio—all within minutes.

    Best Practices for Real-Time ETL with Dataflow

    • Optimize Resources: Use autoscaling and monitor CPU/memory usage in the Dataflow monitoring UI.
    • Handle Errors: Implement dead-letter queues in Pub/Sub for failed messages.
    • Security: Enable IAM roles and encrypt data with Cloud KMS.
    • Testing: Test pipelines locally with DirectRunner before deploying.

    These practices ensure robust, scalable real-time ETL with Google Cloud Dataflow pipelines.

    Benefits in 2025 and Beyond

    As of October 2025, Dataflow’s serverless model aligns with the growing demand for cost-efficient, scalable solutions. Its integration with AI/ML services like Vertex AI for predictive analytics further enhances its value. Companies leveraging real-time ETL report up to 40% faster decision-making, according to recent industry trends.

    External Resource Links

    For deeper dives and references:

    Conclusion

    Mastering real-time ETL with Google Cloud Dataflow unlocks the potential of streaming data pipelines. Its serverless design, GCP integration, and flexibility make it ideal for modern data challenges. Start with the example above, experiment with your data, and scale as needed.

  • Mastering Python Data Pipelines: Extract from APIs & Databases, Load to S3 & Snowflake

    Mastering Python Data Pipelines: Extract from APIs & Databases, Load to S3 & Snowflake

    Introduction to Data Pipelines in Python

    In today’s data-driven world, creating robust data pipelines solutions is essential for businesses to handle large volumes of information efficiently. Whether you’re pulling data from RESTful APIs or external databases, the goal is to extract, transform, and load (ETL) it reliably. This guide walks you through building data pipelines using Python that fetch data from multiple sources, store it in Amazon S3 for scalable storage, and load it into Snowflake for advanced analytics.

    By leveraging Python’s powerful libraries like requests for APIs, sqlalchemy for databases, boto3 for S3, and the Snowflake connector, you can automate these processes. This approach ensures data integrity, scalability, and cost-effectiveness, making it ideal for data engineers and developers.

    Why Use Python for Data Pipelines?

    Python stands out due to its simplicity, extensive ecosystem, and community support. Key benefits include:

    best practices in data engineering
    • Ease of Integration: Seamlessly connect to APIs, databases, S3, and Snowflake.
    • Scalability: Handle large datasets with libraries like Pandas for transformations.
    • Automation: Use schedulers like Airflow or cron jobs to run pipelines periodically.
    • Cost-Effective: Open-source tools reduce overhead compared to proprietary ETL software.

    If you’re dealing with real-time data ingestion or batch processing, Python’s flexibility makes it a top choice for modern data pipelines.

    Step 1: Extracting Data from APIs

    Extracting data from APIs is a common starting point in data pipelines. We’ll use the requests library to fetch JSON data from a public API, such as a weather service or GitHub API.

    First, install the necessary packages:

    pip install requests pandas

    Here’s a sample Python script to extract data from an API:

    import requests
import pandas as pd

def extract_from_api(api_url):
    try:
        response = requests.get(api_url)
        response.raise_for_status()  # Raise error for bad status codes
        data = response.json()
        # Assuming the data is in a list under 'results' key
        df = pd.DataFrame(data.get('results', []))
        print(f"Extracted {len(df)} records from API.")
        return df
    except requests.exceptions.RequestException as e:
        print(f"API extraction error: {e}")
        return pd.DataFrame()

# Example usage
api_url = "https://api.example.com/data"  # Replace with your API endpoint
api_data = extract_from_api(api_url)

    This function handles errors gracefully and converts the API response into a Pandas DataFrame for easy manipulation in your data pipelines Python.

    Step 2: Extracting Data from External Databases

    For external databases like MySQL, PostgreSQL, or Oracle, use sqlalchemy to connect and query data. This is crucial for data pipelines involving legacy systems or third-party DBs.

    Install the required libraries:

    pip install sqlalchemy pandas mysql-connector-python  # Adjust driver for your DB

    Sample code to extract from a MySQL database:

    from sqlalchemy import create_engine
import pandas as pd

def extract_from_db(db_url, query):
    try:
        engine = create_engine(db_url)
        df = pd.read_sql_query(query, engine)
        print(f"Extracted {len(df)} records from database.")
        return df
    except Exception as e:
        print(f"Database extraction error: {e}")
        return pd.DataFrame()

# Example usage
db_url = "mysql+mysqlconnector://user:password@host:port/dbname"  # Replace with your credentials
query = "SELECT * FROM your_table WHERE date > '2023-01-01'"
db_data = extract_from_db(db_url, query)

    This method ensures secure connections and efficient data retrieval, forming a solid foundation for your pipelines in Python.

    Step 3: Transforming Data (Optional ETL Step)

    Before loading, transform the data using Pandas. For instance, merge API and DB data, clean duplicates, or apply calculations.

    # Assuming api_data and db_data are DataFrames
merged_data = pd.merge(api_data, db_data, on='common_column', how='inner')
merged_data.drop_duplicates(inplace=True)
merged_data['new_column'] = merged_data['value1'] + merged_data['value2']

    This step in data pipelines ensures data quality and relevance.

    Step 4: Loading Data to Amazon S3

    Amazon S3 provides durable, scalable storage for your extracted data. Use boto3 to upload files.

    Install boto3:

    pip install boto3

    Code example:

    import boto3
import io

def load_to_s3(df, bucket_name, file_key, aws_access_key, aws_secret_key):
    try:
        s3_client = boto3.client('s3', aws_access_key_id=aws_access_key, aws_secret_access_key=aws_secret_key)
        csv_buffer = io.StringIO()
        df.to_csv(csv_buffer, index=False)
        s3_client.put_object(Bucket=bucket_name, Key=file_key, Body=csv_buffer.getvalue())
        print(f"Data loaded to S3: {bucket_name}/{file_key}")
    except Exception as e:
        print(f"S3 upload error: {e}")

# Example usage
bucket = "your-s3-bucket"
key = "data/processed_data.csv"
load_to_s3(merged_data, bucket, key, "your_access_key", "your_secret_key")  # Use environment variables for security

    Storing in S3 acts as an intermediate layer in data pipelines, enabling versioning and easy access.

    Step 5: Loading Data into Snowflake

    Finally, load the data from S3 into Snowflake for querying and analytics. Use the Snowflake Python connector.

    Install the connector:

    pip install snowflake-connector-python pandas

    Sample Script:

    import snowflake.connector
import pandas as pd

def load_to_snowflake(df, snowflake_account, user, password, warehouse, db, schema, table):
    try:
        conn = snowflake.connector.connect(
            user=user,
            password=password,
            account=snowflake_account,
            warehouse=warehouse,
            database=db,
            schema=schema
        )
        cur = conn.cursor()
        # Create table if not exists (simplified)
        cur.execute(f"CREATE TABLE IF NOT EXISTS {table} (col1 VARCHAR, col2 INT)")  # Adjust schema
        # Load data using Pandas to_sql (for small datasets; use COPY for large ones)
        df.to_sql(table, con=conn, schema=schema, if_exists='append', index=False)
        print(f"Data loaded to Snowflake table: {table}")
    except Exception as e:
        print(f"Snowflake load error: {e}")
    finally:
        cur.close()
        conn.close()

# Example usage
load_to_snowflake(merged_data, "your-account", "user", "password", "warehouse", "db", "schema", "your_table")

    For larger datasets, use Snowflake’s COPY INTO command with S3 stages for better performance in data pipelines Python.

    Best Practices for Data Pipelines in Python

    • Error Handling: Always include try-except blocks to prevent pipeline failures.
    • Security: Use environment variables or AWS Secrets Manager for credentials.
    • Scheduling: Integrate with Apache Airflow or AWS Lambda for automated runs.
    • Monitoring: Log activities and use tools like Datadog for pipeline health.
    • Scalability: For big data, consider PySpark or Dask instead of Pandas.

    Conclusion

    Building data pipelines Python from APIs and databases to S3 and Snowflake streamlines your ETL workflows, enabling faster insights. With the code examples provided, you can start implementing these pipelines today. If you’re optimizing for cloud efficiency, this setup reduces costs while boosting performance.

    Additional materials

  • How to Load Data into Snowflake: Guide to Warehouse, Stages and File Format

    How to Load Data into Snowflake: Guide to Warehouse, Stages and File Format

     In Part 1 of our guide, we covered the revolutionary architecture of Snowflake. Now, it’s time to get hands-on. A data platform is only as good as the data within it, so understanding how to efficiently load data into Snowflake is a fundamental skill for any data professional.

    This guide will walk you through the key concepts and practical steps for data ingestion, covering the role of virtual warehouses, the concept of staging, and the different methods for loading your data.

    Step 1: Choose Your Compute – The Virtual Warehouse

    Before you can load or query any data, you need compute power. In Snowflake, this is handled by a Virtual Warehouse. As we discussed in Part 1, this is an independent cluster of compute resources that you can start, stop, resize, and configure on demand.

    Choosing a Warehouse Size

    For data loading, the size of your warehouse matters.

    • For Bulk Loading: When loading large batches of data (gigabytes or terabytes) using the COPY command, using a larger warehouse (like a Medium or Large) can significantly speed up the process. The warehouse can process more files in parallel.
    • For Snowpipe: For continuous, micro-batch loading with Snowpipe, you don’t use your own virtual warehouse. Snowflake manages the compute for you on its own serverless resources.

    Actionable Tip: Create a dedicated warehouse specifically for your loading and ETL tasks, separate from your analytics warehouses. You can name it something like ETL_WH. This isolates workloads and helps you track costs.

    Step 2: Prepare Your Data – The Staging Area

    You don’t load data directly from your local machine into a massive Snowflake table. Instead, you first upload the data files to a Stage. A stage is an intermediate location where your data files are stored before being loaded.

    There are two main types of stages:

    1. Internal Stage: Snowflake manages the storage for you. You use Snowflake’s tools (like the PUT command) to upload your local files to this secure, internal location.
    2. External Stage: Your data files remain in your own cloud storage (AWS S3, Azure Blob Storage, or Google Cloud Storage). You simply create a stage object in Snowflake that points to your bucket or container.

    Best Practice: For most production data engineering workflows, using an External Stage is the standard. Your data lake already resides in a cloud storage bucket, and creating an external stage allows Snowflake to securely and efficiently read directly from it.

    Step 3: Load the Data – Snowpipe vs. COPY Command

    Once your data is staged, you have two primary methods to load it into a Snowflake table.

    A) The COPY INTO Command for Bulk Loading

    The COPY INTO <table> command is the workhorse for bulk data ingestion. It’s a powerful and flexible command that you execute manually or as part of a scheduled script (e.g., in an Airflow DAG).

    • Use Case: Perfect for large, scheduled batch jobs, like a nightly ETL process that loads all of the previous day’s data at once.
    • How it Works: You run the command, and it uses the resources of your active virtual warehouse to load the data from your stage into the target table.

    Example Code:SQL

    -- This command loads all Parquet files from our external S3 stage
COPY INTO my_raw_table
FROM @my_s3_stage
FILE_FORMAT = (TYPE = 'PARQUET');

    B) Snowpipe for Continuous Loading

    Snowpipe is the serverless, automated way to load data. It uses an event-driven approach to automatically ingest data as soon as new files appear in your stage.

    • Use Case: Ideal for near real-time data from sources like event streams, logs, or IoT devices, where files are arriving frequently.
    • How it Works: You configure a PIPE object that points to your stage. When a new file lands in your S3 bucket, S3 sends an event notification that triggers the pipe, and Snowpipe loads the file.

    Step 4: Know Your File Formats

    Snowflake supports various file formats, but your choice has a big impact on performance and cost.

    • Highly Recommended: Use compressed, columnar formats like Apache Parquet or ORC. Snowflake is highly optimized to load and query these formats. They are smaller in size (saving storage costs) and can be processed more efficiently.
    • Good Support: Formats like CSV and JSON are fully supported. For these, Snowflake also provides a wide range of formatting options to handle different delimiters, headers, and data structures.
    • Semi-Structured Data: Snowflake’s VARIANT data type allows you to load semi-structured data like JSON directly into a single column and query it later using SQL extensions, offering incredible flexibility.

    Conclusion for Part 2

    You now understand the essential mechanics of getting data into Snowflake. The process involves:

    1. Choosing and activating a Virtual Warehouse for compute.
    2. Placing your data files in a Stage (preferably an external one on your own cloud storage).
    3. Using the COPY command for bulk loads or Snowflake for continuous ingestion.

    In Part 3 of our guide, we will explore “Transforming and Querying Data in Snowflake,” where we’ll cover the basics of SQL querying, working with the VARIANT data type, and introducing powerful concepts like Zero-Copy Cloning.

  • Automated ETL with Airflow and Python: A Practical Guide

    Automated ETL with Airflow and Python: A Practical Guide

    In the world of data, consistency is king. Manually running scripts to fetch and process data is not just tedious; it’s prone to errors, delays, and gaps in your analytics. To build a reliable data-driven culture, you need automation. This is where building an automated ETL with Airflow and Python becomes a data engineer’s most valuable skill.

    Apache Airflow is the industry-standard open-source platform for orchestrating complex data workflows. When combined with the power and flexibility of Python for data manipulation, you can create robust, scheduled, and maintainable pipelines that feed your analytics platforms with fresh data, day in and day out.

    This guide will walk you through a practical example: building an Airflow DAG that automatically fetches cryptocurrency data from a public API, processes it with Python, and prepares it for analysis.

    The Architecture: A Simple, Powerful Workflow

    Our automated pipeline will consist of a few key components, orchestrated entirely by Airflow. The goal is to create a DAG (Directed Acyclic Graph) that defines the sequence of tasks required to get data from our source to its destination.

    Here’s the high-level architecture of our ETL pipeline:

    Public API: Our data source. We’ll use the free CoinGecko API to fetch the latest cryptocurrency prices.

    Python Script: The core of our transformation logic. We’ll use the requests library to call the API and pandas to process the JSON response into a clean, tabular format.

    Apache Airflow: The orchestrator. We will define a DAG that runs on a schedule (e.g., daily), executes our Python script, and handles logging, retries, and alerting.

    Data Warehouse/Lake: The destination. The processed data will be saved as a CSV, which in a real-world scenario would be loaded into a data warehouse like Snowflake, BigQuery, or a data lake like Amazon S3.

    Let’s get into the code.

    Step 1: The Python ETL Script

    First, we need a Python script that handles the logic of fetching and processing the data. This script will be called by our Airflow DAG. We’ll use a PythonVirtualenvOperator in Airflow, which means our script can have its own dependencies.

    Create a file named get_crypto_prices.py in your Airflow project’s /include directory.

    /include/get_crypto_prices.py

    Python

    import requests
import pandas as pd
from datetime import datetime

def fetch_and_process_crypto_data():
    """
    Fetches cryptocurrency data from the CoinGecko API and processes it.
    """
    print("Fetching data from CoinGecko API...")
    url = "https://api.coingecko.com/api/v3/simple/price"
    params = {
        'ids': 'bitcoin,ethereum,ripple,cardano,solana',
        'vs_currencies': 'usd',
        'include_market_cap': 'true',
        'include_24hr_vol': 'true',
        'include_24hr_change': 'true'
    }
    
    try:
        response = requests.get(url, params=params)
        response.raise_for_status()  # Raise an exception for bad status codes
        data = response.json()
        print("Data fetched successfully.")

        # Process the JSON data into a list of dictionaries
        processed_data = []
        for coin, details in data.items():
            processed_data.append({
                'coin': coin,
                'price_usd': details.get('usd'),
                'market_cap_usd': details.get('usd_market_cap'),
                'volume_24h_usd': details.get('usd_24h_vol'),
                'change_24h_percent': details.get('usd_24h_change'),
                'timestamp': datetime.now().isoformat()
            })

        # Create a pandas DataFrame
        df = pd.DataFrame(processed_data)
        
        # In a real pipeline, you'd load this to a database.
        # For this example, we'll save it to a CSV in the local filesystem.
        output_path = '/tmp/crypto_prices.csv'
        df.to_csv(output_path, index=False)
        print(f"Data processed and saved to {output_path}")
        
    except requests.exceptions.RequestException as e:
        print(f"Error fetching data from API: {e}")
        raise

if __name__ == "__main__":
    fetch_and_process_crypto_data()

    Step 2: Creating the Airflow DAG

    Now, let’s create the Airflow DAG that will schedule and run this script. This file will live in your Airflow dags/ folder.

    We’ll use the @task decorator and the PythonVirtualenvOperator to create a clean, isolated task.

    dags/crypto_etl_dag.py

    Python

    from __future__ import annotations

import pendulum

from airflow.models.dag import DAG
from airflow.operators.python import PythonVirtualenvOperator

with DAG(
    dag_id="crypto_price_etl_pipeline",
    start_date=pendulum.datetime(2025, 9, 27, tz="UTC"),
    schedule="0 8 * * *",  # Run daily at 8:00 AM UTC
    catchup=False,
    tags=["api", "python", "etl"],
    doc_md="""
    ## Cryptocurrency Price ETL Pipeline
    This DAG fetches the latest crypto prices from the CoinGecko API,
    processes the data with Python, and saves it as a CSV.
    """,
) as dag:
    
    run_etl_task = PythonVirtualenvOperator(
        task_id="run_python_etl_script",
        python_callable_source="""
from include.get_crypto_prices import fetch_and_process_crypto_data
fetch_and_process_crypto_data()
""",
        requirements=["pandas==2.1.0", "requests==2.31.0"],
        system_site_packages=False,
    )

    This DAG is simple but powerful. Airflow will now:

    • Run this pipeline automatically every day at 8:00 AM UTC.
    • Create a temporary virtual environment and install pandas and requests for the task.
    • Execute our Python function to fetch and process the data.
    • Log the entire process, and alert you if anything fails.

    Step 3: The Analytics Payoff

    With our pipeline running automatically, we now have a consistently updated CSV file (/tmp/crypto_prices.csv on the Airflow worker). In a real-world scenario where this data is loaded into a SQL data warehouse, an analyst can now run queries to derive insights, knowing the data is always fresh.

    An analyst could now answer questions like:

    • What is the daily trend of Bitcoin’s market cap?
    • Which coin had the highest percentage change in the last 24 hours?
    • How does trading volume correlate with price changes across different coins?

    Conclusion: Build Once, Benefit Forever

    By investing a little time to build an automated ETL with Airflow and Python, you create a resilient and reliable data asset. This approach eliminates manual, error-prone work and provides your analytics team with the fresh, trustworthy data they need to make critical business decisions. This is the core of modern data engineering: building automated systems that deliver consistent value.

  • How to Build a Data Lakehouse on Azure

    How to Build a Data Lakehouse on Azure

     For years, data teams have faced a difficult choice: the structured, high-performance world of the data warehouse, or the flexible, low-cost scalability of the data lake. But what if you could have the best of both worlds? Enter the Data Lakehouse, an architectural pattern that combines the reliability and performance of a warehouse with the openness and flexibility of a data lake. And when it comes to implementation, building a data lakehouse on Azure has become the go-to strategy for future-focused data teams.

    The traditional data lake, while great for storing vast amounts of raw data, often turned into a “data swamp”—unreliable and difficult to manage. The data warehouse, on the other hand, struggled with unstructured data and could become rigid and expensive. The Lakehouse architecture solves this dilemma.

    In this guide, we’ll walk you through the blueprint for building a powerful and modern data lakehouse on Azure, leveraging a trio of best-in-class services: Azure Data Lake Storage (ADLS) Gen2, Azure Databricks, and Power BI.

    The Azure Lakehouse Architecture: A Powerful Trio

    A successful Lakehouse implementation relies on a few core services working in perfect harmony. This architecture is designed to handle everything from raw data ingestion and large-scale ETL to interactive analytics and machine learning.

    Here’s the high-level architecture we will build:

    1. Azure Data Lake Storage (ADLS) Gen2: This is the foundation. ADLS Gen2 is a highly scalable and cost-effective cloud storage solution that combines the best of a file system with massive scale, making it the perfect storage layer for our Lakehouse.

    2. Azure Databricks: This is the unified analytics engine. Databricks provides a collaborative environment for data engineers and data scientists to run large-scale data processing (ETL/ELT) with Spark, build machine learning models, and manage the entire data lifecycle.

    3. Delta Lake: The transactional storage layer. Built on top of ADLS, Delta Lake is an open-source technology (natively integrated into Databricks) that brings ACID transactions, data reliability, and high performance to your data lake, effectively turning it into a Lakehouse.

    4. Power BI: The visualization and reporting layer. Power BI integrates seamlessly with Azure Databricks, allowing business users to run interactive queries and build insightful dashboards directly on the data in the Lakehouse.

    Let’s explore each component.

    Step 1: The Foundation – Azure Data Lake Storage (ADLS) Gen2

    Every great data platform starts with a solid storage foundation. For a Lakehouse on Azure, ADLS Gen2 is the undisputed choice. Unlike standard object storage, it includes a hierarchical namespace, which allows you to organize your data into directories and folders just like a traditional file system. This is critical for performance and organization in large-scale analytics.

    A best practice is to structure your data lake using a multi-layered approach, often called “medallion architecture”:

    • Bronze Layer (/bronze): Raw, untouched data ingested from various source systems.

    • Silver Layer (/silver): Cleaned, filtered, and standardized data. This is where data quality rules are applied.

    • Gold Layer (/gold): Highly aggregated, business-ready data that is optimized for analytics and reporting.

    Step 2: The Engine – Azure Databricks

    With our storage in place, we need a powerful engine to process the data. Azure Databricks is a first-class service on Azure that provides a managed, high-performance Apache Spark environment.

    Data engineers use Databricks notebooks to:

    • Ingest raw data from the Bronze layer.

    • Perform large-scale transformations, cleaning, and enrichment using Spark.

    • Write the processed data to the Silver and Gold layers.

    Here’s a simple PySpark code snippet you might run in a Databricks notebook to process raw CSV files into a cleaned-up table:

    # Databricks notebook code snippet

    # Define paths for our data layers

    bronze_path = “/mnt/datalake/bronze/raw_orders.csv”

    silver_path = “/mnt/datalake/silver/cleaned_orders”

    # Read raw data from the Bronze layer using Spark

    df_bronze = spark.read.format(“csv”) \

      .option(“header”, “true”) \

      .option(“inferSchema”, “true”) \

      .load(bronze_path)

    # Perform basic transformations

    from pyspark.sql.functions import col, to_date

    df_silver = df_bronze.select(

        col(“OrderID”).alias(“order_id”),

        col(“CustomerID”).alias(“customer_id”),

        to_date(col(“OrderDate”), “MM/dd/yyyy”).alias(“order_date”),

        col(“Amount”).cast(“decimal(18, 2)”).alias(“order_amount”)

      ).where(col(“Amount”).isNotNull())

    # Write the cleaned data to the Silver layer

    df_silver.write.format(“delta”).mode(“overwrite”).save(silver_path)

    print(“Successfully processed raw orders into the Silver layer.”)

    Step 3: The Magic – Delta Lake

    Notice the .format(“delta”) in the code above? That’s the secret sauce. Delta Lake is an open-source storage layer that runs on top of your existing data lake (ADLS) and brings warehouse-like capabilities.

    Key features Delta Lake provides:

    • ACID Transactions: Ensures that your data operations either complete fully or not at all, preventing data corruption.

    • Time Travel (Data Versioning): Allows you to query previous versions of your data, making it easy to audit changes or roll back errors.

    • Schema Enforcement & Evolution: Prevents bad data from corrupting your tables by enforcing a schema, while still allowing you to gracefully evolve it over time.

    • Performance Optimization: Features like data skipping and Z-ordering dramatically speed up queries.

    By writing our data in the Delta format, we’ve transformed our simple cloud storage into a reliable, high-performance Lakehouse.

    Step 4: The Payoff – Visualization with Power BI

    With our data cleaned and stored in the Gold layer of our Lakehouse, the final step is to make it accessible to business users. Power BI has a native, high-performance connector for Azure Databricks.

    You can connect Power BI directly to your Databricks cluster and query the Gold tables. This allows you to:

    • Build interactive dashboards and reports.

    • Leverage Power BI’s powerful analytics and visualization capabilities.

    • Ensure that everyone in the organization is making decisions based on the same, single source of truth from the Lakehouse.

    Conclusion: The Best of Both Worlds on Azure

    By combining the low-cost, scalable storage of Azure Data Lake Storage Gen2 with the powerful processing engine of Azure Databricks and the reliability of Delta Lake, you can build a truly modern data lakehouse on Azure. This architecture eliminates the need to choose between a data lake and a data warehouse, giving you the flexibility, performance, and reliability needed to support all of your data and analytics workloads in a single, unified platform.

  • Building a Serverless Data Pipeline on AWS: A Step-by-Step Guide

    Building a Serverless Data Pipeline on AWS: A Step-by-Step Guide

     For data engineers, the dream is to build pipelines that are robust, scalable, and cost-effective. For years, this meant managing complex clusters and servers. But with the power of the cloud, a new paradigm has emerged: the serverless data pipeline on AWS. This approach allows you to process massive amounts of data without managing a single server, paying only for the compute you actually consume.

    Going serverless means you can say goodbye to idle clusters, patching servers, and capacity planning. Instead, you use a suite of powerful AWS services that automatically scale to meet demand. This isn’t just a technical shift; it’s a strategic advantage that allows your team to focus on delivering value from data, not managing infrastructure.

    In this guide, we’ll walk you through the essential components and steps to build a modern, event-driven serverless data pipeline on AWS using S3, Lambda, AWS Glue, and Athena.

    The Architecture: A Four-Part Harmony

    A successful serverless pipeline relies on a few core AWS services working together seamlessly. Each service has a specific role, creating an efficient and automated workflow from raw data ingestion to analytics-ready insights.

    Here’s a high-level look at our architecture:

    1. Amazon S3 (Simple Storage Service): The foundation of our pipeline. S3 acts as a highly durable and scalable data lake where we will store our raw, processed, and curated data in different stages.
    2. AWS Lambda: The trigger and orchestrator. Lambda functions are small, serverless pieces of code that can run in response to events, such as a new file being uploaded to S3.
    3. AWS Glue: The serverless ETL engine. Glue can automatically discover the schema of our data and run powerful Spark jobs to clean, transform, and enrich it, converting it into an optimized format like Parquet.
    4. Amazon Athena: The interactive query service. Athena allows us to run standard SQL queries directly on our processed data stored in S3, making it instantly available for analysis without needing a traditional data warehouse.

    Now, let’s build it step-by-step.

    Step 1: Setting Up the S3 Data Lake Buckets

    First, we need a place to store our data. A best practice is to use separate prefixes or even separate buckets to represent the different stages of your data pipeline, creating a clear and organized data lake.

    For this guide, we’ll use a single bucket with three prefixes:

    • s3://your-data-lake-bucket/raw/: This is where raw, unaltered data lands from your sources.
    • s3://your-data-lake-bucket/processed/: After cleaning and transformation by our Glue job, the data is stored here in an optimized format (e.g., Parquet).
    • s3://your-data-lake-bucket/curated/: (Optional) A final layer for business-level aggregations or specific data marts.

    Step 2: Creating the Lambda Trigger

    Next, we need a mechanism to automatically start our pipeline when new data arrives. AWS Lambda is perfect for this. We will create a Lambda function that “listens” for a file upload event in our raw/ S3 prefix and then starts our AWS Glue ETL job.

    Here is a sample Python code for the Lambda function:

    lambda_function.py

    Python

    import boto3
import os

def lambda_handler(event, context):
    """
    This Lambda function is triggered by an S3 event and starts an AWS Glue ETL job.
    """
    # Get the Glue job name from environment variables
    glue_job_name = os.environ['GLUE_JOB_NAME']
    
    # Extract the bucket and key from the S3 event
    bucket = event['Records'][0]['s3']['bucket']['name']
    key = event['Records'][0]['s3']['object']['key']
    
    print(f"File uploaded: s3://{bucket}/{key}")
    
    # Initialize the Glue client
    glue_client = boto3.client('glue')
    
    try:
        print(f"Starting Glue job: {glue_job_name}")
        response = glue_client.start_job_run(
            JobName=glue_job_name,
            Arguments={
                '--S3_SOURCE_PATH': f"s3://{bucket}/{key}"
            }
        )
        print(f"Successfully started Glue job run. Run ID: {response['JobRunId']}")
        return {
            'statusCode': 200,
            'body': f"Started Glue job {glue_job_name} for file s3://{bucket}/{key}"
        }
    except Exception as e:
        print(f"Error starting Glue job: {e}")
        raise e

    To make this work, you need to:

    1. Create this Lambda function in the AWS console.
    2. Set an environment variable named GLUE_JOB_NAME with the name of the Glue job you’ll create in the next step.
    3. Configure an S3 trigger on the function, pointing it to your s3://your-data-lake-bucket/raw/ prefix for “All object create events.”

    Step 3: Transforming Data with AWS Glue

    AWS Glue is the heavy lifter in our pipeline. It’s a fully managed ETL service that makes it easy to prepare and load your data for analytics. For this step, you would create a Glue ETL job.

    Inside the Glue Studio, you can visually build a job or write a PySpark script. The job will:

    1. Read the raw data (e.g., CSV) from the source path passed by the Lambda function.
    2. Perform transformations, such as changing data types, dropping columns, or joining with other datasets.
    3. Write the transformed data to the processed/ S3 prefix in Apache Parquet format. Parquet is a columnar storage format that is highly optimized for analytical queries.

    Your Glue job will have a simple script that looks something like this:

    Python

    import sys
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job

# Get job arguments
args = getResolvedOptions(sys.argv, ['JOB_NAME', 'S3_SOURCE_PATH'])

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

# Read the raw CSV data from S3
source_dyf = glueContext.create_dynamic_frame.from_options(
    connection_type="s3",
    connection_options={"paths": [args['S3_SOURCE_PATH']]},
    format="csv",
    format_options={"withHeader": True},
)

# Convert to Parquet and write to the processed location
glueContext.write_dynamic_frame.from_options(
    frame=source_dyf,
    connection_type="s3",
    connection_options={"path": "s3://your-data-lake-bucket/processed/"},
    format="parquet",
)

job.commit()

    Step 4: Querying Processed Data with Amazon Athena

    Once your data is processed and stored as Parquet in S3, it’s ready for analysis. With Amazon Athena, you don’t need to load it into another database. You can query it right where it is.

    1. Create a Database: In the Athena query editor, create a database for your data lake: CREATE DATABASE my_data_lake;
    2. Run a Glue Crawler (or Create a Table): The easiest way to make your data queryable is to run an AWS Glue Crawler on your processed/ S3 prefix. The crawler will automatically detect the schema of your Parquet files and create an Athena table for you.
    3. Query Your Data: Once the table is created, you can run standard SQL queries on it.

    SQL

    SELECT
    customer_id,
    order_status,
    COUNT(order_id) as number_of_orders
FROM
    my_data_lake.processed_data
WHERE
    order_date >= '2025-01-01'
GROUP BY
    1, 2
ORDER BY
    3 DESC;

    Conclusion: The Power of Serverless

    You have now built a fully automated, event-driven, and serverless data pipeline on AWS. When a new file lands in your raw S3 bucket, a Lambda function triggers a Glue job that processes the data and writes it back to S3 in an optimized format, ready to be queried instantly by Athena.

    This architecture is not only powerful but also incredibly efficient. It scales automatically to handle terabytes of data and ensures you only pay for the resources you use, making it the perfect foundation for a modern data engineering stack.