Serverless im Maßstab: IoT-Wearables mit Lambda und Kinesis

Genau hier überzeugt die serverlose Architektur auf AWS. Durch die Kombination von Diensten wie AWS IoT Core, Amazon Kinesis Data Streams und AWS Lambda lässt sich eine robuste Datenverarbeitungspipeline mit hohem Durchsatz aufbauen, die automatisch skaliert und den Betriebsaufwand minimiert.

In diesem Deep Dive entwerfen und bauen wir eine vollständige serverlose Pipeline zur Verarbeitung simulierter Daten von IoT-Wearables. Wir behandeln alles: von der sicheren Datenaufnahme über die Echtzeitverarbeitung bis zur Speicherung für künftige Analysen.

Voraussetzungen:

• Ein AWS-Konto mit IAM-Benutzerberechtigungen zum Erstellen und Verwalten von Ressourcen für IoT Core, Kinesis, Lambda und S3.
• Node.js und npm sind auf Ihrem lokalen Rechner installiert.
• Die AWS CLI ist auf Ihrem lokalen Rechner konfiguriert.
• Grundlegendes Verständnis von JavaScript (Node.js) und Cloud-Computing-Konzepten.

Das Problem verstehen

Stellen Sie sich ein Unternehmen vor, das mit einem eigenen Wearable einen Gesundheitsüberwachungsdienst anbietet. Jedes Gerät erfasst die Herzfrequenz und die Schritte und sendet diese Daten alle paar Sekunden in die Cloud. Bei einer wachsenden Nutzerbasis muss das System Folgendes bewältigen:

• Hoher Durchsatz: Ein massives und stark schwankendes Datenvolumen von einer großen Zahl von Geräten.
• Echtzeitverarbeitung: Die Notwendigkeit, Daten bei ihrem Eintreffen zu analysieren, um Anomalien zu erkennen oder Alarme auszulösen.
• Skalierbarkeit: Die Fähigkeit, nahtlos von einigen Hundert Geräten auf Millionen zu skalieren, ohne manuellen Eingriff.
• Kosteneffizienz: Eine Lösung, die die hohen Kosten ungenutzter, fest bereitgestellter Infrastruktur vermeidet. Das nutzungsbasierte Abrechnungsmodell von Serverless ist hier ideal.

Herkömmliche serverbasierte Architekturen erfordern erheblichen Aufwand bei der Bereitstellung, Skalierung und Verwaltung einer Serverflotte und führen zu hohen Betriebskosten und hoher Komplexität. Unser serverloser Ansatz löst diese Herausforderungen elegant.

Die serverlose Architektur

Hier ein Überblick auf hoher Ebene über unsere Architektur:

1. AWS IoT Core: Dient als sicherer Eingangspunkt für unsere IoT-Geräte. Er authentifiziert Geräte und verwendet eine Rules Engine, um eingehende Daten weiterzuleiten.
2. Amazon Kinesis Data Streams: Ein vollständig verwalteter Dienst für Echtzeit-Datenstreaming. Er puffert die großen Datenmengen aus IoT Core und entkoppelt die Aufnahme von der Verarbeitung.
3. AWS Lambda: Die serverlose Rechenengine. Eine Lambda-Funktion wird durch neue Daten im Kinesis-Stream ausgelöst, um die Daten zu verarbeiten, zu transformieren und anzureichern.
4. Amazon S3 (über Kinesis Data Firehose): Für dauerhafte Langzeitspeicherung. Wir verwenden Kinesis Data Firehose, um verarbeitete Daten einfach zu bündeln und in einem S3-Bucket für Analysen zu speichern.

(Bildvorschlag: Ein klares Architekturdiagramm, das den Datenfluss von IoT-Gerät -> IoT Core -> Kinesis Data Streams -> Lambda -> Kinesis Data Firehose -> S3 zeigt)

Architektur der serverlosen IoT-Pipeline

Das folgende Diagramm zeigt unseren vollständigen serverlosen Datenfluss:

