Kafka, Python & Structured Streaming

# Docker Compose pour Kafka avec ZooKeeper

docker_compose_kafka = '''
version: "3.8"
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.5.0
    container_name: zookeeper
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
    ports:
      - "2181:2181"

  kafka:
    image: confluentinc/cp-kafka:7.5.0
    container_name: kafka
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"      # Pour les clients externes
      - "29092:29092"    # Pour les clients Docker internes
    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 (optionnel mais recommandé)
  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
'''

print(docker_compose_kafka)
print("\n# Démarrer avec : docker-compose up -d")
print("# Vérifier : docker-compose ps")

# Commandes CLI Kafka essentielles

kafka_cli = '''
# ═══════════════════════════════════════════════════════════════
# Gestion des Topics
# ═══════════════════════════════════════════════════════════════

# Créer un topic
docker exec kafka kafka-topics --create \
  --bootstrap-server localhost:9092 \
  --topic orders \
  --partitions 3 \
  --replication-factor 1

# Lister les topics
docker exec kafka kafka-topics --list \
  --bootstrap-server localhost:9092

# Décrire un topic
docker exec kafka kafka-topics --describe \
  --bootstrap-server localhost:9092 \
  --topic orders

# ═══════════════════════════════════════════════════════════════
# Produire et Consommer (test rapide)
# ═══════════════════════════════════════════════════════════════

# Produire des messages (interactif)
docker exec -it kafka kafka-console-producer \
  --bootstrap-server localhost:9092 \
  --topic orders

# Consommer des messages (depuis le début)
docker exec kafka kafka-console-consumer \
  --bootstrap-server localhost:9092 \
  --topic orders \
  --from-beginning

# ═══════════════════════════════════════════════════════════════
# Consumer Groups
# ═══════════════════════════════════════════════════════════════

# Lister les consumer groups
docker exec kafka kafka-consumer-groups --list \
  --bootstrap-server localhost:9092

# Voir le lag d'un group
docker exec kafka kafka-consumer-groups --describe \
  --bootstrap-server localhost:9092 \
  --group my-consumer-group
'''

print(kafka_cli)

# Exemple de schéma Avro

avro_schema = '''
{
  "type": "record",
  "name": "Order",
  "namespace": "com.example",
  "fields": [
    {"name": "order_id", "type": "string"},
    {"name": "customer_id", "type": "string"},
    {"name": "amount", "type": "double"},
    {"name": "timestamp", "type": "long", "logicalType": "timestamp-millis"},
    {"name": "status", "type": {"type": "enum", "name": "Status", "symbols": ["PENDING", "COMPLETED", "CANCELLED"]}},
    {"name": "notes", "type": ["null", "string"], "default": null}  // Champ optionnel
  ]
}
'''

print("Schéma Avro pour les commandes :")
print(avro_schema)
print("\n💡 Points clés :")
print("• 'type': 'record' → structure comme une classe")
print("• 'logicalType' → types avancés (timestamp, date, decimal)")
print("• ['null', 'string'] → champ optionnel (union type)")

# Installation des librairies Kafka Python
# !pip install kafka-python confluent-kafka fastavro

print("Librairies à installer :")
print("pip install kafka-python       # Client simple")
print("pip install confluent-kafka    # Client performant + Schema Registry")
print("pip install fastavro           # Sérialisation Avro")

# Producteur Kafka simple avec kafka-python

