IoT Data Pipeline: Ingestion, Processing & Storage

Traditional Database (e.g., PostgreSQL):
- ✗ Optimized cho OLTP (frequent update, point lookup)
- ✗ Row-oriented storage (bad for time-series)
- ✗ Index bloat với high-cardinality dimension
- ✗ Slow aggregation qua time range
- ✗ Cần VACUUM (stall query)
 
IoT Time-Series Database (e.g., InfluxDB):
- ✓ Optimized cho OLAP (sequential scan, aggregation)
- ✓ Column-oriented storage (compress tỷ monotonic timestamp)
- ✓ Smart index cho time range (không individual row)
- ✓ Efficient downsampling (1 query → giây vs phút)
- ✓ Built-in retention policy (auto cleanup)

MQTT (protocol) ← Devices (telemetry)
         ↓
MQTT Broker (Mosquitto, EMQX, HiveMQ) ← Stores message tạm thời
         ↓
Kafka / RabbitMQ ← Persistent message queue
         ↓
Stream Processor ← Multiple consumer có thể process cùng message

Raw Event: {"deviceId":"temp-007","value":22.5,"ts":1673456789}
         ↓
Transform: Add location từ device registry
         ↓
Enrich:    Add threshold từ device config
         ↓
Filtered:  Only pass nếu value > threshold
         ↓
Output: {"deviceId":"temp-007","value":22.5,"location":"Berlin","threshold":25,"alert":false}

Hot Data (Recent):      TimescaleDB  → Sub-second query, dashboard
Warm Data (Historical): ClickHouse   → Batch analytics, reports
Cold Data (Archive):    S3/GCS       → Compliance, long-term analysis

Bạn có cần accurate historical analytics?
├─ Có
│  └─ Có spare infrastructure?
│     ├─ Có → Lambda (speed + batch layer)
│     └─ Không → Stream-only with replay (Kappa)
└─ Không
   └─ Kappa (simpler, pure streaming)

// Node.js with Kafka Streams (using Kafka-js client)
const kafka = new Kafka({
  clientId: 'iot-processor',
  brokers: ['kafka:9092'],
});
 
const consumer = kafka.consumer({ groupId: 'iot-processing' });
const producer = kafka.producer();
 
await consumer.connect();
await producer.connect();
 
await consumer.subscribe({ topic: 'devices.telemetry' });
 
await consumer.run({
  eachMessage: async ({ topic, partition, message }) => {
    // Parse raw event
    const event = JSON.parse(message.value.toString());
 
    // Transform: lookup device metadata
    const device = await getDeviceMetadata(event.deviceId);
 
    // Enrich: add location and threshold
    const enriched = {
      ...event,
      location: device.location,
      threshold: device.threshold,
      timestamp: Date.now(),
    };
 
    // Filter: only alert if exceeds threshold
    if (enriched.value > enriched.threshold) {
      // Send to alert topic
      await producer.send({
        topic: 'devices.alerts',
        messages: [
          {
            key: enriched.deviceId,
            value: JSON.stringify(enriched),
          },
        ],
      });
    }
 
    // Send all event (enriched) to storage
    await producer.send({
      topic: 'devices.enriched',
      messages: [
        {
          key: enriched.deviceId,
          value: JSON.stringify(enriched),
        },
      ],
    });
  },
});

// Pseudo-code cho stateful Kafka Streams
// Trong Java/Scala, dùng actual Kafka Streams library
 
const windowState = new Map<string, WindowData>();
 
const WINDOW_SIZE = 5 * 60 * 1000; // 5 phút
 
await consumer.run({
  eachMessage: async ({ message }) => {
    const event = JSON.parse(message.value.toString());
    const deviceId = event.deviceId;
 
    // Tạo window key: device + time bucket
    const windowStart = Math.floor(event.ts / WINDOW_SIZE) * WINDOW_SIZE;
    const windowKey = `${deviceId}:${windowStart}`;
 
    // Initialize window nếu new
    if (!windowState.has(windowKey)) {
      windowState.set(windowKey, {
        sum: 0,
        count: 0,
        min: Infinity,
        max: -Infinity,
      });
    }
 
    // Update window
    const window = windowState.get(windowKey);
    window.sum += event.value;
    window.count++;
    window.min = Math.min(window.min, event.value);
    window.max = Math.max(window.max, event.value);
 
    // Emit aggregated result
    const result = {
      deviceId,
      window: {
        start: windowStart,
        end: windowStart + WINDOW_SIZE,
      },
      stats: {
        avg: window.sum / window.count,
        min: window.min,
        max: window.max,
        count: window.count,
      },
    };
 
    await producer.send({
      topic: 'devices.aggregated',
      messages: [{ key: deviceId, value: JSON.stringify(result) }],
    });
  },
});

interface RawEvent {
  deviceId: string;
  sensorId: string;
  value: number;
  ts: number;
}
 
interface EnrichedEvent extends RawEvent {
  location: string;
  deviceType: string;
  zone: string;
  unit: string;
  threshold: number;
  isAnomaly: boolean;
}
 
