Data Engineering Complete Guide 2025: Flink vs Spark, dbt, Iceberg, Airflow — Modern Data Stack Deep Dive

[Data Sources] → [Ingestion/CDC] → [Stream Processing] → [Storage/Lakehouse] → [Transform] → [Analytics/Serving]
       |               |                  |                     |                    |              |
    DB, API,       Airbyte,           Flink,             Iceberg/Delta,           dbt,        ClickHouse,
    IoT, SaaS     Debezium,           Spark              S3/GCS/ADLS          Spark SQL     Pinot, Druid
                   Fivetran          Streaming                                                StarRocks

Evaluation Criteria Checklist:
1. Latency requirements (batch vs near-real-time vs real-time)
2. Data volume (GB/day vs TB/day vs PB/day)
3. Team size and capabilities (SQL-centric vs code-centric)
4. Vendor lock-in tolerance
5. Budget (open source vs managed services)
6. Regulatory requirements (data governance, lineage)

[Event Stream] ──→ [Window] ──→ [Aggregate/Transform] ──→ [Sink]
                      |
              ┌───────┼───────┐
              │       │       │
          Tumbling  Sliding  Session
           Window   Window   Window
           (fixed)  (sliding) (session)

// Flink - Real-time fraud detection example
DataStream<Transaction> transactions = env
    .addSource(new FlinkKafkaConsumer<>("transactions", schema, props));

DataStream<Alert> alerts = transactions
    .keyBy(Transaction::getAccountId)
    .window(SlidingEventTimeWindows.of(Time.minutes(10), Time.minutes(1)))
    .aggregate(new FraudScoreAggregator())
    .filter(score -> score.getValue() > THRESHOLD);

alerts.addSink(new AlertSink());

-- Flink SQL: Real-time revenue aggregation
CREATE TABLE orders (
    order_id STRING,
    amount DECIMAL(10,2),
    order_time TIMESTAMP(3),
    WATERMARK FOR order_time AS order_time - INTERVAL '5' SECOND
) WITH (
    'connector' = 'kafka',
    'topic' = 'orders',
    'format' = 'json'
);

SELECT
    TUMBLE_START(order_time, INTERVAL '1' MINUTE) AS window_start,
    COUNT(*) AS order_count,
    SUM(amount) AS total_revenue
FROM orders
GROUP BY TUMBLE(order_time, INTERVAL '1' MINUTE);

# Spark Structured Streaming - Real-time order analytics
from pyspark.sql import SparkSession
from pyspark.sql.functions import window, sum, count

spark = SparkSession.builder.appName("OrderAnalytics").getOrCreate()

orders = spark.readStream \
    .format("kafka") \
    .option("subscribe", "orders") \
    .load()

order_stats = orders \
    .withWatermark("order_time", "5 minutes") \
    .groupBy(window("order_time", "1 minute")) \
    .agg(
        count("*").alias("order_count"),
        sum("amount").alias("total_revenue")
    )

order_stats.writeStream \
    .format("iceberg") \
    .outputMode("append") \
    .option("checkpointLocation", "/checkpoint/orders") \
    .toTable("catalog.db.order_stats")

# Apache Beam - Multi-runner pipeline
import apache_beam as beam

with beam.Pipeline(options=pipeline_options) as p:
    (p
     | 'ReadKafka' >> beam.io.ReadFromKafka(
         consumer_config=kafka_config,
         topics=['orders'])
     | 'ParseJSON' >> beam.Map(parse_order)
     | 'WindowInto' >> beam.WindowInto(
         beam.window.FixedWindows(60))  # 1-minute window
     | 'CountPerWindow' >> beam.combiners.Count.Globally()
     | 'WriteToGCS' >> beam.io.WriteToText('gs://bucket/output'))

Processing Latency (p99):
┌────────────────────────────────────────────┐
│ Flink     ████  23ms                       │
│ Spark     ████████████████  450ms           │
│ Beam/Flink████  25ms                       │
│ Beam/Spark████████████████  460ms           │
└────────────────────────────────────────────┘