producer_simple = '''
from kafka import KafkaProducer
import json
import time
from datetime import datetime

# ═══════════════════════════════════════════════════════════════
# Configuration du producteur
# ═══════════════════════════════════════════════════════════════

producer = KafkaProducer(
    bootstrap_servers=['localhost:9092'],
    
    # Sérialisation JSON
    value_serializer=lambda v: json.dumps(v).encode('utf-8'),
    key_serializer=lambda k: k.encode('utf-8') if k else None,
    
    # Configuration de fiabilité
    acks='all',              # Attendre confirmation de tous les replicas
    retries=3,               # Réessayer en cas d'erreur
    retry_backoff_ms=100,    # Délai entre retries
)

# ═══════════════════════════════════════════════════════════════
# Envoyer des messages
# ═══════════════════════════════════════════════════════════════

def send_order(order_id, customer_id, amount):
    """Envoyer une commande au topic 'orders'"""
    message = {
        "order_id": order_id,
        "customer_id": customer_id,
        "amount": amount,
        "timestamp": datetime.now().isoformat(),
        "status": "PENDING"
    }
    
    # Envoyer avec une clé (même customer → même partition → ordre garanti)
    future = producer.send(
        topic='orders',
        key=customer_id,    # Clé pour le partitionnement
        value=message
    )
    
    # Attendre confirmation (synchrone)
    try:
        record_metadata = future.get(timeout=10)
        print(f"✅ Message envoyé: partition={record_metadata.partition}, offset={record_metadata.offset}")
    except Exception as e:
        print(f"❌ Erreur: {e}")

# Simuler un flux de commandes
for i in range(10):
    send_order(
        order_id=f"ORD-{i:04d}",
        customer_id=f"CUST-{i % 3:03d}",  # 3 customers
        amount=round(100 + i * 10.5, 2)
    )
    time.sleep(0.5)

# Important : flush avant de quitter
producer.flush()
producer.close()
'''

print(producer_simple)

# Consommateur Kafka simple avec kafka-python

consumer_simple = '''
from kafka import KafkaConsumer
import json

# ═══════════════════════════════════════════════════════════════
# Configuration du consommateur
# ═══════════════════════════════════════════════════════════════

consumer = KafkaConsumer(
    'orders',                             # Topic(s) à consommer
    bootstrap_servers=['localhost:9092'],
    
    # Désérialisation JSON
    value_deserializer=lambda m: json.loads(m.decode('utf-8')),
    key_deserializer=lambda k: k.decode('utf-8') if k else None,
    
    # Consumer Group (pour le parallélisme)
    group_id='order-processor-group',
    
    # Où commencer à lire
    auto_offset_reset='earliest',  # 'earliest' = depuis le début, 'latest' = nouveaux messages seulement
    
    # Commit des offsets
    enable_auto_commit=True,       # Commit automatique
    auto_commit_interval_ms=5000,  # Toutes les 5 secondes
)

# ═══════════════════════════════════════════════════════════════
# Boucle de consommation
# ═══════════════════════════════════════════════════════════════

print("🎧 En attente de messages...")

try:
    for message in consumer:
        # Métadonnées du message
        print(f"\n📩 Message reçu:")
        print(f"   Topic: {message.topic}")
        print(f"   Partition: {message.partition}")
        print(f"   Offset: {message.offset}")
        print(f"   Key: {message.key}")
        print(f"   Timestamp: {message.timestamp}")
        
        # Contenu du message
        order = message.value
        print(f"   Order: {order}")
        
        # Logique métier : alerter si montant élevé
        if order.get('amount', 0) > 500:
            print(f"   ⚠️ ALERTE: Commande de {order['amount']}€ !")

except KeyboardInterrupt:
    print("\n🛑 Arrêt du consommateur")
finally:
    consumer.close()
'''

print(consumer_simple)

# Producteur avec Avro et Schema Registry

