Kafka pour l’IoT : Streaming haute performance avec clients React et persistance CouchDB

Apache Kafka fournit une infrastructure de streaming tolérante aux pannes et à haut débit pour l’ingestion de télémétrie depuis des appareils IoT, le traitement en temps réel et la persistance dans des bases comme CouchDB.
Cette spécification décrit les caractéristiques de performance, la conception architecturale et inclut un exemple de producteur React Native pour le suivi de flotte.

---

Capacités de performance

• Débit :
  • Broker unique : ~200 000 messages/sec (taille de message ~1 Ko).
  • Cluster de 5 brokers : évolue linéairement jusqu’à 1–2M messages/sec.
• Latence : moins de 10 ms en déploiement LAN, moins de 50 ms entre régions.
• Durabilité : Facteur de réplication 3 avec acks=all garantit l’absence de perte de données.
• Rétention : Sujets configurables pour plusieurs jours ou semaines, permettant relecture et rechargement.

---

Composants principaux

1. Appareil IoT (Producteur)
   Publie la télémétrie (ex. données GPS) vers Kafka.

2. Brokers Kafka (Backbone)
   Fournissent un transport distribué et durable des messages.

3. Application cliente React (Consommateur)
   Reçoit les mises à jour via un pont WebSocket/REST pour les tableaux de bord en temps réel.

4. Sink CouchDB
   Stocke les documents JSON pour interrogation, réplication et fonctionnement hors ligne.

---

Cas d’usage : Suivi de flotte

• Échelle : 5 000 véhicules × 1 mise à jour/sec → 300 000 messages/min.
• Workflow :
  • Chaque véhicule publie des événements fleet.position avec position et vitesse.
  • Kafka partitionne les données par vehicle_id pour préserver l’ordre.
  • Le tableau de bord React affiche les déplacements en temps réel.
  • CouchDB stocke les journaux pour conformité et analyse historique.

---

Exemple de producteur personnalisé (React Native)

Une application React Native peut agir comme producteur Kafka (via une passerelle API).
Exemple simplifié pour un événement fleet.position :

// fleetProducer.ts
import Geolocation, { GeoPosition } from "react-native-geolocation-service";

async function getPosition(): Promise<GeoPosition> {
  return new Promise((resolve, reject) => {
    Geolocation.getCurrentPosition(resolve, reject, {
      enableHighAccuracy: true,
      timeout: 10000,
    });
  });
}

export async function sendFleetPosition(device_id: string, token: string) {
  const pos = await getPosition();

  const event = {
    device_id,
    ts: new Date().toISOString(),
    use_case: "fleet.position",
    schema: "v1",
    payload: {
      lat: pos.coords.latitude,
      lon: pos.coords.longitude,
      speed_kmh: (pos.coords.speed ?? 0) * 3.6, // m/s → km/h
      heading_deg: pos.coords.heading ?? undefined,
      status: "delivering",
    },
  };

  await fetch("https://your-gateway.example.com/ingest", {
    method: "POST",
    headers: {
      "Content-Type": "application/json",
      "Authorization": `Bearer ${token}`,
    },
    body: JSON.stringify({ events: [event] }),
  });
}

Fonctionnement :

• L’application mobile capture les données GPS.
• Elles sont encapsulées dans un événement fleet.position.
• L’événement est envoyé à une API d’ingestion sécurisée.
• La passerelle publie vers un sujet Kafka (fleet.updates.v1).
• Kafka garantit l’ordre par véhicule (partitionné par device_id).
• CouchDB consomme les événements pour un stockage durable.

---

Avantages

• Réactivité temps réel : moins de 50 ms de l’appareil au tableau de bord.
• Messagerie évolutive : des millions de messages par minute à l’échelle de la flotte.
• Persistance durable : CouchDB fournit un stockage JSON avec réplication.
• Applications multiplateformes : React Native fonctionne sur iOS et Android avec un code unique.

---

Sujets Kafka (exemple CLI)

# Positions de flotte : 12 partitions pour ~50k appareils à 1 msg/s (ajustez selon la charge)
kafka-topics.sh --create --topic fleet.updates.v1 --partitions 12 --replication-factor 3 \
  --config cleanup.policy=delete --config retention.ms=604800000  # 7 jours

kafka-topics.sh --create --topic coldchain.temp.v1 --partitions 6 --replication-factor 3 \
  --config retention.ms=259200000  # 3 jours

kafka-topics.sh --create --topic driver.panic.v1 --partitions 3 --replication-factor 3 \
  --config retention.ms=1209600000 # 14 jours

---

(Optionnel) Kafka → CouchDB Sink

Pour persister automatiquement les événements dans CouchDB, configurez un connecteur Kafka Connect. Exemple (pseudo-config) :

{
  "name": "couchdb-sink",
  "config": {
    "connector.class": "com.github.couchdb.KafkaCouchDBSinkConnector",
    "tasks.max": "2",
    "topics": "fleet.updates.v1,coldchain.temp.v1,driver.panic.v1",
    "couchdb.url": "http://admin:password@couchdb:5984",
    "couchdb.db.mapping": "fleet.updates.v1:fleet_updates,coldchain.temp.v1:coldchain_temp,driver.panic.v1:driver_panic",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter"
  }
}

Remarque : Il n’existe pas de connecteur CouchDB officiel dans Kafka Connect. Options :

• Utiliser un connecteur communautaire CouchDB.
• Ou envoyer les événements vers un sujet compacté et utiliser Kafka Streams / ksqlDB / Flink pour les écrire dans CouchDB via son API HTTP.

---

Schéma architectural

Appareil Véhicule (Producteur)
   → Brokers Kafka (Backbone de streaming)
      → Tableau de bord React (Consommateur via WebSocket/REST)
      → CouchDB (Stockage et analytique durables)

Ce modèle s’adapte à des dizaines de milliers d’appareils IoT, prend en charge des tableaux de bord instantanés et assure une persistance des événements à long terme.

---