Forem: Varun Joshi

From Python to Production Pipeline :A Practical guide to Apache Airflow

Varun Joshi — Sun, 24 May 2026 17:56:31 +0000

You have been using python and you have written scripts that pull data, clean it and load it somewhere, May be you even have a while True loop running on the server somewhere or a cron job you are afraid to touch.

This tutorial is for you

Apache airflow is the tool that takes your python script from "runs on my machine" to "runs reliably at 6 am every day, retries on failure, sends an alert if something breaks and has UI to see what exactly happened.It is the standard orchestration tool at most data engineering teams and more approachable as it looks.

By the end of this post you will understand how airflow thinks and you will have a working DAG that schedules real data pipeline.

What Airflow actually is (and isn't)?

Airflow is an workflow orchestrator. It does not move the data itself but it schedules and monitors the tasks that do.

Think of it like a very smart CRON job:

Knows the order the task should run.
Retries failed tasks automatically.
Keeps full history of every run.
Shows you everything on the UI.
Sends alert when things go wrong.

What it is not: a data processing engine. You still write Python to do the actual work. Airflow decides when to run it and what to do when it fails.

Core Concepts in 2 minutes

Before touching the code three terms that you should know:

DAG(Directed Acyclic Graph):This is your pipeline. A DAG is a python file that defines a set of tasks and the order they run in. The Acyclic part just means that tasks cannot loop back on their own.

Task: A Single unit of work inside a DAG, could be running a python function, executing a SQL query, calling an API or dozens of other things.

Operator: The building block of task, PythonOperator runs the Python function.BashOperator runs a shell command. PostGresOperator runs a SQL, airflow has operator for almost everything.

Setting up Airflow Locally

The quickest way to get Airflow running locally is with pip:

pip install apache-airflow
airflow dbt init
airflow user create \
   --username admin \
   --password admin \
   --firstname your \
   --lastname name \
   --role admin \
   --email admin@x.com \

Start scheduler and server in two different terminals

#Terminal 1
airflow scheduler

#Terminal 2
airflow webserver --port 8080

Now visit http://localhost:8080 -You will see the Airflow UI with a bunch of example DAG's already there.

Your first DAG: A Real Data Pipeline

Lets build something useful and not just a 'Hello world', we will build a pipeline :

1- Fetches data from a public API
2- Cleans and transforms it
3- Saves it in CSV

This mirrors a real ingestion pipeline without enterprise complexity.

Create a file called weather_pipeline.py inside your airflow dags/ folder.

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta
import requests
import pandas as pd
import os

default_args= {
    'owner': 'varun'
    'retries': 2,
    'retry_delay': timedelta(minutes=5),
    'email_in_future': False,
}

#Define a DAG----

with DAG(
   dag_id='weather_pipeline'
   default_args = default_args,
   description = 'Fetch, transform and save daily weather data'
   schedule_interval = '0 6 * * *' # runs daily at 6 am
   start_date=datetime(2024,1,1)
   catchup = False,
   tags= ['Weather','Tutorial']
) as dag:
   pass

*A few things to note
*

schedule_interval is a CRON expression.
catchup=False tells airflow not to run all the historical dates between start_date and today. You almost always want it False
Retries =2 means airflow will try it twice before marking it as failed.

Adding Tasks

Now adding three tasks that into the place of pass

def fetch_weather(**context):
    """Pull weather from open meteo (free, No API needed)"""
    url = "https://api.open-meteo.com/v1/forecast"
    params = {
       "latitude": 28.6,
       "longitude": 77.2,
       "daily":"temperature_2m_max, temperature_2m_min, precipitation_sum",
       "timezone":"Asia/Kolkata",
       "forecast_days": 1
 }
    response = request.get(url, params=params)
    response.raise_for_status() #raise an exception

    data = response.json()
    #Push data to Xcom so the next task can access it
    context['ti'].xcom_push(key='raw_weather', value =data)
    print(f"Fetched weather data for {data['daily']['time'][0]}")

def transform_weather(**context):
   """clean and reshape the raw API response"""
   raw = context['ti'].xcom_pull(key='raw_weather', task_ids='fetch_weather')
   daily = raw['daily']

   df = pd.DataFrame({
        'date': daily['time']
        'temp_max_c': daily['temperature_2m_max'],
        'temp_min_c':daily['temperature_2m_min'],
        'precipitation_mm':daily['precipitation_mm'],
        })
   context['ti'].xcom_push(key='clean_weather', value=df.to_dict('records'))
        print(f"✓ Transformed {len(df)} rows")


   df save_weather(**context):
      """Append the cleaned data to csv file"""
      records = context["ti"].xcom_pull(key='clean_weather', task_ids='transform_weather')

      df= pd.DataFrame(records)
      output_path= '/tmp/weather_data.csv'
      file_exists = os.path.exists(output_path)

      df.to_csv(output_path, mode='a', header=not file_exists, index=False)

      print(f"✓ Saved to {output_path}")


   #Wiring the tasks
   t1 = PythonOperator(task_id = 'fetch_weather', python_callable=fetch_weather)
   t2 = PythonOperator(task_id = 'transform_weather', python_callable=transform_weather)
   t3 = PythonOperator(task_id = 'Save_weather', python_callable=Save_weather)

The t1>>t2>>t3 at the bottom is a way of saying that 'run these in this order'. This is one of those things that might seems weird but later you will get used to it.

Xcom:How tasks talk to each other

You must have noticed xcom_push and xcom_pull in the code.
This is how airflow task passes data between each other.

XCom(cross-communication) is a small-key store built into Airflow.
One task pushes a value, the next pulls it by TaskID and Key.

#Task-1 Push
context['ti'].xcom_push(key='my_data', value={'rows':42})
#Task-2 Pull
data = context['ti'].xcom_pull(key='my_data', task_ids='task_1')

Xcom is stored in Airflow metadata database, so its not meant for large datasets.Use it for small payloads - counts, file paths,API responses.For large data, write to S3/GCS and pass the file path via Xcom.

How to Run

Once the file is in your dags/ folder.Airflow picks it up automatically. You should see it appear in http://localhost:8080.

To trigger in manually:

Find weather_pipeline in the DAG list.
Toggle it on.
Trigger DAG

Look at the graph view and the task should turn green as it completes.If something fails the task shows Red.

Pattern that Matter in Production

Use execution_date for idempotent pipelines.

def fetch_weather(**context):
    run_date = context['execution_date'].strftime('%Y-%m-%d')
    #fetch data for that specific date

This makes it your job re-runnable so that if this fails for Tuesday, it can be run on Wednesday for fetching Tuesday data.

Always set retries and retry_delay

Network calls fail. APIs go down, Databases go slow.
Setting retries saves you a 2 am call.

default_args = {
      'retries': 3
      'retry_delay' : timedelta(minutes=10)

}

Use on_failure_callback for alerting

def alert_on_failure(context):
    #send a slack message, email etc
    print(f"Task failed:{context['task_instance'].task_id}")

default_args = {
   'on_failure_callback': alert_on_failure,
}

Keep tasks small and focused

One task= one thing, Dont write a single giant function that does everything.Small tasks mean cleaner retries and easy to read logs.

That's it

You went from running the python code manually in your machine to scheduling it with retry functionality along with automatically executing tasks step by step with failure notification.

The learning curve is real, but it is mostly in the setup. Once you have written two or three DAG's, the pattern click and you are not going back to CRON.

Anatomy of Duck DB for Python Developers

Varun Joshi — Sun, 17 May 2026 01:32:16 +0000

Introduction - SQL without a Server

Pandas is widely used for data analysis and almost every data analyst or even data engineers utilize it for faster analysis with table like data structure called DataFrames.The drawback is that it suffers once the data goes beyond few GB's and spinning up a Postgres or a Redshift is an overkill for quick analysis.Duck DB fills this gap with Zero-setup columnar SQL.

Getting Started - zero config, instant power

DuckDb is an open source OLAP database management system designed for analytics and for running within the same process as the application.
It is lightweight, can work directly with data files in csv, parquet etc without needing a server.

Installation and first query

pip install duckdb - No ports to open, No configuration and No daemon

In-Memory and Persistent Database - Two Operating Modes

In-Memory

When DuckDB connection is created without specifying a file, a database lives entirely in RAM.

import duckdb
con = duckdb.connect()          # or duckdb.connect(':memory:')

All data is stored in RAM and no files are written to disk
Extremely fast reads/writes since there is zero I/O overhead.
Data is completely lost when connection closes.
No file locking or concurrency concerns

Persistent Mode

When the user provides a location DuckDB can write the results to disk in .duckDb format.

con = duckdb.connect('my_database.duckdb')

Tables,Schemas and indexes are persisted.
Uses a columnar storage format with compression and buffered I/O
Only one write connection at a time but multiple read connection are allowed.
Supports WAL(Write Ahead Logging) for crash recovery

Powerful Pattern

DuckDb allows you to mix both modes where user can start with in-memory and attach a persistent database or use copy/export to snapshot in-memory result to disk.

con = duckdb.connect()

#Query a CSV, transform it, save the result to a persistent file
con.execute("""
    COPY(SELECT region, SUM(sales) AS total FROM read_csv('data.csv')
         GROUP BY region
     )
    TO 'results.parquet' (FORMAT PARQUET)
""")

Users gets the speed of In-Memory processing which accelerates the pipeline processing with an option to persist.

Reading files directly --CSV,PARQUET,JSON,Arrow,

Query CSV without loading into memory

Select * from read_csv('data_csv', auto_detect=true);

-Auto detects delimiter, compression and data types
-Handles malformed rows gracefully
-Can read multiple CSVs at once read_csv('data/*.csv')

Parquet

Select * from read_parquet('data.parquet');
--even from S3 directly
Select * from read_parquet('s3://bucket/data/*.parquet');

Exploits column pruning as it only reads columns you need
Leverages row group skipping using Parquet's build in min/max stats
Native support for nested types(structs,list,maps)

JSON/NDJSON

SELECT * FROM read_json('events.ndjson', auto_detect=true);

-AUTO INFERS schema from data
-NDJSON(Newline delimited) streams efficiently line by line
-Can unnest deeply nested JSON fields using DuckDB's json_extract, UNNEST, or -> operators

Apache Arrow

import pyarrow as pa
arrow_table = pa.Table.from_pandas(df)
duckdb.query(""SELECT * from arrow_table""")

-Zero copy integration: DuckDB reads from Arrow memory without serialization
-Ideal for pipelines where data never needs to touch disk

SQL Beyond Select

DuckDB is not just a query engine, it supports rich SQL that covers data transformation, creation, and some genuinely unique syntax extensions to available in most databases.

Full Suite of WINDOW Functions

Select
    customer,
    ordered_at,
    amount,

    -- Running total
    SUM(amount) OVER (PARTITION BY customer ORDER BY ordered_at) AS running_tot,

    -- Lag/lead comparisons
    LAG(amount) OVER (PARTITION BY customer ORDER BY ordered_at) AS prev_amt,

    -- Percentile rank
    PERCENT_RANK() OVER (ORDER BY amount) AS pct_rank,

    -- Named window reuse
    FIRST_VALUE(amount) OVER w AS first_order
FROM orders
WINDOW w AS (PARTITION BY customer ORDER BY ordered_at);

DuckDB also allows the use of qualify clause which filters on window result without a subquery.

Select * From orders
QUALIFY ROW_NUMBER() OVER (PARTITION BY customer ORDER BY amount DESC) = 1;

PIVOT and UNPIVOT

Most databases make you write case when manually for PIVOTS.
DuckDB does it natively.

--PIVOT- rows to columns
PIVOT orders on region USING SUM(amount) GROUP BY year;

--UNPIVOT- Column to rows
UNPIVOT sales_wide ON(q1,q2,q3,q4) INTO NAME quarter VALUE revenue;

MULTI DATABASE SQL

--Attach another DuckDB file
ATTACH 'archive.duckdb' AS archive;

-- Cross-database join
SELECT a.*, b.region
FROM main.orders a
JOIN archive.customers b ON a.customer_id = b.id;

--Attach another database
ATTACH 'postgres://user:pass@host/db' AS pg (TYPE POSTGRES);
SELECT * FROM pg.public.users LIMIT 10;

DUCKDB+Pandas+Polars --Choosing your stack

DuckDB does not replace pandas or Polars it solves a problem which is niche.The sweet spot of the industry is to use DuckDB for SQL-shaped operations and pandas/polars for row level python logic.

Where Duck DB shines

Feature Engineering for ML: Window functions or group by's for feature computation are often faster and more readable in DuckDB then pandas before handing it over to Sklearn or pytorch
Unit testing DBT models locally:DuckDB lets you run complete dbt project locally without a cloud warehouse providing fast feedback loop for data engineers.
Light weight ETL Pipelines: One can read raw parquet from S3, transform with SQL, write cleaned output back without any spark cluster or airflow jobs.

Conclusion

DuckDB lets you think in SQL for analytical tasks without worrying about infrastructure setup. Anyone using python can utilize duckdb for analysis of larger files where regular pandas will give headache.
Given the advantages, it is important to know whare DuckDB should not be used which in case of concurrent writes,OLTP workloads and long running multi user services.

Reference-
https://duckdb.org/docs/current/data/overview