Throughput (events/second):
┌────────────────────────────────────────────┐
│ Flink     ████████████████████  5.2M       │
│ Spark     ████████████████  3.8M           │
│ Beam/Flink██████████████████  4.8M         │
└────────────────────────────────────────────┘

Airflow 3.0 Core Upgrades:

1. React-based Modern UI
   - Real-time log streaming
   - Improved DAG visualization
   - Dark mode support

2. Event-Driven Scheduling
   - Dataset-aware scheduling (auto-trigger on data arrival)
   - Enhanced External Triggers

3. TaskFlow API 2.0
   - Decorator-based Python-native DAG authoring
   - Improved dynamic task mapping

4. Edge Labels & DAG Versioning
   - DAG change history tracking
   - A/B testing DAG support

5. Enhanced Security
   - Improved RBAC
   - Expanded Secret Backend integrations

# Airflow 3.0 - TaskFlow API Example
from airflow.decorators import dag, task
from datetime import datetime

@dag(schedule="@daily", start_date=datetime(2025, 1, 1), catchup=False)
def data_pipeline():

    @task()
    def extract():
        """Extract raw data from S3"""
        import boto3
        # Data extraction logic
        return {"records": 15000, "source": "s3://datalake/raw/"}

    @task()
    def transform(data: dict):
        """Transform data via dbt"""
        import subprocess
        subprocess.run(["dbt", "run", "--select", "staging+"])
        return {"transformed": data["records"], "models": 12}

    @task()
    def load(data: dict):
        """Load results into ClickHouse"""
        # ClickHouse loading logic
        return {"loaded": data["transformed"]}

    @task()
    def notify(result: dict):
        """Send Slack notification"""
        # Slack notification logic
        pass

    raw = extract()
    transformed = transform(raw)
    loaded = load(transformed)
    notify(loaded)

data_pipeline()

# Dagster - Software-Defined Assets Example
from dagster import asset, AssetExecutionContext
from dagster_dbt import DbtCliResource, dbt_assets

@asset(
    description="Extract raw order data from S3",
    group_name="raw",
    compute_kind="python"
)
def raw_orders(context: AssetExecutionContext):
    """Extract order data from S3"""
    import pandas as pd
    df = pd.read_parquet("s3://datalake/raw/orders/")
    context.log.info(f"Extracted {len(df)} orders")
    return df

@asset(
    description="Validate order data quality",
    group_name="validated",
    compute_kind="great_expectations"
)
def validated_orders(raw_orders):
    """Validate data quality with Great Expectations"""
    # Quality validation logic
    assert raw_orders["amount"].min() >= 0, "Negative amounts found"
    return raw_orders

# Automatic dbt asset integration
@dbt_assets(manifest=dbt_manifest_path)
def dbt_models(context: AssetExecutionContext, dbt: DbtCliResource):
    yield from dbt.cli(["build"], context=context).stream()

# Prefect - Concise Pipeline
from prefect import flow, task
from prefect.tasks import task_input_hash
from datetime import timedelta

@task(retries=3, retry_delay_seconds=60,
      cache_key_fn=task_input_hash,
      cache_expiration=timedelta(hours=1))
def extract_data(source: str) -> dict:
    """Extraction with automatic retries and caching"""
    # Data extraction
    return {"data": [...], "count": 15000}

@task(log_prints=True)
def transform_data(raw: dict) -> dict:
    """Transformation logic"""
    print(f"Processing {raw['count']} records")
    return {"transformed": raw["count"]}

@flow(name="daily-etl", log_prints=True)
def daily_pipeline():
    raw = extract_data("s3://datalake/raw/")
    result = transform_data(raw)
    print(f"Pipeline complete: {result}")

# Execute
daily_pipeline()

dbt Milestones:
- 2016: Fishtown Analytics founded
- 2020: dbt Cloud launched
- 2022: Rebranded to dbt Labs, Series D $222M
- 2023: MetricFlow open-sourced
- 2024: Semantic Layer GA in dbt Cloud
- 2025: Crossed $100M+ ARR, established de facto standard status

