Cas Pratique : Ingestion IoT et Séries Temporelles avec Cassandra et Python#
Dans l'industrie contemporaine (automobile connectée, compteurs d'énergie communicants, usines 4.0 et télémétrie aéronautique), le déluge de données généré par les capteurs connectés (IoT - Internet of Things) représente un défi d'ingénierie colossal. Des millions de messages affluent chaque seconde, avec une tolérance zéro pour les pertes de données et une exigence d'interrogation en temps réel.
Dans ce cas pratique intégrateur, nous allons concevoir l'architecture de données de « FleetIoT Analytics », une plateforme de supervision pour une flotte de véhicules logistiques connectés. Nous aborderons la technique incontournable du découpage temporel (Bucketing), créerons les tables en CQL, puis implémenterons un pipeline d'ingestion et d'analyse en Python avec le pilote officiel cassandra-driver.
1. Problématique Métier : FleetIoT Analytics#
Une entreprise de fret international gère une flotte de 10 000 camions autonomes et connectés. Chaque véhicule transmet à intervalles réguliers (toutes les quelques secondes) un paquet de télémétrie comportant :
- L'identifiant du camion (vehicule_id).
- La date et l'heure précise de mesure (timestamp).
- La vitesse instantanée (en km/h), le régime moteur (en tr/min) et la température de l'huile (en °C).
- Les coordonnées GPS de localisation (latitude, longitude).
- Un ensemble d'alertes ou statuts matériels.
Objectifs fonctionnels du système :#
- Ingérer les flux sans saturation : Absorber des débits d'écriture massifs et constants sans aucun verrouillage de table.
- Consulter l'historique d'un véhicule sur un créneau horaire précis : Restituer instantanément les relevés chronologiques d'une journée donnée, triés du plus récent au plus ancien.
- Surveiller les anomalies thermiques : Enregistrer immédiatement les alertes de surchauffe moteur (\(> 100^\circ\text{C}\)).
2. Modélisation NoSQL : La technique du « Bucketing » temporel#
L'erreur la plus fréquente des débutants avec Cassandra consiste à concevoir une table temporelle avec une clé primaire naïve :
Pourquoi ce modèle est-il dangereux ?#
Dans ce modèle naïf, l'ensemble des millions de mesures d'un même véhicule sur 5 ans sera dirigé vers une seule et unique partition physique sur un seul nœud du cluster ! - Au bout de quelques mois, la partition dépassera plusieurs gigaoctets. - Dans Cassandra, la taille maximale recommandée d'une partition est de 100 Mo (au-delà, la mémoire RAM s'épuise lors des compactions et les temps de réponse s'effondrent).
La solution : La clé de partition composée avec Bucketing temporel#
Pour garantir des partitions équilibrées et de taille maîtrisée, nous découpons les données en tranches de temps (Buckets), par exemple par jour :
-- ✅ MODÉLISATION OPTIMALE AVEC BUCKETING JOURNALIER :
PRIMARY KEY ((vehicule_id, date_jour), timestamp_mesure)
WITH CLUSTERING ORDER BY (timestamp_mesure DESC);
flowchart TD
subgraph Cle_Primaire ["Anatomie de la Clé Primaire"]
direction LR
PK["(vehicule_id, date_jour)<br/>Partition Key (Le Bucket Journalier)"]
CK["timestamp_mesure DESC<br/>Clustering Key (Tri séquentiel sur disque)"]
end
subgraph Partition_Disque ["Structure physique d'une partition (~8 Mo)"]
R1["2026-03-03 23:59:59 | 92 km/h | 88°C"]
R2["2026-03-03 23:59:58 | 91 km/h | 88°C"]
R3["... 86 400 lignes chronologiques ordonnées sur disque ..."]
R4["2026-03-03 00:00:01 | 0 km/h | 20°C"]
end
PK --> Partition_Disque
CK -.->|Garantit l'ordre| Partition_Disque
Avec un relevé par seconde, une journée représente 86 400 lignes de \(\sim 100\) octets, soit une partition de \(\sim 8.6\) Mo, ce qui est la taille idéale pour les caches mémoire de Cassandra !
3. Schéma CQL de la solution#
Ouvrez votre console cqlsh :
Exécutez le script CQL ci-dessous pour créer le Keyspace et les tables optimisées :
-- 1. Création du Keyspace avec réplication locale
CREATE KEYSPACE IF NOT EXISTS fleet_iot
WITH REPLICATION = {
'class': 'SimpleStrategy',
'replication_factor': 1
};
USE fleet_iot;
-- 2. Table principale de télémétrie (Partitionnée par véhicule et par jour)
CREATE TABLE IF NOT EXISTS telemetrie_vehicules (
vehicule_id text,
date_jour date,
timestamp_mesure timestamp,
vitesse float,
regime_moteur int,
temperature_huile float,
latitude double,
longitude double,
statut map<text, boolean>,
PRIMARY KEY ((vehicule_id, date_jour), timestamp_mesure)
) WITH CLUSTERING ORDER BY (timestamp_mesure DESC);
-- 3. Table dédiée aux alertes critiques (requêtées par date pour le centre de contrôle)
CREATE TABLE IF NOT EXISTS alertes_critiques (
date_jour date,
timestamp_alerte timestamp,
vehicule_id text,
type_anomalie text,
valeur_mesuree float,
PRIMARY KEY (date_jour, timestamp_alerte, vehicule_id)
) WITH CLUSTERING ORDER BY (timestamp_alerte DESC);
4. Pipeline d'ingestion et de requêtage en Python#
Le pilote officiel cassandra-driver fourni par DataStax offre une gestion asynchrone des écritures, la découverte automatique des nœuds du cluster (Load Balancing Policy) et la mise en cache des requêtes préparées (Prepared Statements).
Installation du pilote Python#
Dans votre environnement de travail :
Script complet d'ingestion et d'analyse (iot_fleet_platform.py)#
Créez le fichier iot_fleet_platform.py :
"""
Plateforme d'ingestion et d'analyse de télémétrie IoT avec Apache Cassandra.
Auteur : Joseph Konka (AlgoJungle)
"""
from datetime import datetime, date, timedelta
import random
import time
from cassandra.cluster import Cluster, Session
from cassandra.query import PreparedStatement
class FleetIoTManager:
def __init__(self, contact_points: list[str] = ["localhost"], port: int = 9042):
print(f"⏳ Connexion au cluster Cassandra sur {contact_points}:{port}...")
self.cluster = Cluster(contact_points, port=port)
self.session: Session = self.cluster.connect("fleet_iot")
print("✅ Connexion établie avec succès !")
# Préparation des requêtes (Prepared Statements) pour une vitesse maximale
self._preparer_requetes()
def _preparer_requetes(self):
"""Compile les requêtes une seule fois au niveau du coordinateur Cassandra."""
cql_insert = """
INSERT INTO telemetrie_vehicules (
vehicule_id, date_jour, timestamp_mesure,
vitesse, regime_moteur, temperature_huile,
latitude, longitude, statut
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?);
"""
self.stmt_insert_telemetrie: PreparedStatement = self.session.prepare(cql_insert)
cql_insert_alerte = """
INSERT INTO alertes_critiques (
date_jour, timestamp_alerte, vehicule_id, type_anomalie, valeur_mesuree
) VALUES (?, ?, ?, ?, ?);
"""
self.stmt_insert_alerte: PreparedStatement = self.session.prepare(cql_insert_alerte)
def simuler_ingestion_flux(self, vehicule_id: str, nb_mesures: int = 10):
"""Simule un flux télémétrique à haute fréquence avec détection d'alertes."""
print(f"\n📡 Démarrage de l'ingestion pour le camion [{vehicule_id}]...")
maintenant = datetime.now()
date_actuelle = maintenant.date()
for i in range(nb_mesures):
ts = maintenant - timedelta(seconds=i * 5)
vitesse = round(random.uniform(70.0, 95.0), 1)
regime = random.randint(1800, 2400)
# Génération d'une surchauffe aléatoire pour tester l'alerte
temp_huile = round(random.uniform(85.0, 108.0), 1)
lat = round(48.8566 + random.uniform(-0.05, 0.05), 4)
lon = round(2.3522 + random.uniform(-0.05, 0.05), 4)
statut = {"freins_ok": True, "pression_pneus_ok": True}
# 1. Écriture de la télémétrie brute
self.session.execute(
self.stmt_insert_telemetrie,
(vehicule_id, date_actuelle, ts, vitesse, regime, temp_huile, lat, lon, statut),
)
# 2. Détection d'anomalie critique (> 100°C)
if temp_huile > 100.0:
self.session.execute(
self.stmt_insert_alerte,
(date_actuelle, ts, vehicule_id, "SURCHAUFFE_MOTEUR", temp_huile),
)
print(f" 🚨 ALERTE CRITIQUE enregistrée : {temp_huile}°C à {ts.strftime('%H:%M:%S')} !")
print(f"📥 {nb_mesures} paquets de télémétrie ingérés avec succès.")
def consulter_historique_recent(self, vehicule_id: str, date_recherche: date, limite: int = 5):
"""Récupère les mesures les plus récentes d'un camion sur une journée."""
query = """
SELECT timestamp_mesure, vitesse, regime_moteur, temperature_huile, latitude, longitude
FROM telemetrie_vehicules
WHERE vehicule_id = %s AND date_jour = %s
LIMIT %s;
"""
resultats = self.session.execute(query, (vehicule_id, date_recherche, limite))
return list(resultats)
def lister_alertes_journee(self, date_recherche: date):
"""Consulte toutes les alertes critiques enregistrées sur la journée."""
query = """
SELECT timestamp_alerte, vehicule_id, type_anomalie, valeur_mesuree
FROM alertes_critiques
WHERE date_jour = %s;
"""
return list(self.session.execute(query, (date_recherche,)))
def fermer(self):
self.cluster.shutdown()
print("\n🔒 Connexion au cluster Cassandra fermée proprement.")
def main():
manager = FleetIoTManager(["localhost"], 9042)
try:
id_camion = "CAMION-FR-75"
aujourdhui = date.today()
# 1. Ingestion d'un flux de mesures simulées
manager.simuler_ingestion_flux(id_camion, nb_mesures=12)
# 2. Consultation des dernières mesures (ordonnées par clustering key DESC)
print("\n" + "=" * 70)
print(f"📊 DERNIERS RELEVÉS CHRONOLOGIQUES : {id_camion} ({aujourdhui})")
print("=" * 70)
print(f"{'Horodatage':<12} | {'Vitesse':>10} | {'Régime':>10} | {'Température':>12} | Coordonnées GPS")
print("-" * 70)
mesures = manager.consulter_historique_recent(id_camion, aujourdhui, limite=6)
for m in mesures:
ts_str = m.timestamp_mesure.strftime("%H:%M:%S")
print(
f"{ts_str:<12} | {m.vitesse:>7.1f} km/h | {m.regime_moteur:>6} rpm | "
f"{m.temperature_huile:>9.1f} °C | ({m.latitude}, {m.longitude})"
)
# 3. Consultation du journal des alertes
print("\n" + "=" * 70)
print(f"🚨 JOURNAL DES ALERTES DU JOUR : {aujourdhui}")
print("=" * 70)
alertes = manager.lister_alertes_journee(aujourdhui)
if not alertes:
print("Aucune anomalie détectée aujourd'hui.")
else:
for a in alertes:
ts_str = a.timestamp_alerte.strftime("%H:%M:%S")
print(f"⚠️ [{ts_str}] {a.vehicule_id} : {a.type_anomalie} ({a.valeur_mesuree} °C)")
print("=" * 70)
except Exception as err:
print(f"❌ Erreur d'exécution : {err}")
finally:
manager.fermer()
if __name__ == "__main__":
main()
Exécution du script et résultat en console#
Exécutez le script dans votre terminal :
Trace d'exécution obtenue :
⏳ Connexion au cluster Cassandra sur ['localhost']:9042...
✅ Connexion établie avec succès !
📡 Démarrage de l'ingestion pour le camion [CAMION-FR-75]...
🚨 ALERTE CRITIQUE enregistrée : 104.8°C à 14:22:15 !
🚨 ALERTE CRITIQUE enregistrée : 106.2°C à 14:22:30 !
📥 12 paquets de télémétrie ingérés avec succès.
======================================================================
📊 DERNIERS RELEVÉS CHRONOLOGIQUES : CAMION-FR-75 (2026-03-03)
======================================================================
Horodatage | Vitesse | Régime | Température | Coordonnées GPS
----------------------------------------------------------------------
14:22:45 | 88.4 km/h | 2150 rpm | 89.2 °C | (48.8312, 2.3781)
14:22:40 | 91.0 km/h | 2230 rpm | 93.5 °C | (48.8415, 2.3654)
14:22:35 | 76.2 km/h | 1940 rpm | 98.1 °C | (48.8501, 2.3590)
14:22:30 | 82.7 km/h | 2080 rpm | 106.2 °C | (48.8610, 2.3421)
14:22:25 | 89.5 km/h | 2190 rpm | 91.4 °C | (48.8702, 2.3312)
14:22:20 | 94.1 km/h | 2310 rpm | 95.0 °C | (48.8814, 2.3205)
======================================================================
🚨 JOURNAL DES ALERTES DU JOUR : 2026-03-03
======================================================================
⚠️ [14:22:30] CAMION-FR-75 : SURCHAUFFE_MOTEUR (106.2 °C)
⚠️ [14:22:15] CAMION-FR-75 : SURCHAUFFE_MOTEUR (104.8 °C)
======================================================================
🔒 Connexion au cluster Cassandra fermée proprement.
5. Défis d'extension pour l'ingénieur Big Data#
Pour parfaire votre expertise :
Défi 1 : Ingestion ultra-haute cadence avec concurrent.execute_concurrent#
- Utilisez le module
cassandra.concurrentdu pilote Python pour soumettre des milliers d'écritures simultanées de manière asynchrone sans bloquer le thread principal.
Défi 2 : Expiration automatique des données télémétriques#
- Modifiez la définition de la table
telemetrie_vehiculespour lui associer un paramètre de tableAND default_time_to_live = 2592000;(30 jours en secondes). Observez comment Cassandra purge automatiquement les données anciennes sans nécessiter de tâche de suppression manuelle.
Commentaires