Diese serverlose Architektur skaliert automatisch von 10 Geräten auf 10 Millionen, ohne betrieblichen Eingriff.

Voraussetzungen

Bevor wir mit dem Aufbau beginnen, richten wir unsere Umgebung ein.

1. AWS CDK installieren: Wir verwenden das AWS Cloud Development Kit (CDK), um unsere Cloud-Infrastruktur als Code zu definieren.

   npm install -g aws-cdk
   

2. Ihre AWS-Umgebung für CDK bootstrappen:

   cdk bootstrap aws://ACCOUNT-NUMBER/REGION
   

Kinesis Data Stream als Aufnahmepuffer einrichten

Zuerst benötigen wir eine „Leitung“ für unsere Streaming-Daten. Kinesis Data Streams eignet sich dafür hervorragend. Die grundlegende Skalierungseinheit in Kinesis ist ein „Shard“. Jeder Shard stellt eine bestimmte Kapazität für die Datenaufnahme und den Datenausgang bereit.

Was wir tun

Wir erstellen einen Kinesis Data Stream, der als Puffer für die eingehenden IoT-Daten dient. Dadurch wird die Datenaufnahme von der Verarbeitungslogik entkoppelt.

Implementierung

Wir können unseren Kinesis-Stream mit dem AWS CDK definieren. Hier ein Beispiel-CDK-Stack in TypeScript:

// lib/iot-pipeline-stack.ts
import * as cdk from 'aws-cdk-lib';
import { Construct } from 'constructs';
import * as kinesis from 'aws-cdk-lib/aws-kinesis';

export class IotPipelineStack extends cdk.Stack {
  constructor(scope: Construct, id: string, props?: cdk.StackProps) {
    super(scope, id, props);

    // Create a Kinesis Data Stream with one shard to start
    const dataStream = new kinesis.Stream(this, 'WearableDataStream', {
      streamName: 'iot-wearable-data-stream',
      shardCount: 1, // Start with one shard and scale as needed
    });
  }
}

So funktioniert es

Dieser CDK-Code definiert einen neuen Kinesis Data Stream namens iot-wearable-data-stream mit einem einzelnen Shard. Für eine Produktionsumgebung mit Tausenden von Geräten würden Sie den shardCount erhöhen. Kinesis kann auch für automatische Skalierung konfiguriert werden.

AWS IoT Core für die sichere Geräteaufnahme konfigurieren

Als Nächstes richten wir AWS IoT Core ein, um Daten von unseren Geräten zu empfangen und an unseren Kinesis-Stream weiterzuleiten. Dazu gehören das Erstellen eines „IoT Thing“, das Generieren von Sicherheitszertifikaten und das Definieren einer Regel.

Was wir tun

Wir erstellen eine Regel in der AWS IoT Core Rules Engine, die auf MQTT-Nachrichten zu einem bestimmten Topic lauscht und sie direkt an unseren Kinesis Data Stream sendet.

Implementierung

1. Ein Thing erstellen: Navigieren Sie in der AWS-IoT-Core-Konsole zu „Manage“ -> „Things“ und erstellen Sie ein neues Thing. Nennen Sie es wearable_device_01.
2. Zertifikate erstellen und anhängen: Folgen Sie den Eingabeaufforderungen der Konsole, um ein Zertifikat für Ihr Gerät zu erstellen. Laden Sie das Zertifikat, den privaten Schlüssel und das Root-CA-Zertifikat herunter. Hängen Sie dieses Zertifikat an Ihr Thing an.
3. Eine Regel erstellen: Gehen Sie zu „Message Routing“ -> „Rules“ und erstellen Sie eine neue Regel.
   • Name: ForwardToKinesisRule
   • Rule-Query-Anweisung: SELECT * FROM 'iot/wearable/data'
   • Aktion: Wählen Sie „Send a message to a Kinesis Stream“. Wählen Sie den von Ihnen erstellten iot-wearable-data-stream.
   • Partition Key: Verwenden Sie ${newuuid()}, um eine gleichmäßige Verteilung der Daten über die Shards sicherzustellen.
   • IAM-Rolle: Erstellen Sie eine neue IAM-Rolle, die IoT Core die Berechtigung erteilt, Datensätze in Ihren Kinesis-Stream zu schreiben.