dbt_project/
├── dbt_project.yml
├── packages.yml
├── profiles.yml
├── models/
│   ├── staging/           # Source data cleansing
│   │   ├── stg_orders.sql
│   │   ├── stg_customers.sql
│   │   └── _staging.yml   # Tests & docs
│   ├── intermediate/      # Business logic composition
│   │   ├── int_order_items.sql
│   │   └── _intermediate.yml
│   ├── marts/             # Final analytics tables
│   │   ├── finance/
│   │   │   ├── fct_revenue.sql
│   │   │   └── dim_customers.sql
│   │   └── marketing/
│   │       └── fct_campaigns.sql
│   └── metrics/           # MetricFlow metrics
│       └── revenue.yml
├── tests/                 # Custom tests
│   └── assert_positive_revenue.sql
├── macros/                # Reusable SQL macros
│   └── cents_to_dollars.sql
├── seeds/                 # Static data (CSV)
│   └── country_codes.csv
└── snapshots/             # SCD Type 2 snapshots
    └── snap_customers.sql

-- models/staging/stg_orders.sql
WITH source AS (
    SELECT * FROM {{ source('raw', 'orders') }}
),

renamed AS (
    SELECT
        id AS order_id,
        user_id AS customer_id,
        status AS order_status,
        amount_cents / 100.0 AS amount_dollars,
        created_at AS ordered_at
    FROM source
    WHERE status != 'cancelled'
)

SELECT * FROM renamed

-- models/marts/finance/fct_revenue.sql
WITH orders AS (
    SELECT * FROM {{ ref('stg_orders') }}
),

customers AS (
    SELECT * FROM {{ ref('dim_customers') }}
),

final AS (
    SELECT
        o.order_id,
        o.customer_id,
        c.customer_segment,
        o.amount_dollars,
        o.ordered_at,
        DATE_TRUNC('month', o.ordered_at) AS order_month
    FROM orders o
    JOIN customers c ON o.customer_id = c.customer_id
    WHERE o.order_status = 'completed'
)

SELECT * FROM final

# models/staging/_staging.yml
version: 2

models:
  - name: stg_orders
    description: 'Cleansed order data'
    columns:
      - name: order_id
        description: 'Unique order identifier'
        tests:
          - unique
          - not_null
      - name: amount_dollars
        description: 'Order amount in dollars'
        tests:
          - not_null
          - dbt_utils.accepted_range:
              min_value: 0
              max_value: 100000

# models/metrics/revenue.yml
semantic_models:
  - name: orders
    defaults:
      agg_time_dimension: ordered_at
    model: ref('fct_revenue')
    entities:
      - name: order_id
        type: primary
      - name: customer_id
        type: foreign
    measures:
      - name: revenue
        agg: sum
        expr: amount_dollars
      - name: order_count
        agg: count
        expr: order_id
    dimensions:
      - name: ordered_at
        type: time
      - name: customer_segment
        type: categorical

metrics:
  - name: monthly_revenue
    type: simple
    type_params:
      measure: revenue
  - name: revenue_per_customer
    type: derived
    type_params:
      expr: monthly_revenue / active_customers
      metrics:
        - name: monthly_revenue
        - name: active_customers

Top 10 dbt Project Best Practices:

1. Enforce layer structure: staging -> intermediate -> marts
2. Complete all renaming and type casting in staging
3. Apply unique + not_null tests on all models
4. Monitor data freshness with source freshness checks
5. Optimize large tables with incremental models
6. Leverage dbt-utils, dbt-expectations via packages.yml
7. SQL linting with pre-commit hooks (sqlfluff)
8. Use slim CI (state:modified+) in CI pipelines
9. Separate profiles by environment (dev / staging / prod)
10. Centralize metric management with Semantic Layer

Lakehouse = Low-cost storage of data lakes + ACID/performance of warehouses

Traditional Architecture:
[Sources] -> [Data Lake (S3)] -> [ETL] -> [Data Warehouse (Redshift)]
                                           High cost, duplicate storage

Lakehouse Architecture:
[Sources] -> [Object Storage (S3)] + [Table Format (Iceberg)]
              Low cost                 ACID, schema evolution, time travel
              └──> [Analytics Engine (Spark/Trino/Flink)] Direct query