producer_avro = '''
from confluent_kafka import Producer
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroSerializer
from confluent_kafka.serialization import StringSerializer, SerializationContext, MessageField

# ═══════════════════════════════════════════════════════════════
# Configuration Schema Registry
# ═══════════════════════════════════════════════════════════════

schema_registry_conf = {'url': 'http://localhost:8081'}
schema_registry_client = SchemaRegistryClient(schema_registry_conf)

# Schéma Avro
order_schema = """
{
  "type": "record",
  "name": "Order",
  "fields": [
    {"name": "order_id", "type": "string"},
    {"name": "customer_id", "type": "string"},
    {"name": "amount", "type": "double"},
    {"name": "timestamp", "type": "long"}
  ]
}
"""

# Sérialiseurs
avro_serializer = AvroSerializer(
    schema_registry_client,
    order_schema,
    lambda obj, ctx: obj  # Conversion dict → Avro
)
string_serializer = StringSerializer('utf-8')

# ═══════════════════════════════════════════════════════════════
# Configuration producteur
# ═══════════════════════════════════════════════════════════════

producer_conf = {
    'bootstrap.servers': 'localhost:9092',
    'acks': 'all'
}
producer = Producer(producer_conf)

# Callback de confirmation
def delivery_report(err, msg):
    if err:
        print(f"❌ Erreur: {err}")
    else:
        print(f"✅ Message livré: {msg.topic()}[{msg.partition()}] @ {msg.offset()}")

# ═══════════════════════════════════════════════════════════════
# Envoyer des messages
# ═══════════════════════════════════════════════════════════════

import time

for i in range(5):
    order = {
        "order_id": f"ORD-{i:04d}",
        "customer_id": f"CUST-{i % 3:03d}",
        "amount": 100.0 + i * 25.5,
        "timestamp": int(time.time() * 1000)
    }
    
    producer.produce(
        topic='orders-avro',
        key=string_serializer(order["customer_id"]),
        value=avro_serializer(order, SerializationContext('orders-avro', MessageField.VALUE)),
        callback=delivery_report
    )
    producer.poll(0)  # Déclencher les callbacks

producer.flush()  # Attendre que tous les messages soient envoyés
'''

print(producer_avro)

# Exemple Faust (aperçu)

faust_example = '''
import faust

# Créer l'application Faust
app = faust.App(
    'order-processor',
    broker='kafka://localhost:9092',
    value_serializer='json'
)

# Définir le schéma du message
class Order(faust.Record):
    order_id: str
    customer_id: str
    amount: float

# Topic source
orders_topic = app.topic('orders', value_type=Order)

# Agent de traitement (comme un consumer intelligent)
@app.agent(orders_topic)
async def process_orders(orders):
    async for order in orders:
        print(f"Processing: {order.order_id}")
        
        # Logique métier
        if order.amount > 1000:
            print(f"⚠️ High-value order: {order.amount}")
            # Envoyer vers un autre topic
            await high_value_topic.send(value=order)

# Table pour agrégation (state)
order_counts = app.Table('order-counts', default=int)

@app.agent(orders_topic)
async def count_by_customer(orders):
    async for order in orders:
        order_counts[order.customer_id] += 1
        print(f"{order.customer_id}: {order_counts[order.customer_id]} orders")

# Lancer avec: faust -A myapp worker -l info
'''

print(faust_example)
print("\n💡 Quand utiliser Faust vs Spark SSS :")
print("• Faust : Traitement léger, alertes, routing, < 100K events/s")
print("• Spark SSS : Agrégations complexes, ML, joins, gros volumes")

# Configuration Spark pour le streaming avec Kafka

spark_streaming_config = '''
from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("Kafka Streaming Demo") \
    .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") \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
    .getOrCreate()

# Réduire les logs pour plus de clarté
spark.sparkContext.setLogLevel("WARN")
'''

print(spark_streaming_config)

# Lire un stream Kafka