So funktioniert es

Die Rules Engine ist eine leistungsstarke Funktion von AWS IoT Core. Sie wertet jede Nachricht aus, die zu MQTT-Topics veröffentlicht wird. Entspricht eine Nachricht der SQL-ähnlichen Abfrage in einer Regel (SELECT * FROM 'iot/wearable/data'), löst sie die zugehörige Aktion aus. In unserem Fall leitet sie die gesamte Nachrichtennutzlast an unseren Kinesis-Stream weiter.

Lambda-Funktion für die Echtzeit-Datenverarbeitung erstellen

Nun zum Kern unserer Lösung. Wir schreiben eine Lambda-Funktion, die Datensatzstapel aus dem Kinesis-Stream liest, verarbeitet und zur Speicherung weiterleitet.

Was wir tun

Wir erstellen eine Node.js-Lambda-Funktion, die durch unseren Kinesis-Stream ausgelöst wird. Die Funktion analysiert die eingehenden Daten, führt eine einfache Transformation durch (z. B. das Hinzufügen eines Zeitstempels) und protokolliert das Ergebnis.

Implementierung

Hier ist der Code für unsere Lambda-Funktion:

// src/data-processor/index.js
exports.handler = async (event) => {
  console.log('Processing Kinesis event:', JSON.stringify(event, null, 2));

  for (const record of event.Records) {
    // Kinesis data is base64 encoded, so we need to decode it.
    const payload = Buffer.from(record.kinesis.data, 'base64').toString('ascii');
    const wearableData = JSON.parse(payload);

    // Perform a simple transformation
    const processedData = {
      ...wearableData,
      processedAt: new Date().toISOString(),
      status: wearableData.heartRate > 140 ? 'HIGH_HEART_RATE' : 'NORMAL'
    };

    console.log('Processed record:', processedData);
    // In a real application, you would send this to a database or another service.
  }

  return `Successfully processed ${event.Records.length} records.`;
};

Wir müssen außerdem unseren CDK-Stack aktualisieren, um die Lambda-Funktion zu erstellen und den Kinesis-Stream als ihre Ereignisquelle festzulegen.

// lib/iot-pipeline-stack.ts (continued)
import * as lambda from 'aws-cdk-lib/aws-lambda';
import { KinesisEventSource } from 'aws-cdk-lib/aws-lambda-event-sources';

// ... inside the IotPipelineStack constructor

const dataProcessor = new lambda.Function(this, 'DataProcessorFunction', {
  runtime: lambda.Runtime.NODEJS_18_X,
  handler: 'index.handler',
  code: lambda.Code.fromAsset('src/data-processor'),
});

// Grant the Lambda function permission to read from the stream
dataStream.grantRead(dataProcessor);

// Add the Kinesis stream as an event source for the Lambda function
dataProcessor.addEventSource(new KinesisEventSource(dataStream, {
  batchSize: 100, // Process records in batches of 100
  startingPosition: lambda.StartingPosition.LATEST,
}));

So funktioniert es

Lambda integriert sich mit Kinesis, indem es den Stream nach neuen Datensätzen abfragt. Wenn neue Datensätze verfügbar sind, ruft Lambda unsere Funktion mit einem Stapel davon auf. Unser Code durchläuft die Datensätze, dekodiert die Base64-kodierten Daten, analysiert das JSON und fügt einen Verarbeitungszeitstempel sowie einen Status auf Grundlage der Herzfrequenz hinzu. Diese verarbeiteten Daten können anschließend an eine Datenbank wie DynamoDB oder, wie wir als Nächstes sehen, zur Langzeitspeicherung an S3 gesendet werden.

