Pipeline ML Temps Réel pour l’Analyse Boursière

.gitignore

@@ -0,0 +1,22 @@
+# Python
+__pycache__/
+*.pyc
+.venv/
+.env
+
+# Logs
+*.log
+/tmp/
+
+# OS files
+.DS_Store
+
+# IDEs
+.vscode/
+.idea/
+
+# Dataset
+data-ingestion-kafka/data/creditcard.csv
+
+package-lock.json 
+package-lock.json

docker-compose.yml

@@ -1,27 +1,34 @@
-version: "3.8"
 
 services:
   zookeeper:
     image: confluentinc/cp-zookeeper:7.5.0
+    env_file:
+      - .env
     environment:
       ZOOKEEPER_CLIENT_PORT: ${ZOOKEEPER_CLIENT_PORT}
     ports:
       - "${ZOOKEEPER_CLIENT_PORT}:${ZOOKEEPER_CLIENT_PORT}"
 
   kafka:
     image: confluentinc/cp-kafka:7.5.0
+    env_file:
+      - .env
     depends_on:
       - zookeeper
     ports:
       - "${KAFKA_PORT}:${KAFKA_PORT}"
     environment:
       KAFKA_BROKER_ID: ${KAFKA_BROKER_ID}
       KAFKA_ZOOKEEPER_CONNECT: ${KAFKA_ZOOKEEPER_CONNECT}