read_kafka = '''
from pyspark.sql.functions import from_json, col
from pyspark.sql.types import StructType, StructField, StringType, DoubleType, TimestampType

# ═══════════════════════════════════════════════════════════════
# Lecture du stream Kafka
# ═══════════════════════════════════════════════════════════════

kafka_df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "orders") \
    .option("startingOffsets", "earliest") \
    .load()

# Le DataFrame brut contient ces colonnes :
# key (binary), value (binary), topic, partition, offset, timestamp, timestampType

print("Schéma Kafka brut:")
kafka_df.printSchema()

# ═══════════════════════════════════════════════════════════════
# Désérialisation JSON
# ═══════════════════════════════════════════════════════════════

# Définir le schéma du message JSON
order_schema = StructType([
    StructField("order_id", StringType(), True),
    StructField("customer_id", StringType(), True),
    StructField("amount", DoubleType(), True),
    StructField("timestamp", StringType(), True),
    StructField("status", StringType(), True)
])

# Parser le JSON
orders_df = kafka_df \
    .selectExpr("CAST(key AS STRING) as customer_key", "CAST(value AS STRING) as json_value") \
    .select(
        col("customer_key"),
        from_json(col("json_value"), order_schema).alias("data")
    ) \
    .select("customer_key", "data.*")

print("Schéma après parsing:")
orders_df.printSchema()
'''

print(read_kafka)

# Écrire le stream (différents sinks)

write_stream = '''
# ═══════════════════════════════════════════════════════════════
# Sink Console (pour debug)
# ═══════════════════════════════════════════════════════════════

query_console = orders_df.writeStream \
    .format("console") \
    .outputMode("append") \
    .option("truncate", False) \
    .start()

# ═══════════════════════════════════════════════════════════════
# Sink Parquet/Delta avec Checkpointing
# ═══════════════════════════════════════════════════════════════

query_delta = orders_df.writeStream \
    .format("delta") \
    .outputMode("append") \
    .option("checkpointLocation", "/tmp/checkpoints/orders") \
    .option("path", "s3a://silver/orders_streaming/") \
    .start()

# ═══════════════════════════════════════════════════════════════
# Sink Kafka (pour pipeline)
# ═══════════════════════════════════════════════════════════════

query_kafka = orders_df \
    .selectExpr("customer_id AS key", "to_json(struct(*)) AS value") \
    .writeStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("topic", "orders-processed") \
    .option("checkpointLocation", "/tmp/checkpoints/orders-kafka") \
    .start()

# Attendre la terminaison
query_console.awaitTermination()
'''

print(write_stream)
print("\n💡 Le checkpointLocation est CRUCIAL pour :")
print("• Reprendre après un crash (exactly-once)")
print("• Stocker l'état des agrégations")
print("• Suivre les offsets Kafka")

# Windowing avec Spark SSS

windowing_example = '''
from pyspark.sql.functions import window, col, count, sum as spark_sum, to_timestamp

# ═══════════════════════════════════════════════════════════════
# Préparer le timestamp d'événement
# ═══════════════════════════════════════════════════════════════

orders_with_ts = orders_df \
    .withColumn("event_time", to_timestamp(col("timestamp")))

# ═══════════════════════════════════════════════════════════════
# Tumbling Window : Agrégation par fenêtre de 5 minutes
# ═══════════════════════════════════════════════════════════════

tumbling_agg = orders_with_ts \
    .groupBy(
        window(col("event_time"), "5 minutes"),  # Fenêtre de 5 min
        col("customer_id")
    ) \
    .agg(
        count("*").alias("order_count"),
        spark_sum("amount").alias("total_amount")
    )

# ═══════════════════════════════════════════════════════════════
# Sliding Window : Fenêtre de 10 min, glissant toutes les 5 min
# ═══════════════════════════════════════════════════════════════

sliding_agg = orders_with_ts \
    .groupBy(
        window(col("event_time"), "10 minutes", "5 minutes"),  # 10 min, slide 5 min
        col("customer_id")
    ) \
    .agg(
        count("*").alias("order_count")
    )

# Le résultat contient une colonne "window" avec start/end
'''

print(windowing_example)

# Watermarks avec Spark SSS

