# Edge Computing für IoT **Meta-Description:** Edge Computing für IoT-Anwendungen. Lokale Datenverarbeitung, ML Inference, Fog Computing und Edge-Cloud Architektur. **Keywords:** Edge Computing, IoT, Fog Computing, ML Inference, Local Processing, Edge Gateway, TensorFlow Lite --- ## Einführung **Edge Computing** bringt Rechenleistung dorthin, wo Daten entstehen. Für IoT bedeutet das: **niedrige Latenz, Offline-Fähigkeit, Datenschutz** und reduzierte Bandbreite durch lokale Verarbeitung direkt am Gerät oder Gateway. --- ## Edge Computing Architecture ``` ┌─────────────────────────────────────────────────────────────┐ │ EDGE COMPUTING ARCHITECTURE │ ├─────────────────────────────────────────────────────────────┤ │ │ │ Cloud Layer: │ │ ┌─────────────────────────────────────────────────────┐ │ │ │ Cloud Platform (AWS IoT / Azure IoT / GCP IoT) │ │ │ │ - Long-term Storage │ │ │ │ - Complex Analytics │ │ │ │ - ML Model Training │ │ │ │ - Global Dashboards │ │ │ └─────────────────────────────────────────────────────┘ │ │ ↑↓ Aggregated Data │ │ Edge Layer (Fog): │ │ ┌─────────────────────────────────────────────────────┐ │ │ │ Edge Gateway / Server │ │ │ │ ┌───────────┐ ┌───────────┐ ┌───────────┐ │ │ │ │ │ Local DB │ │ ML Infer- │ │ Rules │ │ │ │ │ │ (InfluxDB)│ │ ence │ │ Engine │ │ │ │ │ └───────────┘ └───────────┘ └───────────┘ │ │ │ │ - Data Aggregation │ │ │ │ - Real-time Processing │ │ │ │ - Local Decisions │ │ │ └─────────────────────────────────────────────────────┘ │ │ ↑↓ Filtered Data │ │ Device Layer: │ │ ┌─────────────────────────────────────────────────────┐ │ │ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌───────┐ │ │ │ │ │ Sensor │ │ Camera │ │ Actuator│ │ PLC │ │ │ │ │ │ │ │ (CV) │ │ │ │ │ │ │ │ │ └─────────┘ └─────────┘ └─────────┘ └───────┘ │ │ │ │ - Data Collection │ │ │ │ - Simple Processing │ │ │ └─────────────────────────────────────────────────────┘ │ │ │ │ Latency: Cloud ~100ms | Edge ~10ms | Device ~1ms │ │ │ └─────────────────────────────────────────────────────────────┘ ``` --- ## Edge Gateway mit Node.js ```typescript // edge-gateway/index.ts import mqtt from 'mqtt'; import { InfluxDBClient } from './lib/influxdb'; import { RulesEngine } from './lib/rules-engine'; import { MLInference } from './lib/ml-inference'; interface EdgeConfig { mqttBroker: string; cloudEndpoint: string; localDb: { url: string; token: string; org: string; bucket: string; }; syncInterval: number; // Millisekunden offlineBufferSize: number; } class EdgeGateway { private localMqtt: mqtt.MqttClient; private cloudMqtt: mqtt.MqttClient | null = null; private influx: InfluxDBClient; private rules: RulesEngine; private ml: MLInference; private offlineBuffer: any[] = []; private isCloudConnected = false; constructor(private config: EdgeConfig) { this.influx = new InfluxDBClient(config.localDb); this.rules = new RulesEngine(); this.ml = new MLInference(); } async start(): Promise { // Local MQTT Broker verbinden this.localMqtt = mqtt.connect(this.config.mqttBroker); this.localMqtt.on('connect', () => { console.log('Connected to local MQTT broker'); this.localMqtt.subscribe('sensors/#'); this.localMqtt.subscribe('devices/#'); }); this.localMqtt.on('message', (topic, payload) => { this.handleLocalMessage(topic, payload.toString()); }); // Cloud Connection (mit Retry) this.connectToCloud(); // Periodische Cloud-Synchronisation setInterval(() => this.syncToCloud(), this.config.syncInterval); // ML Models laden await this.ml.loadModels(); console.log('Edge Gateway started'); } private async handleLocalMessage(topic: string, payload: string): Promise { try { const data = JSON.parse(payload); const timestamp = new Date(); // 1. Lokale Speicherung await this.storeLocally(topic, data, timestamp); // 2. Echtzeit-Verarbeitung const processed = await this.processData(topic, data); // 3. Rules Engine const actions = this.rules.evaluate(topic, processed); await this.executeActions(actions); // 4. ML Inference (wenn relevant) if (this.shouldRunInference(topic, data)) { const prediction = await this.ml.predict(data); await this.handlePrediction(topic, prediction); } // 5. Cloud-Queue (aggregierte Daten) this.queueForCloud(topic, processed, timestamp); } catch (error) { console.error('Error processing message:', error); } } private async storeLocally( topic: string, data: any, timestamp: Date ): Promise { const measurement = topic.split('/')[0]; const deviceId = topic.split('/')[1]; await this.influx.writePoint({ measurement, tags: { device_id: deviceId }, fields: data, timestamp }); } private async processData(topic: string, data: any): Promise { // Daten normalisieren, filtern, aggregieren const processed = { ...data }; // Outlier Detection if (data.temperature !== undefined) { if (data.temperature < -50 || data.temperature > 100) { processed.temperature_valid = false; processed.temperature_original = data.temperature; delete processed.temperature; } } // Unit Conversion (wenn nötig) // ... return processed; } private queueForCloud(topic: string, data: any, timestamp: Date): void { // Nur relevante Daten für Cloud const cloudData = { topic, data: this.aggregateForCloud(data), timestamp: timestamp.toISOString(), gatewayId: process.env.GATEWAY_ID }; if (this.isCloudConnected) { this.sendToCloud(cloudData); } else { // Offline Buffer this.offlineBuffer.push(cloudData); // Buffer Limit if (this.offlineBuffer.length > this.config.offlineBufferSize) { this.offlineBuffer.shift(); // Älteste Daten verwerfen } } } private aggregateForCloud(data: any): any { // Nur wichtige Felder für Cloud const { temperature, humidity, battery, state } = data; return { temperature, humidity, battery, state }; } private async syncToCloud(): Promise { if (!this.isCloudConnected || this.offlineBuffer.length === 0) { return; } console.log(`Syncing ${this.offlineBuffer.length} messages to cloud`); // Batch Upload const batch = this.offlineBuffer.splice(0, 100); try { await this.sendToCloud({ type: 'batch', messages: batch }); } catch (error) { // Bei Fehler: zurück in Buffer this.offlineBuffer.unshift(...batch); console.error('Cloud sync failed:', error); } } private sendToCloud(data: any): void { this.cloudMqtt?.publish( `gateways/${process.env.GATEWAY_ID}/data`, JSON.stringify(data) ); } private async executeActions(actions: any[]): Promise { for (const action of actions) { console.log('Executing action:', action); switch (action.type) { case 'publish': this.localMqtt.publish(action.topic, JSON.stringify(action.payload)); break; case 'alert': await this.sendAlert(action.message, action.severity); break; case 'device_command': this.localMqtt.publish( `devices/${action.deviceId}/command`, JSON.stringify(action.command) ); break; } } } private connectToCloud(): void { this.cloudMqtt = mqtt.connect(this.config.cloudEndpoint, { clientId: `gateway-${process.env.GATEWAY_ID}`, username: process.env.CLOUD_USER, password: process.env.CLOUD_PASS, reconnectPeriod: 10000 }); this.cloudMqtt.on('connect', () => { console.log('Connected to cloud'); this.isCloudConnected = true; // Subscribe to cloud commands this.cloudMqtt?.subscribe(`gateways/${process.env.GATEWAY_ID}/command`); }); this.cloudMqtt.on('close', () => { console.log('Cloud connection closed'); this.isCloudConnected = false; }); this.cloudMqtt.on('message', (topic, payload) => { this.handleCloudCommand(JSON.parse(payload.toString())); }); } private handleCloudCommand(command: any): void { console.log('Cloud command received:', command); switch (command.type) { case 'update_config': this.updateConfig(command.config); break; case 'update_rules': this.rules.updateRules(command.rules); break; case 'update_model': this.ml.updateModel(command.modelId, command.modelUrl); break; case 'device_command': this.localMqtt.publish( `devices/${command.deviceId}/command`, JSON.stringify(command.payload) ); break; } } } ``` --- ## ML Inference am Edge ```typescript // lib/ml-inference.ts import * as tf from '@tensorflow/tfjs-node'; interface Model { id: string; model: tf.LayersModel; inputShape: number[]; labels: string[]; } class MLInference { private models: Map = new Map(); async loadModels(): Promise { // Anomaly Detection Model const anomalyModel = await tf.loadLayersModel( 'file://./models/anomaly/model.json' ); this.models.set('anomaly', { id: 'anomaly', model: anomalyModel, inputShape: [1, 10], // 10 Features labels: ['normal', 'anomaly'] }); // Predictive Maintenance Model const maintenanceModel = await tf.loadLayersModel( 'file://./models/maintenance/model.json' ); this.models.set('maintenance', { id: 'maintenance', model: maintenanceModel, inputShape: [1, 24], // 24 Stunden Historie labels: ['ok', 'warning', 'critical'] }); console.log(`Loaded ${this.models.size} ML models`); } async predict( modelId: string, input: number[] ): Promise<{ label: string; confidence: number }> { const modelInfo = this.models.get(modelId); if (!modelInfo) { throw new Error(`Model ${modelId} not found`); } // Input Tensor erstellen const inputTensor = tf.tensor2d([input], modelInfo.inputShape as [number, number]); // Inference const prediction = modelInfo.model.predict(inputTensor) as tf.Tensor; const probabilities = await prediction.data(); // Cleanup inputTensor.dispose(); prediction.dispose(); // Beste Prediction finden const maxIndex = probabilities.indexOf(Math.max(...Array.from(probabilities))); return { label: modelInfo.labels[maxIndex], confidence: probabilities[maxIndex] }; } async detectAnomaly( sensorData: { temperature: number; humidity: number; pressure: number; vibration: number; } ): Promise<{ isAnomaly: boolean; confidence: number }> { // Feature Engineering const features = [ sensorData.temperature, sensorData.humidity, sensorData.pressure, sensorData.vibration, // Normalisierte Werte (sensorData.temperature - 20) / 30, (sensorData.humidity - 50) / 50, // Interaktionen sensorData.temperature * sensorData.humidity / 1000, // Placeholder für historische Features 0, 0, 0 ]; const result = await this.predict('anomaly', features); return { isAnomaly: result.label === 'anomaly', confidence: result.confidence }; } async updateModel(modelId: string, modelUrl: string): Promise { console.log(`Updating model ${modelId} from ${modelUrl}`); try { const newModel = await tf.loadLayersModel(modelUrl); const existing = this.models.get(modelId); if (existing) { existing.model.dispose(); existing.model = newModel; } console.log(`Model ${modelId} updated successfully`); } catch (error) { console.error(`Failed to update model ${modelId}:`, error); } } } export { MLInference }; ``` --- ## Rules Engine ```typescript // lib/rules-engine.ts interface Rule { id: string; name: string; condition: RuleCondition; actions: RuleAction[]; enabled: boolean; priority: number; } interface RuleCondition { type: 'simple' | 'compound'; field?: string; operator?: '>' | '<' | '==' | '!=' | '>=' | '<='; value?: any; logic?: 'AND' | 'OR'; conditions?: RuleCondition[]; } interface RuleAction { type: 'publish' | 'alert' | 'device_command' | 'log'; [key: string]: any; } class RulesEngine { private rules: Rule[] = []; constructor() { this.loadDefaultRules(); } private loadDefaultRules(): void { this.rules = [ // High Temperature Alert { id: 'high-temp-alert', name: 'High Temperature Alert', condition: { type: 'simple', field: 'temperature', operator: '>', value: 35 }, actions: [ { type: 'alert', message: 'High temperature detected', severity: 'warning' }, { type: 'publish', topic: 'alerts/temperature', payload: { alert: 'high_temperature' } } ], enabled: true, priority: 1 }, // Low Battery Warning { id: 'low-battery', name: 'Low Battery Warning', condition: { type: 'simple', field: 'battery', operator: '<', value: 20 }, actions: [ { type: 'alert', message: 'Device battery low', severity: 'info' } ], enabled: true, priority: 2 }, // Motion → Light On { id: 'motion-light', name: 'Motion Activated Light', condition: { type: 'compound', logic: 'AND', conditions: [ { type: 'simple', field: 'occupancy', operator: '==', value: true }, { type: 'simple', field: 'ambient_light', operator: '<', value: 100 } ] }, actions: [ { type: 'device_command', deviceId: 'hallway_light', command: { state: 'ON', brightness: 80 } } ], enabled: true, priority: 1 } ]; } evaluate(topic: string, data: any): RuleAction[] { const actions: RuleAction[] = []; // Regeln nach Priorität sortieren const sortedRules = [...this.rules] .filter(r => r.enabled) .sort((a, b) => a.priority - b.priority); for (const rule of sortedRules) { if (this.evaluateCondition(rule.condition, data)) { console.log(`Rule triggered: ${rule.name}`); actions.push(...rule.actions); } } return actions; } private evaluateCondition(condition: RuleCondition, data: any): boolean { if (condition.type === 'simple') { return this.evaluateSimple(condition, data); } if (condition.type === 'compound' && condition.conditions) { const results = condition.conditions.map(c => this.evaluateCondition(c, data) ); return condition.logic === 'AND' ? results.every(r => r) : results.some(r => r); } return false; } private evaluateSimple(condition: RuleCondition, data: any): boolean { const value = data[condition.field!]; if (value === undefined) return false; switch (condition.operator) { case '>': return value > condition.value; case '<': return value < condition.value; case '>=': return value >= condition.value; case '<=': return value <= condition.value; case '==': return value === condition.value; case '!=': return value !== condition.value; default: return false; } } updateRules(newRules: Rule[]): void { this.rules = newRules; console.log(`Updated ${newRules.length} rules`); } addRule(rule: Rule): void { this.rules.push(rule); } removeRule(ruleId: string): void { this.rules = this.rules.filter(r => r.id !== ruleId); } getRules(): Rule[] { return this.rules; } } export { RulesEngine }; ``` --- ## Edge-Cloud Synchronisation ```typescript // lib/edge-cloud-sync.ts interface SyncConfig { syncInterval: number; batchSize: number; retentionLocal: number; // Tage compressionEnabled: boolean; } class EdgeCloudSync { private config: SyncConfig; private pendingSync: any[] = []; private lastSyncTime: Date | null = null; constructor(config: SyncConfig) { this.config = config; } // Daten für Cloud-Sync vorbereiten prepareForSync(data: any[]): any { // Aggregation const aggregated = this.aggregate(data); // Kompression (optional) if (this.config.compressionEnabled) { return this.compress(aggregated); } return aggregated; } private aggregate(data: any[]): any { // Nach Device gruppieren const byDevice = new Map(); data.forEach(item => { const deviceId = item.device_id; if (!byDevice.has(deviceId)) { byDevice.set(deviceId, []); } byDevice.get(deviceId)!.push(item); }); // Aggregierte Statistiken pro Device const result: any[] = []; byDevice.forEach((items, deviceId) => { const temps = items .map(i => i.temperature) .filter(t => t !== undefined); const humidities = items .map(i => i.humidity) .filter(h => h !== undefined); result.push({ device_id: deviceId, period_start: items[0].timestamp, period_end: items[items.length - 1].timestamp, sample_count: items.length, temperature: temps.length > 0 ? { min: Math.min(...temps), max: Math.max(...temps), avg: temps.reduce((a, b) => a + b, 0) / temps.length } : null, humidity: humidities.length > 0 ? { min: Math.min(...humidities), max: Math.max(...humidities), avg: humidities.reduce((a, b) => a + b, 0) / humidities.length } : null }); }); return result; } private compress(data: any): Buffer { const zlib = require('zlib'); const json = JSON.stringify(data); return zlib.gzipSync(json); } // Conflict Resolution bei Cloud-Sync resolveConflict(localData: any, cloudData: any): any { // Last-Write-Wins Strategie const localTime = new Date(localData.updated_at); const cloudTime = new Date(cloudData.updated_at); if (localTime > cloudTime) { return { ...cloudData, ...localData, conflict_resolved: true }; } return { ...localData, ...cloudData, conflict_resolved: true }; } } ``` --- ## Edge Deployment ```yaml # docker-compose.edge.yml version: '3.8' services: edge-gateway: build: ./edge-gateway restart: always environment: - GATEWAY_ID=${GATEWAY_ID} - MQTT_BROKER=mqtt://mosquitto:1883 - CLOUD_ENDPOINT=${CLOUD_MQTT_URL} - INFLUX_URL=http://influxdb:8086 volumes: - ./models:/app/models - ./config:/app/config depends_on: - mosquitto - influxdb mosquitto: image: eclipse-mosquitto:2 ports: - "1883:1883" volumes: - mosquitto-data:/mosquitto/data influxdb: image: influxdb:2.7 ports: - "8086:8086" volumes: - influxdb-data:/var/lib/influxdb2 environment: - DOCKER_INFLUXDB_INIT_MODE=setup - DOCKER_INFLUXDB_INIT_USERNAME=admin - DOCKER_INFLUXDB_INIT_PASSWORD=edgepassword - DOCKER_INFLUXDB_INIT_ORG=edge - DOCKER_INFLUXDB_INIT_BUCKET=sensors volumes: mosquitto-data: influxdb-data: ``` --- ## Edge vs Cloud Comparison | Aspect | Edge Computing | Cloud Computing | |--------|---------------|-----------------| | **Latency** | 1-10 ms | 50-200 ms | | **Bandwidth** | Minimal (local) | High (all data) | | **Offline** | Fully operational | Limited | | **Privacy** | Data stays local | Data in cloud | | **Compute** | Limited | Unlimited | | **Storage** | Limited | Unlimited | | **Cost** | Hardware upfront | Pay-per-use | | **ML Training** | Not practical | Ideal | | **ML Inference** | Real-time | Batch | --- ## Fazit Edge Computing für IoT bietet: 1. **Low Latency**: Lokale Entscheidungen in Millisekunden 2. **Offline-Fähigkeit**: Funktioniert ohne Cloud 3. **Datenschutz**: Sensible Daten bleiben lokal 4. **Bandbreite**: Nur aggregierte Daten zur Cloud Die optimale Architektur kombiniert Edge und Cloud. --- ## Bildprompts 1. "Edge computing diagram with local processing near sensors" 2. "IoT gateway device processing data streams locally" 3. "Fog computing layers between devices and cloud" --- ## Quellen - [AWS IoT Greengrass](https://aws.amazon.com/greengrass/) - [Azure IoT Edge](https://azure.microsoft.com/en-us/products/iot-edge) - [TensorFlow Lite for Edge](https://www.tensorflow.org/lite) - [InfluxDB Edge Replication](https://www.influxdata.com/blog/simplify-industrial-iot-use-influxdb-edge-replication/)