Projet Intégrateur : Pipeline E-commerce Olist

docker-compose.yml

version: '3.8'

services:
  # ═══════════════════════════════════════════════════════════════
  # KAFKA
  # ═══════════════════════════════════════════════════════════════
  zookeeper:
    image: confluentinc/cp-zookeeper:7.5.0
    container_name: zookeeper
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
    ports:
      - "2181:2181"

  kafka:
    image: confluentinc/cp-kafka:7.5.0
    container_name: kafka
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
      - "29092:29092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true"

  schema-registry:
    image: confluentinc/cp-schema-registry:7.5.0
    container_name: schema-registry
    depends_on:
      - kafka
    ports:
      - "8081:8081"
    environment:
      SCHEMA_REGISTRY_HOST_NAME: schema-registry
      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: kafka:29092

  # ═══════════════════════════════════════════════════════════════
  # STORAGE (MinIO = S3 local)
  # ═══════════════════════════════════════════════════════════════
  minio:
    image: minio/minio
    container_name: minio
    ports:
      - "9000:9000"
      - "9001:9001"
    environment:
      MINIO_ROOT_USER: minioadmin
      MINIO_ROOT_PASSWORD: minioadmin
    command: server /data --console-address ":9001"
    volumes:
      - minio_data:/data

  # Créer le bucket au démarrage
  minio-setup:
    image: minio/mc
    depends_on:
      - minio
    entrypoint: >
      /bin/sh -c "
      sleep 5;
      mc alias set myminio http://minio:9000 minioadmin minioadmin;
      mc mb myminio/lakehouse --ignore-existing;
      exit 0;
      "

  # ═══════════════════════════════════════════════════════════════
  # SPARK
  # ═══════════════════════════════════════════════════════════════
  spark-master:
    image: bitnami/spark:3.5
    container_name: spark-master
    environment:
      - SPARK_MODE=master
      - SPARK_RPC_AUTHENTICATION_ENABLED=no
      - SPARK_RPC_ENCRYPTION_ENABLED=no
    ports:
      - "8080:8080"
      - "7077:7077"
    volumes:
      - ./spark_jobs:/opt/spark_jobs
      - ./data:/opt/data

  spark-worker:
    image: bitnami/spark:3.5
    container_name: spark-worker
    depends_on:
      - spark-master
    environment:
      - SPARK_MODE=worker
      - SPARK_MASTER_URL=spark://spark-master:7077
      - SPARK_WORKER_MEMORY=2G
      - SPARK_WORKER_CORES=2
    volumes:
      - ./spark_jobs:/opt/spark_jobs
      - ./data:/opt/data

  # ═══════════════════════════════════════════════════════════════
  # AIRFLOW
  # ═══════════════════════════════════════════════════════════════
  postgres:
    image: postgres:15
    container_name: postgres
    environment:
      POSTGRES_USER: airflow
      POSTGRES_PASSWORD: airflow
      POSTGRES_DB: airflow
    ports:
      - "5432:5432"
    volumes:
      - postgres_data:/var/lib/postgresql/data

  airflow-webserver:
    image: apache/airflow:2.8.0
    container_name: airflow-webserver
    depends_on:
      - postgres
    environment:
      AIRFLOW__CORE__EXECUTOR: LocalExecutor
      AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres/airflow
      AIRFLOW__CORE__FERNET_KEY: ''
      AIRFLOW__CORE__LOAD_EXAMPLES: 'false'
    ports:
      - "8082:8080"
    volumes:
      - ./dags:/opt/airflow/dags
      - ./dbt_olist:/opt/dbt
    command: bash -c "airflow db init && airflow users create --username admin --password admin --firstname Admin --lastname User --role Admin --email admin@example.com && airflow webserver"

  airflow-scheduler:
    image: apache/airflow:2.8.0
    container_name: airflow-scheduler
    depends_on:
      - airflow-webserver
    environment:
      AIRFLOW__CORE__EXECUTOR: LocalExecutor
      AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres/airflow
    volumes:
      - ./dags:/opt/airflow/dags
      - ./dbt_olist:/opt/dbt
    command: airflow scheduler

volumes:
  minio_data:
  postgres_data:

producers/orders_producer.py

from kafka import KafkaProducer
import pandas as pd
import json
import time
import random
from datetime import datetime, timedelta

# Configuration
KAFKA_BOOTSTRAP_SERVERS = ['localhost:9092']
TOPIC = 'raw_orders'
CSV_PATH = 'data/olist_orders_dataset.csv'

# Producteur Kafka
producer = KafkaProducer(
    bootstrap_servers=KAFKA_BOOTSTRAP_SERVERS,
    value_serializer=lambda v: json.dumps(v, default=str).encode('utf-8'),
    key_serializer=lambda k: k.encode('utf-8') if k else None
)

# Lire le CSV
df = pd.read_csv(CSV_PATH)
print(f"Loaded {len(df)} orders")