Verarbeitete Daten mit Kinesis Data Firehose in S3 archivieren

Während unser Lambda Daten in Echtzeit verarbeitet, müssen wir sie auch für historische Analysen speichern. Kinesis Data Firehose ist eine einfache Möglichkeit, Streaming-Daten in Datenspeicher und Analysewerkzeuge zu laden.

Implementierung

Wir können einen Firehose-Delivery-Stream einrichten, der unsere verarbeiteten Lambda-Daten übernimmt und in S3 archiviert. Die Lambda-Funktion müsste so angepasst werden, dass sie die verarbeiteten Datensätze in den Firehose-Stream schreibt.

Ein direkteres und häufigeres Muster besteht darin, dass die IoT-Regel Daten sowohl an einen Kinesis Data Stream (für die Echtzeitverarbeitung) als auch an einen Kinesis Data Firehose-Stream (für die Archivierung) sendet. Dadurch vereinfacht sich die Rolle der Lambda-Funktion, sodass sie sich ausschließlich auf die Echtzeitlogik konzentriert.

Alles zusammenführen: Ein Gerät simulieren

Testen wir unsere Pipeline, indem wir ein Wearable simulieren, das Daten sendet. Wir verwenden das AWS SDK für Node.js, um MQTT-Nachrichten zu veröffentlichen.

// scripts/device-simulator.js
const { IoTDataPlaneClient, PublishCommand } = require("@aws-sdk/client-iot-data-plane");

const client = new IoTDataPlaneClient({ region: "YOUR_REGION" });
const deviceId = "wearable_device_01";

async function publishData() {
  const heartRate = 60 + Math.floor(Math.random() * 80); // 60-140 bpm
  const steps = Math.floor(Math.random() * 5000);

  const payload = {
    deviceId,
    heartRate,
    steps,
    timestamp: new Date().toISOString(),
  };

  const command = new PublishCommand({
    topic: "iot/wearable/data",
    qos: 1,
    payload: JSON.stringify(payload),
  });

  try {
    await client.send(command);
    console.log("Published data:", payload);
  } catch (error) {
    console.error("Error publishing data:", error);
  }
}

// Publish data every 5 seconds
setInterval(publishData, 5000);

Führen Sie dieses Skript aus, und Sie sollten sehen, wie die Daten durch die Pipeline fließen. Prüfen Sie die CloudWatch-Protokolle Ihrer Lambda-Funktion, um die verarbeiteten Datensätze zu sehen!

Leistungsaspekte

• Verwaltung der Kinesis-Shards: Die Anzahl der Shards ist Ihr wichtigster Hebel für die Skalierung. Überwachen Sie die Metriken IncomingBytes und IncomingRecords Ihres Streams und fügen Sie bei Bedarf Shards hinzu, um die Last zu bewältigen.
• Lambda-Parallelität: Kinesis ruft pro Shard eine Lambda-Funktion gleichzeitig auf. Um die parallele Verarbeitung zu erhöhen, müssen Sie die Shard-Anzahl erhöhen.
• Stapelgröße: Passen Sie die batchSize in der Zuordnung Ihrer Lambda-Ereignisquelle an. Größere Stapel sind kosteneffizienter, da sie die Zahl der Lambda-Aufrufe verringern, erhöhen aber auch die Latenz.

Bewährte Sicherheitspraktiken

• Prinzip der geringsten Berechtigung: Stellen Sie sicher, dass IAM-Rollen für IoT Core und Lambda nur die Berechtigungen haben, die sie benötigen. Die IoT-Regel sollte nur in den spezifischen Kinesis-Stream schreiben dürfen, und die Lambda-Funktion sollte nur Lesezugriff haben.
• Eindeutige Gerätezertifikate: Jedes IoT-Gerät sollte über ein eigenes eindeutiges Zertifikat für eine sichere und individuelle Authentifizierung verfügen.
• Datenverschlüsselung: Kinesis Data Streams verschlüsselt Daten im Ruhezustand. Stellen Sie sicher, dass Sie die Verschlüsselung während der Übertragung aktivieren, wenn Sie Daten von Geräten veröffentlichen.