-- Iceberg table creation and usage
CREATE TABLE catalog.db.orders (
    order_id    BIGINT,
    customer_id BIGINT,
    amount      DECIMAL(10,2),
    status      STRING,
    ordered_at  TIMESTAMP
)
USING iceberg
PARTITIONED BY (days(ordered_at));

-- Time travel: query data as of yesterday
SELECT * FROM catalog.db.orders
VERSION AS OF '2025-03-21T00:00:00';

-- Snapshot-based incremental reads
SELECT * FROM catalog.db.orders
WHERE ordered_at > (
    SELECT max(ordered_at)
    FROM catalog.db.orders
    VERSION AS OF 'snapshot_id_123'
);

# Delta Lake - UniForm for Iceberg compatibility
from delta.tables import DeltaTable

# Create Delta table (UniForm enabled)
spark.sql("""
    CREATE TABLE catalog.db.events (
        event_id BIGINT,
        event_type STRING,
        payload STRING,
        event_time TIMESTAMP
    )
    USING DELTA
    TBLPROPERTIES (
        'delta.universalFormat.enabledFormats' = 'iceberg'
    )
""")

# Delta table MERGE (Upsert)
delta_table = DeltaTable.forName(spark, "catalog.db.events")

delta_table.alias("target").merge(
    new_events.alias("source"),
    "target.event_id = source.event_id"
).whenMatchedUpdateAll() \
 .whenNotMatchedInsertAll() \
 .execute()