# Envoyer les messages
for idx, row in df.iterrows():
    message = row.to_dict()
    
    # Timestamp d'ingestion
    ingestion_ts = datetime.now()
    
    # Simuler late data (5%)
    if random.random() < 0.05:
        ingestion_ts -= timedelta(minutes=random.randint(1, 5))
    
    message['_ingestion_timestamp'] = ingestion_ts.isoformat()
    
    # Envoyer
    producer.send(
        topic=TOPIC,
        key=message['order_id'],
        value=message
    )
    
    # Simuler doublons (2%)
    if random.random() < 0.02:
        producer.send(topic=TOPIC, key=message['order_id'], value=message)
    
    # Log progress
    if idx % 1000 == 0:
        print(f"Sent {idx}/{len(df)} messages")
    
    # Délai aléatoire (simuler streaming)
    time.sleep(random.uniform(0.05, 0.2))

producer.flush()
print(f"Done! Sent {len(df)} messages to {TOPIC}")

spark_jobs/bronze/ingest_orders.py

from pyspark.sql import SparkSession
from pyspark.sql.functions import from_json, col, current_timestamp, to_date
from pyspark.sql.types import StructType, StructField, StringType

# Spark Session
spark = SparkSession.builder \
    .appName("Bronze - Ingest Orders") \
    .config("spark.jars.packages", 
            "org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.0,"
            "io.delta:delta-spark_2.12:3.1.0,"
            "org.apache.hadoop:hadoop-aws:3.3.4") \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
    .config("spark.hadoop.fs.s3a.endpoint", "http://minio:9000") \
    .config("spark.hadoop.fs.s3a.access.key", "minioadmin") \
    .config("spark.hadoop.fs.s3a.secret.key", "minioadmin") \
    .config("spark.hadoop.fs.s3a.path.style.access", "true") \
    .getOrCreate()

# Schema des orders
order_schema = StructType([
    StructField("order_id", StringType()),
    StructField("customer_id", StringType()),
    StructField("order_status", StringType()),
    StructField("order_purchase_timestamp", StringType()),
    StructField("order_approved_at", StringType()),
    StructField("order_delivered_carrier_date", StringType()),
    StructField("order_delivered_customer_date", StringType()),
    StructField("order_estimated_delivery_date", StringType()),
    StructField("_ingestion_timestamp", StringType())
])

# Lire depuis Kafka
kafka_df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "kafka:29092") \
    .option("subscribe", "raw_orders") \
    .option("startingOffsets", "earliest") \
    .load()

# Parser le JSON
parsed_df = kafka_df \
    .selectExpr("CAST(value AS STRING) as json_value") \
    .select(from_json(col("json_value"), order_schema).alias("data")) \
    .select("data.*") \
    .withColumn("_bronze_ingested_at", current_timestamp()) \
    .withColumn("_ingestion_date", to_date(col("_ingestion_timestamp")))

# Écrire en Bronze (Append)
query = parsed_df.writeStream \
    .format("delta") \
    .outputMode("append") \
    .option("checkpointLocation", "s3a://lakehouse/checkpoints/bronze_orders") \
    .option("path", "s3a://lakehouse/bronze/orders") \
    .partitionBy("_ingestion_date") \
    .start()

query.awaitTermination()

spark_jobs/silver/silver_orders.py

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, current_timestamp, row_number
from pyspark.sql.window import Window
from delta.tables import DeltaTable

# Spark Session (même config que Bronze)
spark = SparkSession.builder \
    .appName("Silver - Orders MERGE") \
    .config("spark.jars.packages", 
            "org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.0,"
            "io.delta:delta-spark_2.12:3.1.0,"
            "org.apache.hadoop:hadoop-aws:3.3.4") \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
    .config("spark.hadoop.fs.s3a.endpoint", "http://minio:9000") \
    .config("spark.hadoop.fs.s3a.access.key", "minioadmin") \
    .config("spark.hadoop.fs.s3a.secret.key", "minioadmin") \
    .config("spark.hadoop.fs.s3a.path.style.access", "true") \
    .getOrCreate()

BRONZE_PATH = "s3a://lakehouse/bronze/orders"
SILVER_PATH = "s3a://lakehouse/silver/orders"
CHECKPOINT_PATH = "s3a://lakehouse/checkpoints/silver_orders"