watermark_example = '''
from pyspark.sql.functions import window, col, count, to_timestamp

# ═══════════════════════════════════════════════════════════════
# Agrégation avec Watermark
# ═══════════════════════════════════════════════════════════════

# Le watermark dit : "Je tolère jusqu'à 10 minutes de retard"
# Après 10 min, les données en retard sont ignorées et l'état nettoyé

windowed_counts = orders_with_ts \
    .withWatermark("event_time", "10 minutes") \  # ← Crucial !
    .groupBy(
        window(col("event_time"), "5 minutes"),
        col("customer_id")
    ) \
    .count()

# ═══════════════════════════════════════════════════════════════
# Comment ça marche ?
# ═══════════════════════════════════════════════════════════════
#
# 1. Spark suit le "max event time" vu jusqu'ici
# 2. Watermark = max_event_time - 10 minutes
# 3. Les fenêtres dont window.end < watermark sont finalisées
# 4. Les données avec event_time < watermark sont ignorées
#
# Exemple :
# - Max event time vu : 10:30
# - Watermark : 10:20
# - Fenêtre 10:00-10:05 : FINALISÉE (end 10:05 < 10:20)
# - Message avec event_time 10:18 : ACCEPTÉ
# - Message avec event_time 10:15 : IGNORÉ (< watermark)

# Écrire en mode Update (pour les agrégations)
query = windowed_counts.writeStream \
    .format("console") \
    .outputMode("update") \
    .option("truncate", False) \
    .start()
'''

print(watermark_example)
print("\n💡 Choisir le bon watermark :")
print("• Trop court (1 min) → Perd des données en retard")
print("• Trop long (1 heure) → Trop de mémoire utilisée")
print("• Règle : Analyser le retard typique de tes données")

# Streaming Joins

join_examples = '''
# ═══════════════════════════════════════════════════════════════
# Stream-Static Join : Enrichir avec une dimension
# ═══════════════════════════════════════════════════════════════

# Table statique (dimension)
customers_df = spark.read.parquet("s3a://dims/customers/")

# Stream
orders_stream = spark.readStream.format("kafka")...

# Join
enriched_orders = orders_stream.join(
    customers_df,
    orders_stream.customer_id == customers_df.id,
    "left"  # LEFT JOIN pour garder toutes les commandes
)

# ═══════════════════════════════════════════════════════════════
# Stream-Stream Join : Deux flux
# ═══════════════════════════════════════════════════════════════

# Deux streams
clicks_stream = spark.readStream.format("kafka").option("subscribe", "clicks")...
impressions_stream = spark.readStream.format("kafka").option("subscribe", "impressions")...

# Join avec watermark (obligatoire pour stream-stream)
clicks_with_wm = clicks_stream \
    .withWatermark("click_time", "10 minutes")

impressions_with_wm = impressions_stream \
    .withWatermark("impression_time", "10 minutes")

# Join avec condition de temps
matched = clicks_with_wm.join(
    impressions_with_wm,
    expr("""
        clicks.ad_id = impressions.ad_id AND
        click_time >= impression_time AND
        click_time <= impression_time + interval 1 hour
    """),
    "inner"
)
'''

print(join_examples)

# foreachBatch avec MERGE INTO Delta

foreach_batch_example = '''
from delta.tables import DeltaTable

# ═══════════════════════════════════════════════════════════════
# Fonction de traitement par micro-batch
# ═══════════════════════════════════════════════════════════════

def upsert_to_delta(micro_batch_df, batch_id):
    """
    Appelée pour chaque micro-batch.
    micro_batch_df : DataFrame Spark normal (pas streaming)
    batch_id : Identifiant unique du batch
    """
    print(f"Processing batch {batch_id} with {micro_batch_df.count()} records")
    
    # Vérifier si la table existe
    if DeltaTable.isDeltaTable(spark, "s3a://silver/customers/"):
        # Table existe → MERGE (upsert)
        target = DeltaTable.forPath(spark, "s3a://silver/customers/")
        
        target.alias("target").merge(
            micro_batch_df.alias("source"),
            "target.customer_id = source.customer_id"
        ).whenMatchedUpdate(
            set={
                "name": "source.name",
                "email": "source.email",
                "updated_at": "current_timestamp()"
            }
        ).whenNotMatchedInsert(
            values={
                "customer_id": "source.customer_id",
                "name": "source.name",
                "email": "source.email",
                "created_at": "current_timestamp()",
                "updated_at": "current_timestamp()"
            }
        ).execute()
    else:
        # Table n'existe pas → CREATE
        micro_batch_df.write \
            .format("delta") \
            .mode("overwrite") \
            .save("s3a://silver/customers/")

# ═══════════════════════════════════════════════════════════════
# Utiliser foreachBatch
# ═══════════════════════════════════════════════════════════════

query = customers_stream.writeStream \
    .foreachBatch(upsert_to_delta) \
    .option("checkpointLocation", "/tmp/checkpoints/customers-upsert") \
    .outputMode("update") \
    .start()
'''

