What Skills Do You Need to Become a Data Engineer?

-- Keep only the latest record per order (common DE task)
WITH ranked AS (
  SELECT *,
         ROW_NUMBER() OVER (PARTITION BY order_id ORDER BY updated_at DESC) AS rn
  FROM raw_orders
)
SELECT order_id, customer_id, status, amount, updated_at
FROM ranked
WHERE rn = 1;

# pip install requests pandas
import requests
import pandas as pd

def extract_users() -> pd.DataFrame:
    resp = requests.get("https://jsonplaceholder.typicode.com/users", timeout=10)
    resp.raise_for_status()
    return pd.json_normalize(resp.json())

df = extract_users()
clean = df[["id", "name", "email", "address.city"]].rename(columns={"address.city": "city"})
clean.to_csv("users.csv", index=False)
print(f"extracted {len(clean)} users")

import sqlite3

con = sqlite3.connect("warehouse.db")
con.execute("""CREATE TABLE IF NOT EXISTS fact_sales
               (sale_date TEXT, city TEXT, amount REAL)""")

def load_day(day: str, rows: list):
    """Safe to rerun: replaces only that day's data."""
    with con:
        con.execute("DELETE FROM fact_sales WHERE sale_date = ?", (day,))
        con.executemany("INSERT INTO fact_sales VALUES (?, ?, ?)",
                        [(day, c, a) for c, a in rows])

load_day("2026-11-05", [("Noida", 500), ("Delhi", 750)])
load_day("2026-11-05", [("Noida", 500), ("Delhi", 750)])   # rerun
print(con.execute("SELECT COUNT(*) FROM fact_sales").fetchone()[0], "rows")  # 2

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator

def extract(**ctx): print("extract for", ctx["ds"])
def load(**ctx): print("load for", ctx["ds"])
def check(**ctx): print("check for", ctx["ds"])

with DAG("daily_sales", start_date=datetime(2026, 11, 1), schedule="@daily",
         catchup=False,
         default_args={"retries": 2, "retry_delay": timedelta(minutes=5)}) as dag:
    e = PythonOperator(task_id="extract", python_callable=extract)
    l = PythonOperator(task_id="load", python_callable=load)
    c = PythonOperator(task_id="check", python_callable=check)
    e >> l >> c

import sqlite3

con = sqlite3.connect("warehouse.db")
checks = {
    "no rows for yesterday": "SELECT COUNT(*) = 0 FROM fact_sales WHERE sale_date = '2026-11-05'",
    "negative amounts": "SELECT COUNT(*) > 0 FROM fact_sales WHERE amount < 0",
    "missing city": "SELECT COUNT(*) > 0 FROM fact_sales WHERE city IS NULL OR city = ''",
}
failed = [name for name, sql in checks.items() if con.execute(sql).fetchone()[0]]
if failed:
    raise SystemExit(f"data quality failed: {failed}")
print("all checks passed")

Dataset: fact_sales
Grain: one row per sale_date x city
Columns: sale_date (TEXT, YYYY-MM-DD), city (TEXT), amount (REAL, INR)
Source: daily orders export
Refresh: daily at 02:00; idempotent by sale_date
Quality checks: rows exist for yesterday, no negative amounts, city present
Owner: data engineering team
Known limits: refunds processed next day

Stage 1  SQL deep dive (joins, CTEs, window functions)
Stage 2  Python for data (files, APIs, pandas, errors) + Git
Stage 3  Data modelling (grain, facts, dimensions, keys)
Stage 4  Pipelines (idempotent loads, Airflow or cron)
Stage 5  Docker and Compose
Stage 6  One cloud: storage, warehouse, IAM basics
Stage 7  Data quality tests and monitoring
Stage 8  Two end-to-end portfolio projects

Rate 1-5 and plan the lowest first:

SQL window functions          ___
Python data scripts           ___
Git workflow                  ___
Data modelling and grain      ___
Idempotent pipelines          ___
Scheduling / orchestration    ___
Docker                        ___
Cloud basics                  ___
Data quality testing          ___
Documentation and communication ___

Project 1: Batch pipeline
  API or CSV -> clean -> warehouse tables -> quality checks -> schedule

Project 2: Modelled mart
  Raw events -> staging -> fact and dimension tables -> dataset cards

Both: Docker Compose setup, README, idempotent reruns.