-- ClickHouse - Real-time dashboard queries
CREATE TABLE events (
    event_id UInt64,
    user_id UInt32,
    event_type LowCardinality(String),
    properties Map(String, String),
    event_time DateTime64(3)
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(event_time)
ORDER BY (event_type, user_id, event_time);

-- Materialized View for real-time aggregation
CREATE MATERIALIZED VIEW hourly_events_mv
ENGINE = SummingMergeTree()
ORDER BY (event_type, hour)
AS SELECT
    event_type,
    toStartOfHour(event_time) AS hour,
    count() AS event_count,
    uniqHLL12(user_id) AS unique_users
FROM events
GROUP BY event_type, hour;

-- Millisecond responses across billions of rows
SELECT
    event_type,
    sum(event_count) AS total_events,
    sum(unique_users) AS total_users
FROM hourly_events_mv
WHERE hour >= now() - INTERVAL 24 HOUR
GROUP BY event_type
ORDER BY total_events DESC
LIMIT 20;

TPC-H 10GB Query Performance (seconds, lower is better):
┌──────────────────────────────────────────────────┐
│ StarRocks 3.x  ████  2.1s (2.2x faster than CH) │
│ ClickHouse     █████████  4.6s                   │
│ Druid          ████████████████████  18.7s       │
│ Pinot          ██████████████████  16.3s         │
└──────────────────────────────────────────────────┘

1. Domain Ownership
   -> Each domain team owns their data

2. Data as a Product
   -> Apply SLA, documentation, quality guarantees to data

3. Self-Serve Platform
   -> Infrastructure team provides data platform tools

4. Federated Governance
   -> Balance between enterprise standards + domain autonomy

# data-contract.yml example
dataContractSpecification: 0.9.3
id: orders-contract
info:
  title: 'Orders Data Contract'
  version: 2.1.0
  owner: order-domain-team
  contact:
    name: 'Orders Team'
    email: orders@company.com

servers:
  production:
    type: iceberg
    catalog: polaris
    database: orders_db
    table: orders

models:
  orders:
    description: 'Completed order data'
    fields:
      order_id:
        type: bigint
        required: true
        unique: true
        description: 'Unique order identifier'
      customer_id:
        type: bigint
        required: true
      amount:
        type: decimal
        required: true
        minimum: 0
        description: 'Order amount (USD)'
      ordered_at:
        type: timestamp
        required: true

quality:
  type: SodaCL
  specification:
    checks for orders:
      - row_count > 0
      - missing_count(order_id) = 0
      - duplicate_count(order_id) = 0
      - avg(amount) between 10 and 500

sla:
  freshness: 1 hour
  availability: 99.9%
  latency: p99 < 500ms

Data Mesh + Fabric Hybrid:

                    ┌──────────────────────────┐
                    │    Data Fabric Layer      │
                    │ (Unified Governance/Catalog)│
                    └──────────┬───────────────┘
            ┌──────────────────┼──────────────────┐
            │                  │                  │
     ┌──────┴──────┐   ┌──────┴──────┐   ┌──────┴──────┐
     │ Order Domain│   │Customer Domn│   │Product Domn │
     │ Data Product│   │ Data Product│   │ Data Product│
     │  (Iceberg)  │   │  (Iceberg)  │   │  (Iceberg)  │
     └─────────────┘   └─────────────┘   └─────────────┘

Requirements:
- Daily active users: 5 million
- Events per second: 50,000
- Recommendation latency: p99 < 100ms
- Data sources: Click streams, purchases, searches, inventory
- Feature store: Combined real-time + batch features
- A/B testing: Performance comparison across model versions

Real-Time Recommendation System Architecture:

[Click/Purchase/Search Events]
        |
        v
   ┌─────────┐     ┌──────────────┐
   │  Kafka  │────>│  Flink (CEP) │──> Real-time feature computation
   │ Cluster │     │ Window Agg.  │    (recent 5-min click categories)
   └────┬────┘     └──────┬───────┘
        │                 │
        v                 v
   ┌─────────┐     ┌──────────────┐
   │ Iceberg │     │    Redis     │──> Real-time feature serving
   │  (raw)  │     │ Feature Store│    (p99 < 5ms)
   └────┬────┘     └──────────────┘
        │                 ^
        v                 │
   ┌─────────┐     ┌──────────────┐
   │   dbt   │────>│  ClickHouse  │──> Batch feature computation
   │(transform│    │  (agg/query) │    (30-day purchase patterns)
   └────┬────┘     └──────────────┘
        │
        v
   ┌─────────────────┐     ┌──────────────┐
   │ ML Model Serving │<───│ Feature Store │
   │ (TorchServe/     │    │ (Redis+DuckDB)│
   │  Triton)         │    └──────────────┘
   └────────┬────────┘
            │
            v
   ┌─────────────────┐
   │   API Gateway   │──> Serve recommendations to users
   │  (p99 < 100ms)  │
   └─────────────────┘

// Real-time user behavior feature computation
DataStream<UserFeature> realtimeFeatures = clickEvents
    .keyBy(ClickEvent::getUserId)
    .window(SlidingEventTimeWindows.of(
        Time.minutes(5), Time.minutes(1)))
    .aggregate(new UserBehaviorAggregator());

// Store to Redis feature store
realtimeFeatures.addSink(
    new RedisSink<>(redisConfig, new FeatureRedisMapper()));

-- models/features/user_purchase_patterns.sql
-- 30-day purchase pattern features (batch)
WITH purchase_stats AS (
    SELECT
        customer_id,
        COUNT(*) AS purchase_count_30d,
        SUM(amount_dollars) AS total_spend_30d,
        AVG(amount_dollars) AS avg_order_value_30d,
        COUNT(DISTINCT category) AS unique_categories_30d,
        MAX(ordered_at) AS last_purchase_at
    FROM {{ ref('fct_revenue') }}
    WHERE ordered_at >= CURRENT_DATE - INTERVAL '30' DAY
    GROUP BY customer_id
)

SELECT
    customer_id,
    purchase_count_30d,
    total_spend_30d,
    avg_order_value_30d,
    unique_categories_30d,
    DATEDIFF('day', last_purchase_at, CURRENT_DATE) AS days_since_last_purchase
FROM purchase_stats

# Recommendation system daily pipeline
@dag(schedule="0 2 * * *", catchup=False)
def recommendation_pipeline():

    @task()
    def run_dbt_features():
        """Generate batch features with dbt"""
        import subprocess
        result = subprocess.run(
            ["dbt", "run", "--select", "features+"],
            capture_output=True, text=True
        )
        return result.returncode == 0

    @task()
    def export_to_feature_store(dbt_success: bool):
        """Export batch features to Feature Store"""
        if not dbt_success:
            raise Exception("dbt run failed")
        # ClickHouse -> Redis feature store sync
        pass

    @task()
    def retrain_model():
        """Retrain ML model"""
        # Daily model retraining logic
        pass

    @task()
    def validate_model(retrain_result):
        """Validate model performance"""
        # A/B test metric comparison
        pass

    dbt_result = run_dbt_features()
    export = export_to_feature_store(dbt_result)
    retrain = retrain_model()
    validate_model(retrain)

recommendation_pipeline()

Monthly Estimated Costs (AWS, 5M daily users):

| Component       | Instance Type          | Monthly Cost  |
|-----------------|------------------------|---------------|
| Kafka (MSK)     | kafka.m5.2xlarge x3    | ~$2,100       |
| Flink (EMR)     | m5.2xlarge x4          | ~$2,800       |
| Iceberg (S3)    | 50TB storage           | ~$1,150       |
| ClickHouse      | m5.4xlarge x3          | ~$4,200       |
| Redis           | r6g.2xlarge x2         | ~$1,800       |
| Airflow (MWAA)  | mw1.medium             | ~$350         |

Total estimated cost: ~$12,400/month

Cost optimization strategies:
1. Spot instances for Flink workers -> 60% savings
2. Iceberg compaction optimization -> 30% storage savings
3. ClickHouse TTL policies -> Auto-delete old data
4. Reserved Instances -> 40% savings for long-term workloads

12-Month Data Engineer Learning Path:

[Months 1-3: Foundations]
├── Advanced SQL (window functions, CTEs, optimization)
├── Python data processing (pandas, polars)
├── Linux / Docker basics
└── Git / CI/CD basics

[Months 4-6: Core Tools]
├── Apache Spark (DataFrame API, SparkSQL)
├── dbt (project structure, tests, documentation)
├── Airflow (DAG authoring, operators)
└── Kafka basics (producers, consumers, topics)

[Months 7-9: Advanced]
├── Apache Flink (stream processing, state management)
├── Iceberg / Delta Lake (lakehouse)
├── ClickHouse (OLAP analytics)
└── Great Expectations (data quality)

[Months 10-12: Production & Certification]
├── Complete 3 portfolio projects
├── Earn 1-2 certifications
├── Learn Data Mesh / governance
└── Start a technical blog

Tech Stack: Kafka + Flink + Iceberg + dbt + ClickHouse

Architecture:
NYC Taxi Data (API) -> Kafka -> Flink (real-time features) -> Iceberg
                                                                 |
                                                      dbt (batch transform)
                                                                 |
                                                      ClickHouse (dashboard)

Implementation highlights:
1. Real-time regional demand forecasting features with Flink
2. Iceberg partition evolution (daily -> hourly)
3. dbt incremental models for daily aggregation
4. ClickHouse Materialized Views for real-time dashboards
5. Airflow DAG orchestrating the entire pipeline

Tech Stack: Debezium + Kafka + Spark + Delta Lake + dbt

Architecture:
PostgreSQL (OLTP) -> Debezium (CDC) -> Kafka -> Spark Streaming -> Delta Lake
                                                                       |
                                                          dbt (layered transform)
                                                                       |
                                                        Superset (BI dashboard)

Implementation highlights:
1. Real-time PostgreSQL change capture with Debezium
2. CDC event processing with Spark Structured Streaming
3. SCD Type 2 implementation with Delta Lake MERGE
4. dbt staging/intermediate/marts structure
5. Superset real-time revenue/inventory dashboard

Tech Stack: OpenMetadata + Soda + dbt + Iceberg + Dagster

Architecture:
Domain A (Orders) ──> Data Contract (YAML) ──> Iceberg
Domain B (Customers) -> Data Contract (YAML) -> Iceberg
Domain C (Products) -> Data Contract (YAML) -> Iceberg
                                                   |
                              OpenMetadata (catalog/lineage)
                              Soda (quality validation)
                              Dagster (orchestration)

Implementation highlights:
1. Define Data Contract YAMLs for 3 domains
2. Automated data quality validation with Soda
3. Lineage and catalog management with OpenMetadata
4. Pipeline building with Dagster Software-Defined Assets
5. Enterprise metric layer (MetricFlow)