print(foreach_batch_example)
print("\n💡 Avantages de foreachBatch :")
print("• Utiliser MERGE INTO (upserts)")
print("• Écrire vers plusieurs sinks")
print("• Appliquer n'importe quelle logique batch")
print("• Exactly-once avec checkpointing")

# Exemple de configuration Debezium

debezium_config = '''
{
  "name": "postgres-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "secret",
    "database.dbname": "inventory",
    "database.server.name": "dbserver1",
    "table.include.list": "public.customers,public.orders",
    "plugin.name": "pgoutput",
    
    "transforms": "unwrap",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
    "transforms.unwrap.drop.tombstones": "false"
  }
}
'''

print("Configuration Debezium pour PostgreSQL :")
print(debezium_config)
print("\n💡 Quand utiliser Debezium vs code custom :")
print("• Debezium : CDC standard depuis DB, pas de code à maintenir")
print("• Code custom : Logique complexe, sources non supportées")

# SparkApplication pour un job streaming

spark_streaming_k8s = '''
apiVersion: sparkoperator.k8s.io/v1beta2
kind: SparkApplication
metadata:
  name: kafka-to-delta-streaming
  namespace: spark
spec:
  type: Python
  pythonVersion: "3"
  mode: cluster
  image: my-registry/spark-streaming:latest
  mainApplicationFile: s3a://code/streaming_job.py
  sparkVersion: "3.5.0"
  
  # Important pour le streaming !
  restartPolicy:
    type: Always  # Redémarrer automatiquement si crash
    onFailureRetries: 3
    onFailureRetryInterval: 60
  
  driver:
    cores: 1
    memory: "2g"
  
  executor:
    cores: 2
    instances: 3
    memory: "4g"
  
  deps:
    packages:
      - org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.0
      - io.delta:delta-spark_2.12:3.1.0
'''

print(spark_streaming_k8s)

# Monitoring du streaming

monitoring = '''
# ═══════════════════════════════════════════════════════════════
# Accéder aux métriques dans le code
# ═══════════════════════════════════════════════════════════════

query = df.writeStream...

# Progression du dernier batch
print(query.lastProgress)

# Status actuel
print(query.status)

# Exemple de lastProgress :
# {
#   "id": "abc123",
#   "runId": "def456",
#   "batchId": 42,
#   "numInputRows": 1000,
#   "inputRowsPerSecond": 500.0,
#   "processedRowsPerSecond": 450.0,
#   "durationMs": {
#     "triggerExecution": 2000,
#     "getBatch": 100,
#     "queryPlanning": 50
#   },
#   "eventTime": {
#     "watermark": "2024-01-15T10:20:00.000Z"
#   }
# }

# ═══════════════════════════════════════════════════════════════
# Listener pour métriques custom
# ═══════════════════════════════════════════════════════════════

class MetricsListener(StreamingQueryListener):
    def onQueryStarted(self, event):
        print(f"Query started: {event.id}")
    
    def onQueryProgress(self, event):
        # Envoyer vers Prometheus/Grafana
        metrics.gauge("streaming_input_rate", event.progress.inputRowsPerSecond)
        metrics.gauge("streaming_batch_duration", event.progress.durationMs["triggerExecution"])
    
    def onQueryTerminated(self, event):
        print(f"Query terminated: {event.id}")

spark.streams.addListener(MetricsListener())
'''