Fazit

Wir haben erfolgreich eine hoch skalierbare, kosteneffiziente und robuste serverlose Pipeline zur Verarbeitung von IoT-Wearable-Daten entworfen und aufgebaut. Durch den Einsatz von AWS IoT Core, Kinesis Data Streams und AWS Lambda haben wir eine Architektur geschaffen, die große Datenmengen bewältigen kann, ohne den Aufwand der Serververwaltung. Dieser ereignisgesteuerte Ansatz ermöglicht es Entwicklerinnen und Entwicklern, sich auf Funktionen zu konzentrieren, nicht auf die Verwaltung der Infrastruktur.

Auswirkung auf die Gesundheit: Serverlose IoT-Pipelines ermöglichen die Echtzeitverarbeitung von über 10 Millionen Geräteereignissen mit einer Latenz unter 100 ms für die Anomalieerkennung. Gesundheitseinrichtungen, die ähnliche Architekturen nutzen, berichten von 45–60 % schnelleren Reaktionszeiten bei kardialen Ereignissen und der Fähigkeit, abnorme Muster 3–5 Stunden früher zu erkennen als Batch-Verarbeitungssysteme. Die automatische Skalierung von 10 auf 10 Millionen Geräte gewährleistet eine Verfügbarkeit von 99,9 % bei Belastungsspitzen der Gesundheitsüberwachung.

Ihre nächsten Schritte:

• Ersetzen Sie das console.log in Lambda durch Code, der verarbeitete Daten zur schnellen Abfrage in einer DynamoDB-Tabelle speichert.
• Ergänzen Sie die Lambda-Funktion um Logik zur Anomalieerkennung, um Alarme über Amazon SNS zu senden.
• Erstellen Sie mit Amazon QuickSight ein Dashboard, um die in S3 gespeicherten Daten zu visualisieren.

Ressourcen

• AWS IoT Core Documentation
• Amazon Kinesis Data Streams Developer Guide
• Using AWS Lambda with Amazon Kinesis - AWS Documentation
• Serverless Patterns Collection on Serverless Land
• Verwandte Artikel:
  • Real-Time Pipeline with Kafka & Flink – Stream-Verarbeitung im großen Maßstab
  • Real-Time Dashboard with React & Node.js – WebSocket-Datenvisualisierung

Kurzreferenz

Häufig gestellte Fragen

F: Wie weiß ich, ob das auf mich zutrifft? A: Besprechen Sie Ihre persönliche und familiäre Krankengeschichte mit Ihrer Ärztin oder Ihrem Arzt, um Ihre individuellen Risikofaktoren zu beurteilen und geeignete nächste Schritte festzulegen.

F: Was sollte ich meine Ärztin oder meinen Arzt fragen? A: Gute Fragen sind: „Was bedeuten meine Ergebnisse?“, „Welche Möglichkeiten habe ich?“, „Welche Lebensstiländerungen würden helfen?“ und „Wann sollte ich zur Nachkontrolle kommen?“

F: Gibt es Lebensstiländerungen, die helfen können? A: Die meisten Erkrankungen profitieren von grundlegenden Maßnahmen: ausgewogene Ernährung, regelmäßige körperliche Aktivität (150+ Minuten/Woche), erholsamer Schlaf (7–9 Stunden), Stressbewältigung und Verzicht auf Tabak.

Haftungsausschluss

Die in diesem Artikel vorgestellten Algorithmen und Verfahren dienen ausschließlich der technischen Weiterbildung. Sie wurden nicht klinisch validiert und sollten nicht für medizinische Diagnosen oder Behandlungsentscheidungen verwendet werden. Holen Sie medizinischen Rat stets bei qualifizierten medizinischen Fachpersonen ein.