async function enrichEvent(raw: RawEvent): Promise<EnrichedEvent> {
  // 1. Lookup device metadata
  const device = await deviceRegistry.get(raw.deviceId);
  if (!device) throw new DeviceNotFoundError(raw.deviceId);
 
  // 2. Lookup sensor configuration
  const sensor = device.sensors.find((s) => s.id === raw.sensorId);
  if (!sensor) throw new SensorNotFoundError(raw.sensorId);
 
  // 3. Validate value
  if (raw.value < sensor.min || raw.value > sensor.max) {
    // Out of range
    console.warn(`Out of range: ${raw.deviceId}.${raw.sensorId} = ${raw.value}`);
  }
 
  // 4. Apply unit conversion (if needed)
  const convertedValue = convertUnits(raw.value, sensor.unit, 'SI');
 
  // 5. Detect anomaly
  const historical = await getHistorical(raw.deviceId, raw.sensorId, 24 * 60 * 60 * 1000);
  const isAnomaly = detectAnomaly(
    convertedValue,
    historical,
    sensor.anomalyThreshold,
  );
 
  // 6. Enrich
  return {
    ...raw,
    value: convertedValue,
    location: device.location,
    deviceType: device.deviceType,
    zone: device.zone,
    unit: sensor.unit,
    threshold: sensor.threshold,
    isAnomaly,
  };
}
 
function detectAnomaly(
  value: number,
  historical: number[],
  threshold: number,
): boolean {
  const mean = historical.reduce((a, b) => a + b) / historical.length;
  const std = Math.sqrt(
    historical.reduce((sq, n) => sq + (n - mean) ** 2, 0) / historical.length,
  );
 
  // Flag nếu > 3 standard deviation từ mean
  return Math.abs(value - mean) > threshold * std;
}

interface DataQualityReport {
  total: number;
  valid: number;
  invalid: number;
  duplicate: number;
  outOfRange: number;
  anomaly: number;
  issues: string[];
}
 