print(monitoring)

# ÉTAPE 1 : Producteur Python

producer_code = '''
from kafka import KafkaProducer
import json
import time
import random
from datetime import datetime

producer = KafkaProducer(
    bootstrap_servers=["localhost:9092"],
    value_serializer=lambda v: json.dumps(v).encode("utf-8")
)

customers = ["CUST-001", "CUST-002", "CUST-003", "CUST-004", "CUST-005"]
products = ["Laptop", "Phone", "Tablet", "Watch", "Headphones"]

print("🚀 Sending transactions...")

for i in range(100):
    transaction = {
        "transaction_id": f"TXN-{i:06d}",
        "customer_id": random.choice(customers),
        "product": random.choice(products),
        "amount": round(random.uniform(10, 500), 2),
        "timestamp": datetime.now().isoformat()
    }
    
    producer.send("transactions", value=transaction)
    print(f"Sent: {transaction[\'transaction_id\']}")
    time.sleep(0.5)  # Simuler un flux

producer.flush()
print("✅ Done!")
'''

print("# producer.py")
print(producer_code)

# ÉTAPE 2 : Job Spark Streaming

streaming_job = '''
from pyspark.sql import SparkSession
from pyspark.sql.functions import from_json, col, window, sum as spark_sum, count, to_timestamp
from pyspark.sql.types import StructType, StructField, StringType, DoubleType
from delta.tables import DeltaTable

# Spark Session
spark = SparkSession.builder \
    .appName("Realtime Transactions") \
    .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") \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .getOrCreate()

# Schéma
schema = StructType([
    StructField("transaction_id", StringType()),
    StructField("customer_id", StringType()),
    StructField("product", StringType()),
    StructField("amount", DoubleType()),
    StructField("timestamp", StringType())
])

# Lire Kafka
raw_stream = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "transactions") \
    .option("startingOffsets", "earliest") \
    .load()

# Parser JSON
transactions = raw_stream \
    .selectExpr("CAST(value AS STRING)") \
    .select(from_json(col("value"), schema).alias("data")) \
    .select("data.*") \
    .withColumn("event_time", to_timestamp(col("timestamp")))

# Agrégation par fenêtre de 5 min avec watermark 10 min
windowed_stats = transactions \
    .withWatermark("event_time", "10 minutes") \
    .groupBy(
        window(col("event_time"), "5 minutes"),
        col("customer_id")
    ) \
    .agg(
        count("*").alias("transaction_count"),
        spark_sum("amount").alias("total_amount")
    )

# Fonction pour upsert vers Delta
def upsert_to_delta(batch_df, batch_id):
    # Flatten window column
    flat_df = batch_df.selectExpr(
        "window.start as window_start",
        "window.end as window_end",
        "customer_id",
        "transaction_count",
        "total_amount"
    )
    
    if flat_df.count() == 0:
        return
    
    target_path = "/tmp/delta/customer_stats"
    
    if DeltaTable.isDeltaTable(spark, target_path):
        target = DeltaTable.forPath(spark, target_path)
        target.alias("t").merge(
            flat_df.alias("s"),
            "t.window_start = s.window_start AND t.customer_id = s.customer_id"
        ).whenMatchedUpdate(set={
            "transaction_count": "s.transaction_count",
            "total_amount": "s.total_amount"
        }).whenNotMatchedInsertAll().execute()
    else:
        flat_df.write.format("delta").save(target_path)
    
    print(f"Batch {batch_id}: Processed {flat_df.count()} records")

# Lancer le stream
query = windowed_stats.writeStream \
    .foreachBatch(upsert_to_delta) \
    .option("checkpointLocation", "/tmp/checkpoints/customer_stats") \
    .outputMode("update") \
    .trigger(processingTime="30 seconds") \
    .start()

query.awaitTermination()
'''

print("# streaming_job.py")
print(streaming_job)