def upsert_to_silver(batch_df, batch_id):
    """Upsert vers Silver avec déduplication."""
    
    if batch_df.count() == 0:
        print(f"Batch {batch_id}: No data")
        return
    
    # 1. Déduplication (garder le plus récent par order_id)
    window = Window.partitionBy("order_id").orderBy(col("_bronze_ingested_at").desc())
    deduped = batch_df \
        .withColumn("_row_num", row_number().over(window)) \
        .filter(col("_row_num") == 1) \
        .drop("_row_num")
    
    # 2. Ajouter timestamp Silver
    enriched = deduped.withColumn("_silver_updated_at", current_timestamp())
    
    # 3. MERGE INTO
    if DeltaTable.isDeltaTable(spark, SILVER_PATH):
        delta_table = DeltaTable.forPath(spark, SILVER_PATH)
        
        delta_table.alias("target").merge(
            enriched.alias("source"),
            "target.order_id = source.order_id"
        ).whenMatchedUpdate(
            condition="source._bronze_ingested_at > target._bronze_ingested_at",
            set={
                "order_status": "source.order_status",
                "order_delivered_carrier_date": "source.order_delivered_carrier_date",
                "order_delivered_customer_date": "source.order_delivered_customer_date",
                "_bronze_ingested_at": "source._bronze_ingested_at",
                "_silver_updated_at": "source._silver_updated_at"
            }
        ).whenNotMatchedInsertAll().execute()
        
        print(f"Batch {batch_id}: MERGED {enriched.count()} records")
    else:
        # Première exécution
        enriched.write.format("delta").mode("overwrite").save(SILVER_PATH)
        print(f"Batch {batch_id}: CREATED table with {enriched.count()} records")

# Lire Bronze en streaming
bronze_stream = spark.readStream \
    .format("delta") \
    .load(BRONZE_PATH)

# Écrire avec foreachBatch
query = bronze_stream.writeStream \
    .foreachBatch(upsert_to_silver) \
    .option("checkpointLocation", CHECKPOINT_PATH) \
    .trigger(processingTime="30 seconds") \
    .start()

query.awaitTermination()

dbt_olist/models/gold/gold_daily_sales.sql

{{ config(
    materialized='incremental',
    unique_key='order_date',
    incremental_strategy='merge'
) }}

WITH orders AS (
    SELECT
        o.order_id,
        o.order_status,
        DATE(o.order_purchase_timestamp) AS order_date,
        oi.price,
        oi.freight_value
    FROM {{ ref('stg_orders') }} o
    JOIN {{ ref('stg_order_items') }} oi ON o.order_id = oi.order_id
    WHERE o.order_status = 'delivered'
    {% if is_incremental() %}
    AND DATE(o.order_purchase_timestamp) >= (
        SELECT MAX(order_date) - INTERVAL 2 DAY FROM {{ this }}
    )
    {% endif %}
)

SELECT
    order_date,
    COUNT(DISTINCT order_id) AS total_orders,
    SUM(price + freight_value) AS total_revenue,
    AVG(price + freight_value) AS avg_order_value,
    COUNT(*) AS total_items
FROM orders
GROUP BY order_date
ORDER BY order_date

dbt_olist/models/gold/gold_customer_rfm.sql

{{ config(materialized='table') }}

WITH customer_orders AS (
    SELECT
        c.customer_unique_id,
        o.order_id,
        o.order_purchase_timestamp,
        oi.price + oi.freight_value AS order_value
    FROM {{ ref('stg_customers') }} c
    JOIN {{ ref('stg_orders') }} o ON c.customer_id = o.customer_id
    JOIN {{ ref('stg_order_items') }} oi ON o.order_id = oi.order_id
    WHERE o.order_status = 'delivered'
),

rfm_base AS (
    SELECT
        customer_unique_id,
        DATEDIFF(day, MAX(order_purchase_timestamp), CURRENT_DATE) AS recency_days,
        COUNT(DISTINCT order_id) AS frequency,
        SUM(order_value) AS monetary
    FROM customer_orders
    GROUP BY customer_unique_id
)

SELECT
    customer_unique_id,
    recency_days,
    frequency,
    ROUND(monetary, 2) AS monetary,
    {{ rfm_segment('recency_days', 'frequency', 'monetary') }} AS rfm_segment
FROM rfm_base

dbt_olist/models/gold/_gold__models.yml

version: 2

models:
  - name: gold_daily_sales
    description: "Chiffre d'affaires quotidien"
    columns:
      - name: order_date
        tests:
          - unique
          - not_null
      - name: total_revenue
        tests:
          - not_null
          - dbt_expectations.expect_column_values_to_be_between:
              min_value: 0

  - name: gold_customer_rfm
    description: "Segmentation RFM des clients"
    columns:
      - name: customer_unique_id
        tests:
          - unique
          - not_null
      - name: rfm_segment
        tests:
          - accepted_values:
              values: ['Champions', 'Loyal Customers', 'Potential Loyalists', 'At Risk', 'Hibernating', 'Others']

  - name: gold_seller_performance
    description: "Performance des vendeurs"
    columns:
      - name: seller_id
        tests:
          - unique
          - not_null

  - name: gold_product_analytics
    description: "Analyse des produits par catégorie"
    columns:
      - name: product_category
        tests:
          - unique
          - not_null

  - name: gold_delivery_performance
    description: "Performance des livraisons"
    columns:
      - name: on_time_rate
        tests:
          - dbt_expectations.expect_column_values_to_be_between:
              min_value: 0
              max_value: 1