async function validateEvent(event: EnrichedEvent): Promise<DataQualityReport> {
  const report: DataQualityReport = {
    total: 1,
    valid: 1,
    invalid: 0,
    duplicate: 0,
    outOfRange: 0,
    anomaly: 0,
    issues: [],
  };
 
  // Check 1: Missing field
  if (!event.deviceId || !event.sensorId || event.value === undefined) {
    report.valid = 0;
    report.invalid = 1;
    report.issues.push('Missing required field');
    return report;
  }
 
  // Check 2: Duplicate detection (seen recently?)
  const recentKey = `${event.deviceId}:${event.sensorId}:${event.value}`;
  const lastSeen = await cache.get(recentKey);
  if (lastSeen && Date.now() - lastSeen < 1000) {
    report.duplicate = 1;
    report.issues.push(`Duplicate trong 1s');
  }
 
  // Check 3: Out of range
  if (event.value < -50 || event.value > 150) {
    // Temperature sensor
    report.outOfRange = 1;
    report.issues.push(`Out of valid range: ${event.value}`);
  }
 
  // Check 4: Anomaly (already detected during enrichment)
  if (event.isAnomaly) {
    report.anomaly = 1;
    report.issues.push('Detected anomaly');
  }
 
  // Check 5: Schema validation
  // (done at ingestion, nhưng good re-check)
 
  return report;
}
 
// Route event dựa trên quality
async function routeEvent(event: EnrichedEvent) {
  const quality = await validateEvent(event);
 
  if (quality.issues.length > 0) {
    // Send to quarantine topic để investigate
    await producer.send({
      topic: 'devices.quarantine',
      messages: [
        {
          key: event.deviceId,
          value: JSON.stringify({ event, quality }),
        },
      ],
    });
    return;
  }
 
  // Send to storage
  await producer.send({
    topic: 'devices.enriched',
    messages: [{ key: event.deviceId, value: JSON.stringify(event) }],
  });
}

// Lifecycle management
interface StorageConfig {
  hot: {
    database: 'timescaledb';
    retention: 7 * 24 * 60 * 60 * 1000; // 7 ngày
    compression: 'snappy';
  };
  warm: {
    database: 'clickhouse';
    retention: 365 * 24 * 60 * 60 * 1000; // 1 năm
    compression: 'gzip';
  };
  cold: {
    storage: 's3';
    bucket: 'iot-archive';
    retention: null; // Indefinite
    format: 'parquet';
  };
}
 
// Mỗi giờ, migrate old data
setInterval(async () => {
  const now = Date.now();
 
  // Send 7-100 ngày old data từ TimescaleDB sang ClickHouse
  const toDrop = now - 14 * 24 * 60 * 60 * 1000; // Older than 14 days
  const toWarm = now - 7 * 24 * 60 * 60 * 1000; // Older than 7 days
 
  const warmData = await timescaledb.query(`
    SELECT * FROM events
    WHERE ts > ${toWarm} AND ts < ${toDrop}
  `);
 
  // Insert vào ClickHouse
  await clickhouse.insert('events', warmData);
 
  // Delete từ TimescaleDB
  await timescaledb.query(`
    DELETE FROM events WHERE ts < ${toWarm}
  `);
 
  // Mỗi tuần, archive ClickHouse sang S3
  const toArchive = now - 365 * 24 * 60 * 60 * 1000;
  const archiveData = await clickhouse.query(`
    SELECT * FROM events WHERE ts < ${toArchive}
  `);
 
  // Convert sang Parquet và upload S3
  const parquetFile = await convertToParquet(archiveData);
  await s3.upload(`events/${datePartition(toArchive)}.parquet`, parquetFile);
 
  // Delete từ ClickHouse
  await clickhouse.query(`DELETE FROM events WHERE ts < ${toArchive}`);
}, 60 * 60 * 1000); // Mỗi giờ

# docker-compose.yml cho IoT pipeline
version: '3.8'
 
services:
  # MQTT Broker
  emqx:
    image: emqx/emqx:5.0
    ports:
      - '1883:1883' # MQTT
      - '8083:8083' # WebSocket
      - '18083:18083' # Dashboard
    environment:
      EMQX_LOADED_PLUGINS: 'emqx_management,emqx_recon'
 
  # Kafka
  kafka:
    image: confluentinc/cp-kafka:7.5.0
    ports:
      - '9092:9092'
    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
    depends_on:
      - zookeeper
 
  zookeeper:
    image: confluentinc/cp-zookeeper:7.5.0
    ports:
      - '2181:2181'
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
 
  # TimescaleDB (PostgreSQL + TimescaleDB extension)
  timescaledb:
    image: timescale/timescaledb:latest-pg15
    ports:
      - '5432:5432'
    environment:
      POSTGRES_PASSWORD: password
      POSTGRES_DB: iot
 
  # Stream Processor (chạy Node.js app)
  processor:
    build: .
    depends_on:
      - kafka
      - emqx
      - timescaledb
    environment:
      KAFKA_BROKERS: kafka:29092
      MQTT_URL: mqtt://emqx:1883
      DATABASE_URL: postgresql://postgres:password@timescaledb:5432/iot
 
  # Grafana cho dashboard
  grafana:
    image: grafana/grafana:latest
    ports:
      - '3000:3000'
    environment:
      GF_SECURITY_ADMIN_PASSWORD: admin
    depends_on:
      - timescaledb

// processor.js - Glue that binds it all together
const mqtt = require('mqtt');
const { Kafka } = require('kafkajs');
const { Client: PG } = require('pg');
 
const kafka = new Kafka({
  clientId: 'iot-processor',
  brokers: process.env.KAFKA_BROKERS.split(','),
});
 
const consumer = kafka.consumer({ groupId: 'iot-processing' });
const producer = kafka.producer();
 
const pgClient = new PG({
  connectionString: process.env.DATABASE_URL,
});
 
const mqttClient = mqtt.connect(process.env.MQTT_URL);
 
async function start() {
  await consumer.connect();
  await producer.connect();
  await pgClient.connect();
 
  // Subscribe to raw telemetry
  await consumer.subscribe({ topic: 'devices.telemetry' });
 
  console.log('🚀 IoT Stream Processor started');
 
  await consumer.run({
    eachMessage: async ({ topic, partition, message }) => {
      try {
        const event = JSON.parse(message.value.toString());
 
        // Transform: add timestamp
        event.processedAt = Date.now();
 
        // Enrich: add device metadata
        const { rows } = await pgClient.query(
          'SELECT * FROM devices WHERE device_id = $1',
          [event.deviceId],
        );
        if (rows.length === 0) {
          console.warn(`Device not found: ${event.deviceId}`);
          return;
        }
 
        event.location = rows[0].location;
        event.zone = rows[0].zone;
 
        // Detect anomaly
        event.isAnomaly = event.value > rows[0].threshold;
 
        // Insert vào TimescaleDB
        await pgClient.query(
          `INSERT INTO metrics (device_id, value, location, is_anomaly, ts)
           VALUES ($1, $2, $3, $4, to_timestamp($5 / 1000.0))`,
          [
            event.deviceId,
            event.value,
            event.location,
            event.isAnomaly,
            event.ts,
          ],
        );
 
        // Nếu anomaly, gửi alert
        if (event.isAnomaly) {
          await producer.send({
            topic: 'alerts',
            messages: [{ key: event.deviceId, value: JSON.stringify(event) }],
          });
        }
      } catch (error) {
        console.error('Error processing message:', error);
      }
    },
  });
}
 
start().catch(console.error);

1. **Ingest từ device** → MQTT Broker (Mosquitto, EMQX)
2. **Buffer trong queue** → Kafka (guarantee durability)
3. **Stream process** → Kafka Streams (transform, enrich, filter)
4. **Detect anomaly** → Real-time threshold + statistical model
5. **Store hot data** → TimescaleDB (7 ngày, sub-100ms query)
6. **Migrate warm data** → ClickHouse (analytical query, 1 năm)
7. **Archive cold data** → S3 Parquet (compliance, indefinite)
8. **Visualize** → Grafana dashboard
9. **Alert** → Automatic rule evaluation
10. **Iterate** → Monitor pipeline health, adjust threshold