-      KAFKA_ADVERTISED_LISTENERS: ${KAFKA_ADVERTISED_LISTENER}
+      KAFKA_ADVERTISED_LISTENERS: ${KAFKA_ADVERTISED_LISTENERS}
+      KAFKA_LISTENERS: ${KAFKA_LISTENERS:-PLAINTEXT://0.0.0.0:9092}  # Ajout de KAFKA_LISTENERS avec valeur par défaut
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: ${KAFKA_LISTENER_SECURITY_PROTOCOL_MAP:-PLAINTEXT:PLAINTEXT}  # Ajout pour mapper les protocoles
       KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
 
   spark:
     image: bitnami/spark:latest
+    env_file:
+      - .env
     environment:
       - SPARK_MODE=master
     ports:
@@ -30,8 +37,10 @@ services:
 
   superset:
     image: apache/superset
+    env_file:
+      - .env
     ports:
-      - ${SUPERSET_PORT}:${SUPERSET_PORT}
+      - "${SUPERSET_PORT}:${SUPERSET_PORT}"
     environment:
       SUPERSET_SECRET_KEY: ${SUPERSET_SECRET_KEY}
     depends_on:
@@ -41,10 +50,12 @@ services:
         superset db upgrade &&
         superset fab create-admin --username ${SUPERSET_ADMIN_USERNAME} --firstname Admin --lastname User --email ${SUPERSET_ADMIN_EMAIL} --password ${SUPERSET_ADMIN_PASSWORD} &&
         superset init &&
-        superset run -h 0.0.0.0 -p 8088"
+        superset run -h 0.0.0.0 -p ${SUPERSET_PORT}"
 
   postgres:
     image: postgres:15
+    env_file:
+      - .env
     environment:
       POSTGRES_DB: ${POSTGRES_DB}
       POSTGRES_USER: ${POSTGRES_USER}

kafka-test/README.md

@@ -0,0 +1,79 @@
+# Kafka Ingestion Demo – FinanceLake
+
+This directory contains a working demo of a real-time ingestion pipeline using **Apache Kafka**, as part of the FinanceLake project.
+
+## ⚙️ Architecture
+
+- Kafka & Zookeeper run locally using Docker Compose (defined at the project root)
+- A Python producer sends messages to a Kafka topic
+- A Python consumer reads those messages from the topic
+- Configuration is handled via environment variables loaded with `python-dotenv` in a `config.py` file
+
+## 🧱 Prerequisites
+
+- Docker installed  
+- Python 3 installed  
+- Required packages: `kafka-python` and `python-dotenv`  
+- Port `9092` available for Kafka  
+
+```bash
+pip install kafka-python python-dotenv
+```
+
+## 🚀 Getting Started
+
+### 1. Start Kafka & Zookeeper services
+
+From the project root, run:
+
+```bash
+docker-compose up -d
+```
+
+### 2. Check if the containers are running
+
+```bash
+docker ps
+```
+You should see `kafka-test-kafka-1` and `kafka-test-zookeeper-1` in the output.
+
+### 3. Run the Kafka producer
+
+```bash
+python kafka_producer.py
+```
+This script sends sample messages to the Kafka topic (e.g., `test-topic`).
+
+### 4. Run the Kafka consumer
+
+```bash
+python kafka_consumer.py
+```
+This script reads and displays the messages from the Kafka topic.
+
+### 5. Stop the services
+
+```bash
+docker-compose down
+```
+
+## 📁 Project Structure
+
+```
+financeLake/
+├── .env.example           # Example environment file to configure variables
+├── docker-compose.yml     # Docker Compose (Kafka + Zookeeper) - at the root
+└── kafka-test/
+    ├── resources/         # Screenshots and demo video
+    │   ├── docker_ps.png
+    │   ├── kafka_producer.png
+    │   └── kafka_consumer.png
+    ├── config.py          # Loads environment variables
+    ├── kafka_producer.py  # Kafka producer script
+    ├── kafka_consumer.py  # Kafka consumer script
+    └── README.md          # This file
+```
+
+## 🔐 Notes
+
+Make sure to copy `.env.example` to `.env` and update it with your real values when testing locally. The `.env` file should **not** be committed to version control.

kafka-test/config.py

@@ -0,0 +1,19 @@
+from dotenv import load_dotenv
+import os
+
+# Charge automatiquement le fichier .env à la racine du projet
+load_dotenv()
+
+# Récupération des variables d'environnement
+KAFKA_BROKER = os.getenv("KAFKA_BROKER")
+KAFKA_TOPIC = os.getenv("KAFKA_TOPIC")
+POSTGRES_USER = os.getenv("POSTGRES_USER")
+POSTGRES_PASSWORD = os.getenv("POSTGRES_PASSWORD")
+POSTGRES_DB = os.getenv("POSTGRES_DB")
+
+# Tu peux ajouter toutes les variables utiles ici
+
+if __name__ == "__main__":
+    # Test rapide pour vérifier que ça marche
+    print(f"KAFKA_BROKER = {KAFKA_BROKER}")
+    print(f"KAFKA_TOPIC = {KAFKA_TOPIC}")

kafka-test/kafka_consumer.py

@@ -0,0 +1,19 @@
+from kafka import KafkaConsumer
+import json
+from config import KAFKA_BROKER, KAFKA_TOPIC
+
+# Connexion au broker Kafka via les variables d'environnement
+consumer = KafkaConsumer(
+    KAFKA_TOPIC,
+    bootstrap_servers=KAFKA_BROKER,
+    auto_offset_reset='earliest',
+    enable_auto_commit=True,
+    group_id='finance-group',
+    value_deserializer=lambda x: json.loads(x.decode('utf-8'))
+)
+
+print(f"🎧 Listening to topic '{KAFKA_TOPIC}' on broker '{KAFKA_BROKER}'...")
+
+# Lire les messages
+for message in consumer:
+    print(f"📨 Message reçu: {message.value}")

kafka-test/kafka_producer.py

@@ -0,0 +1,17 @@
+from kafka import KafkaProducer
+import json
+import time
+from config import KAFKA_BROKER, KAFKA_TOPIC
+
+# Connexion au broker Kafka
+producer = KafkaProducer(
+    bootstrap_servers=KAFKA_BROKER,
+    value_serializer=lambda v: json.dumps(v).encode('utf-8')
+)
+
+# Envoi d’un message unique (comme dans ta version)
+message = {"symbol": "AAPL", "price": 183.45, "timestamp": time.time()}
+producer.send(KAFKA_TOPIC, message)
+producer.flush()
+
+print(f"✅ Message envoyé avec succès à {KAFKA_TOPIC} sur {KAFKA_BROKER}")

producer.py

@@ -1,14 +1,23 @@
 from kafka import KafkaProducer
 import json
 import time
+from config import KAFKA_BROKER, KAFKA_TOPIC
 
+# Create Kafka producer
 producer = KafkaProducer(
-    bootstrap_servers='localhost:9092',
+    bootstrap_servers=KAFKA_BROKER,
     value_serializer=lambda v: json.dumps(v).encode('utf-8')
 )
 
+print(f"Producing messages to topic '{KAFKA_TOPIC}' at broker '{KAFKA_BROKER}'")
+
+# Send data in a loop
 while True:
-    data = {"symbol": "AAPL", "price": 187.21, "timestamp": time.time()}
-    producer.send("finance-topic", value=data)
+    data = {
+        "symbol": "AAPL",
+        "price": 187.21,
+        "timestamp": time.time()
+    }
+    producer.send(KAFKA_TOPIC, value=data)
     print(f"Sent: {data}")
     time.sleep(5)