Ereignisgesteuerte Architektur für Wearable-Daten

Im Zeitalter der vernetzten Gesundheit erzeugen Wearable-Geräte eine Flut von Daten – Herzfrequenz, Aktivitätsniveau, Schlafmuster und vieles mehr. Für Healthtech-Unternehmen sind diese Daten eine Goldgrube für die Bereitstellung von Gesundheitseinblicken in Echtzeit, personalisierten Rückmeldungen und kritischen Warnungen. Der Aufbau eines Systems, das diesen hohen Datendurchsatz und diese hohe Geschwindigkeit von Tausenden von Geräten zuverlässig aufnehmen, verarbeiten und analysieren kann, ist jedoch eine erhebliche technische Herausforderung.

Dieser Artikel führt Sie durch das Design und den Aufbau einer skalierbaren, ereignisgesteuerten Architektur für genau diesen Zweck. Wir werden eine Pipeline erstellen, die einen massiven Zustrom von Wearable-Daten bewältigen, sie in Echtzeit auf Anomalien verarbeiten und effizient für verschiedene Analysetypen speichern kann.

Was wir bauen werden:

Wir werden eine Pipeline entwerfen, in der Wearable-Daten an einen zentralen Message Broker (Apache Kafka) gesendet werden. Von dort aus werden verschiedene Microservices diese Daten verarbeiten, um:

1. Die rohen Sensormesswerte zu normalisieren.
2. Anomalien in Echtzeit zu erkennen (z. B. einen plötzlichen Herzfrequenzsprung) unter Verwendung von Apache Flink.
3. Die Zeitreihen-Vitalzeichen in TimescaleDB zu speichern.
4. Rohereignisprotokolle in Elasticsearch für Prüfung und Suche zu archivieren.

Diese Architektur ist nicht nur ein theoretisches Konzept; sie ist ein praktischerBlueprint, der von führenden Technologieunternehmen verwendet wird, um reaktionsschnelle und widerstandsfähige Systeme zu entwickeln.

Voraussetzungen:

• Grundlegendes Verständnis von Java/Python und Docker.
• Vertrautheit mit Konzepten wie Microservices und Pub/Sub-Messaging.
• Docker und Docker Compose auf Ihrem Rechner installiert.

Warum dies für Entwickler relevant ist:

Traditionelle Anfrage-Antwort-Architekturen brechen unter dem konstanten Datenstrom von IoT-Geräten zusammen. Ein ereignisgesteuerter Ansatz ermöglicht es Ihnen, entkoppelte, skalierbare Systeme zu erstellen, bei denen Komponenten asynchron kommunizieren, was Ihre Anwendung widerstandsfähiger und einfacher zu warten macht.

Das Problem verstehen

Die Kernherausforderung bei Wearable-Daten ist dreifach: Volumen, Geschwindigkeit und Vielfalt.

• Volumen: Tausende von Geräten, die jeweils alle paar Sekunden Daten senden, summieren sich schnell auf Millionen von Ereignissen pro Tag.
• Geschwindigkeit: Die Daten müssen mit sehr geringer Latenz verarbeitet werden, um rechtzeitige Rückmeldungen oder Warnungen zu geben. Ein Batch-Verarbeitungssystem, das über Nacht läuft, kommt nicht in Frage.
• Vielfalt: Die Daten liegen in verschiedenen Formen vor. Vitalzeichen (wie Herzfrequenz) sind Zeitreihendaten, perfekt für eine Zeitreihendatenbank. Systemprotokolle und Fehlermeldungen sind unstrukturierter Text, besser geeignet für eine Suchmaschine wie Elasticsearch.

Eine monolithische Anwendung würde mit diesen konkurrierenden Anforderungen kämpfen. Eine ereignisgesteuerte Architektur hingegen brilliert in diesem Szenario. Durch die Verwendung eines zentralen Message Brokers wie Apache Kafka können wir ein "zentrales Nervensystem" für unsere Daten schaffen. Verschiedene Dienste können dann auf diesen Stream zugreifen, um spezialisierte Aufgaben unabhängig auszuführen.

Architekturübersicht

Das folgende Diagramm zeigt, wie Daten von Wearables durch unsere Verarbeitungspipeline fließen:

Diese Architektur ermöglicht Anomalieerkennung in unter 100 ms für kritische Gesundheitsereignisse wie Arrhythmieerkennung und erlaubt eine sofortige Benachrichtigung der Gesundheitsdienstleister, wenn die Herzfrequenz eines Patienten gefährliche Schwellenwerte überschreitet.

Voraussetzungen

Bevor wir mit dem Bau beginnen, richten wir unsere Umgebung ein. Wir werden Docker Compose verwenden, um alle notwendigen Dienste zu starten: Kafka, Flink, TimescaleDB und Elasticsearch.

Hinweis: Dieses Beispiel verwendet synthetische/simulierte Daten zur Demonstration. In der Produktion stellen Sie sicher, dass alle Gesundheitsdaten anonymisiert und gemäß HIPAA/GDPR behandelt werden.

Infrastruktur-Setup mit Docker Compose

Erstellen Sie eine Datei namens docker-compose.yml:

# docker-compose.yml
version: '3.8'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.3.0
    container_name: zookeeper
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000

  kafka:
    image: confluentinc/cp-kafka:7.3.0
    container_name: kafka
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_INTERNAL:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092,PLAINTEXT_INTERNAL://kafka:29092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1

  timescaledb:
    image: timescale/timescaledb:latest-pg14
    container_name: timescaledb
    ports:
      - "5432:5432"
    environment:
      - POSTGRES_USER=user
      - POSTGRES_PASSWORD=password
      - POSTGRES_DB=wearables
    volumes:
      - ./init-db.sql:/docker-entrypoint-initdb.d/init-db.sql

  elasticsearch:
    image: docker.elastic.co/elasticsearch/elasticsearch:8.5.0
    container_name: elasticsearch
    ports:
      - "9200:9200"
    environment:
      - "discovery.type=single-node"
      - "xpack.security.enabled=false"

  flink-jobmanager:
    image: apache/flink:1.16.0-scala_2.12
    container_name: flink-jobmanager
    ports:
      - "8081:8081"
    command: jobmanager
    environment:
      - |
        FLINK_PROPERTIES=
        jobmanager.rpc.address: flink-jobmanager

  flink-taskmanager:
    image: apache/flink:1.16.0-scala_2.12
    container_name: flink-taskmanager
    depends_on:
      - flink-jobmanager
    command: taskmanager
    scale: 1
    environment:
      - |
        FLINK_PROPERTIES=
        jobmanager.rpc.address: flink-jobmanager
        taskmanager.numberOfTaskSlots: 2

Wir benötigen auch ein SQL-Skript zur Initialisierung unserer TimescaleDB. Erstellen Sie init-db.sql:

-- init-db.sql
CREATE TABLE vitals (
    time TIMESTAMPTZ NOT NULL,
    device_id VARCHAR(50) NOT NULL,
    heart_rate INTEGER,
    steps INTEGER
);

SELECT create_hypertable('vitals', 'time');

Führen Sie nun docker-compose up -d in Ihrem Terminal aus. Dies startet die gesamte notwendige Infrastruktur im Hintergrund.

Aufnahme von Wearable-Daten mit einem Kafka-Producer

Was wir tun

Zunächst müssen wir simulieren, dass Wearables Daten in unser System senden. Wir schreiben ein einfaches Python-Skript, das als Kafka-Producer fungiert. Dieses Skript generiert Mock-Wearable-Daten und veröffentlicht sie in einem Kafka-Topic namens raw_wearable_data.

Implementierung

Erstellen Sie ein Python-Skript producer.py:

# producer.py
import json
import time
import random
from kafka import KafkaProducer

# Configuration
KAFKA_TOPIC = 'raw_wearable_data'
KAFKA_BOOTSTRAP_SERVERS = 'localhost:9092'

# List of simulated device IDs
DEVICES = [f"device_{i}" for i in range(10)]

def create_producer():
    """Creates and returns a Kafka producer."""
    return KafkaProducer(
        bootstrap_servers=KAFKA_BOOTSTRAP_SERVERS,
        value_serializer=lambda v: json.dumps(v).encode('utf-8')
    )

def generate_data(device_id):
    """Generates a single data point for a wearable device."""
    return {
        'deviceId': device_id,
        'timestamp': int(time.time() * 1000),
        'heartRate': random.randint(55, 120),
        'steps': random.randint(0, 100),
        'lat': round(random.uniform(34.0, 36.0), 5),
        'lon': round(random.uniform(-118.0, -120.0), 5),
    }

def main():
    producer = create_producer()
    print("🚀 Starting wearable data producer...")
    try:
        while True:
            device_id = random.choice(DEVICES)
            data = generate_data(device_id)

            # Occasionally send an anomalous heart rate
            if random.random() < 0.05: # 5% chance
                data['heartRate'] = random.randint(180, 200)
                print(f"❗️ Anomaly sent for {device_id}: {data}")

            producer.send(KAFKA_TOPIC, value=data)
            print(f"Sent data for {device_id}: {data}")
            time.sleep(1)
    except KeyboardInterrupt:
        print("🛑 Producer stopped.")
    finally:
        producer.close()

if __name__ == "__main__":
    main()

Wie es funktioniert

Das Skript verwendet die kafka-python-Bibliothek, um sich mit unserem Kafka-Container zu verbinden. Es generiert kontinuierlich JSON-Nutzlasten, die Sensormesswerte darstellen, und sendet sie an das raw_wearable_data-Topic. Wir haben auch eine kleine Wahrscheinlichkeit eingebaut, eine anomal hohe Herzfrequenz zu generieren, um unsere Anomalieerkennung später zu testen.

Um dies auszuführen, müssen Sie die Bibliothek installieren: pip install kafka-python

Führen Sie dann das Skript aus: python producer.py

Erkennung kardialer Anomalien in Echtzeit mit Apache Flink

Was wir tun

Dies ist der Kern unserer Echtzeitverarbeitung. Wir werden einen einfachen Apache Flink-Job erstellen, der aus dem raw_wearable_data-Topic liest, Herzfrequenzen über einem bestimmten Schwellenwert überprüft und alle Anomalien in einem neuen Kafka-Topic namens heart_rate_alerts veröffentlicht.

Implementierung

Wir werden den Flink SQL Client verwenden, der es uns ermöglicht, unsere Stream-Verarbeitungslogik mit vertrautem SQL zu definieren.

1. Zugriff auf den Flink SQL Client: Öffnen Sie ein Terminal und verbinden Sie sich mit dem Flink-Jobmanager-Container: docker exec -it flink-jobmanager /bin/bash

2. Starten Sie den SQL Client: Führen Sie im Container aus: ./bin/sql-client.sh

3. Definieren Sie Quell- und Zieltabellen: Im Flink SQL-Prompt definieren wir unsere Tabellen. Dies erstellt keine Datenbanktabellen; es teilt Flink mit, wie es sich mit Kafka verbinden und die Daten interpretieren soll.

   -- Flink SQL
   -- Define the source table pointing to our raw data topic
   CREATE TABLE raw_data (
     `deviceId` STRING,
     `timestamp` BIGINT,
     `heartRate` INT,
     `steps` INT,
     `lat` DOUBLE,
     `lon` DOUBLE
   ) WITH (
     'connector' = 'kafka',
     'topic' = 'raw_wearable_data',
     'properties.bootstrap.servers' = 'kafka:29092',
     'properties.group.id' = 'flink-consumer-group',
     'format' = 'json',
     'scan.startup.mode' = 'latest-offset'
   );
   
   -- Define the sink table for publishing alerts
   CREATE TABLE heart_rate_alerts (
     `deviceId` STRING,
     `alert_time` TIMESTAMP(3),
     `heartRate` INT
   ) WITH (
     'connector' = 'kafka',
     'topic' = 'heart_rate_alerts',
     'properties.bootstrap.servers' = 'kafka:29092',
     'format' = 'json'
   );
   

4. Erstellen und Ausführen des Anomalieerkennungs-Jobs: Schließlich schreiben wir eine einfache INSERT INTO ... SELECT-Anweisung. Dies ist eine kontinuierliche Abfrage, die endlos läuft und Daten verarbeitet, sobald sie eintreffen.

   -- This is the actual streaming job!
   INSERT INTO heart_rate_alerts
   SELECT
     deviceId,
     TO_TIMESTAMP_LTZ(`timestamp`, 3) AS alert_time,
     heartRate
   FROM raw_data
   WHERE heartRate > 170;
   

Wie es funktioniert

Flink liest kontinuierlich JSON-Nachrichten aus dem raw_wearable_data-Topic. Für jeden Datensatz prüft es, ob das heartRate-Feld größer als 170 ist. Wenn ja, wird ein neuer JSON-Datensatz erstellt und im heart_rate_alerts-Topic veröffentlicht. Dieser gesamte Prozess erfolgt mit Millisekunden-Latenz. Flinks zustandsbehaftete Verarbeitungsfähigkeiten ermöglichen eine viel komplexere Logik, wie z. B. die Erkennung von Anomalien basierend auf einem gleitenden Durchschnitt über die Zeit.

Persistenz von Gesundheitsdaten in spezialisierten Datenbanken

Was wir tun

Unser letzter Schritt ist es, die verarbeiteten Daten aus Kafka in Datenbanken zu übertragen, die für unsere zwei Hauptanwendungsfälle optimiert sind: Zeitreihenanalyse und Protokollsuche. Wir werden zwei separate Consumer-Dienste schreiben.

1. Vitals Consumer: Persistiert Vitalzeichen (Herzfrequenz, Schritte) in TimescaleDB.
2. Logging Consumer: Speichert die rohen, unverarbeiteten Ereignisdaten in Elasticsearch zur Prüfung.

Implementierung: Vitals Consumer für TimescaleDB

Dieses Python-Skript verarbeitet Daten aus dem raw_wearable_data-Topic und schreibt in unsere vitals-Hypertable in TimescaleDB.

# vitals_consumer.py
import json
import psycopg2
from kafka import KafkaConsumer
from datetime import datetime, timezone

# Kafka Configuration
KAFKA_TOPIC = 'raw_wearable_data'
KAFKA_BOOTSTRAP_SERVERS = 'localhost:9092'
KAFKA_CONSUMER_GROUP = 'timescale-consumer-group'

# TimescaleDB Configuration
DB_HOST = 'localhost'
DB_PORT = '5432'
DB_NAME = 'wearables'
DB_USER = 'user'
DB_PASS = 'password'

def create_consumer():
    """Creates and returns a Kafka consumer."""
    return KafkaConsumer(
        KAFKA_TOPIC,
        bootstrap_servers=KAFKA_BOOTSTRAP_SERVERS,
        group_id=KAFKA_CONSUMER_GROUP,
        auto_offset_reset='earliest',
        value_deserializer=lambda x: json.loads(x.decode('utf-8'))
    )

def main():
    consumer = create_consumer()
    db_conn = psycopg2.connect(host=DB_HOST, port=DB_PORT, dbname=DB_NAME, user=DB_USER, password=DB_PASS)
    cursor = db_conn.cursor()

    print("🚀 Starting vitals consumer...")
    try:
        for message in consumer:
            data = message.value
            try:
                dt_object = datetime.fromtimestamp(data['timestamp'] / 1000, tz=timezone.utc)

                insert_query = """
                INSERT INTO vitals (time, device_id, heart_rate, steps)
                VALUES (%s, %s, %s, %s);
                """
                cursor.execute(insert_query, (dt_object, data['deviceId'], data['heartRate'], data['steps']))
                db_conn.commit()

                print(f"✅ Inserted vitals for {data['deviceId']} into TimescaleDB")
            except (KeyError, TypeError) as e:
                print(f"⚠️ Skipping malformed message: {data}. Error: {e}")

    except KeyboardInterrupt:
        print("🛑 Vitals consumer stopped.")
    finally:
        cursor.close()
        db_conn.close()
        consumer.close()

if __name__ == "__main__":
    main()

Sie benötigen: pip install psycopg2-binary kafka-python

Implementierung: Logging Consumer für Elasticsearch

Dieses Skript verarbeitet dasselbe Topic, führt aber das vollständige, rohe JSON-Dokument in Elasticsearch ein.

# logging_consumer.py
import json
from kafka import KafkaConsumer
from elasticsearch import Elasticsearch

# Kafka Configuration
KAFKA_TOPIC = 'raw_wearable_data'
KAFKA_BOOTSTRAP_SERVERS = 'localhost:9092'
KAFKA_CONSUMER_GROUP = 'elastic-consumer-group'

# Elasticsearch Configuration
ES_HOST = 'localhost'
ES_PORT = 9200
ES_INDEX = 'wearable_logs'

def create_consumer():
    """Creates and returns a Kafka consumer."""
    return KafkaConsumer(
        KAFKA_TOPIC,
        bootstrap_servers=KAFKA_BOOTSTRAP_SERVERS,
        group_id=KAFKA_CONSUMER_GROUP,
        auto_offset_reset='earliest',
        value_deserializer=lambda x: json.loads(x.decode('utf-8'))
    )

def main():
    consumer = create_consumer()
    es = Elasticsearch([{'host': ES_HOST, 'port': ES_PORT, 'scheme': 'http'}])

    print("🚀 Starting logging consumer...")
    try:
        for message in consumer:
            data = message.value
            try:
                # Add a processing timestamp for context
                data['processingTimestamp'] = datetime.utcnow().isoformat()
                es.index(index=ES_INDEX, document=data)
                print(f"📄 Indexed log for {data['deviceId']} in Elasticsearch")
            except (KeyError, TypeError) as e:
                print(f"⚠️ Skipping malformed message for logging: {data}. Error: {e}")

    except KeyboardInterrupt:
        print("🛑 Logging consumer stopped.")
    finally:
        consumer.close()

if __name__ == "__main__":
    main()

Sie benötigen: pip install elasticsearch kafka-python

Wie es funktioniert

Durch die Verwendung von zwei verschiedenen Consumer-Gruppen (timescale-consumer-group und elastic-consumer-group) ermöglicht Kafka beiden Diensten, den gesamten Datenstrom unabhängig voneinander zu lesen. Dies ist ein Kernprinzip des ereignisgesteuerten Designs: Dienste sind entkoppelt und müssen nichts voneinander wissen. Ein Dienst kann ausfallen, ohne den anderen zu beeinträchtigen.

Alles zusammenfügen

Um die gesamte Pipeline in Aktion zu sehen:

1. Führen Sie docker-compose up -d aus.
2. Starten Sie den Flink SQL-Job wie in Schritt 2 beschrieben.
3. Führen Sie die Python-Consumer in separaten Terminals aus:
   • python vitals_consumer.py
   • python logging_consumer.py
4. Führen Sie den Producer in einem vierten Terminal aus:
   • python producer.py

Sie sehen nun einen Echtzeit-Datenfluss:

• Der Producer gibt Nachrichten aus, während er sie sendet.
• Wenn eine Anomalie gesendet wird, verarbeitet der Flink-Job sie und erstellt eine Warnung.
• Der Vitals Consumer gibt Nachrichten aus, während er Daten in TimescaleDB einfügt.
• Der Logging Consumer gibt Nachrichten aus, während er Dokumente in Elasticsearch indiziert.

Sie können die Daten in den Datenbanken überprüfen:

• TimescaleDB: Verbinden Sie sich mit psql -h localhost -p 5432 -U user -d wearables und führen Sie SELECT * FROM vitals ORDER BY time DESC LIMIT 10; aus.
• Elasticsearch: Öffnen Sie Ihren Browser unter http://localhost:9200/wearable_logs/_search, um die Rohprotokolle zu sehen.

Sicherheits- und Leistungsüberlegungen

• Sicherheit: In einer Produktionsumgebung sollten Sie Authentifizierung und TLS-Verschlüsselung für Kafka, Flink und die Datenbanken aktivieren. Führen Sie in der Produktion niemals ohne Sicherheit aus.
• Leistung: Kafka-Topics sollten partitioniert werden, um eine parallele Verarbeitung zu ermöglichen. Die Anzahl der Flink-Taskmanager kann erweitert werden, um erhöhte Last zu bewältigen.
• Datenschema: Die Verwendung eines Schema-Registries wie Confluent Schema Registry mit Avro- oder Protobuf-Format wird für Produktionssysteme sehr empfohlen, anstatt einfaches JSON. Es gewährleistet Datenkonsistenz und bietet bessere Leistung.

Fazit

Wir haben erfolgreich eine robuste Echtzeit-Datenaufnahme-Pipeline für Wearable-Gerätedaten entworfen und aufgebaut. Durch den Einsatz einer ereignisgesteuerten Architektur mit Kafka als Rückgrat haben wir ein System geschaffen, das skalierbar, widerstandsfähig und flexibel ist.

Wir haben erreicht:

• Entkoppelte Dienste für Aufnahme, Verarbeitung und Speicherung.
• Echtzeit-Anomalieerkennung unter Nutzung der Leistungsfähigkeit von Apache Flink.
• Effiziente Speicherung von Zeitreihen- und Protokolldaten in spezialisierten Datenbanken.

Gesundheitliche Auswirkungen: Diese Architektur ermöglicht Anomalieerkennung bei kardialen Ereignissen in unter 100 ms, was Gesundheitssystemen erlaubt, auf kritische Ereignisse wie Tachykardie nahezu sofort zu reagieren. In Produktionsbereitstellungen haben ähnliche Pipelines die Warnungsreaktionszeiten von Minuten auf unter 200 ms reduziert, was in Notfallszenarien potenziell Leben retten kann.

Dieses Muster ist unglaublich leistungsstark und kann für viele andere Echtzeit-Anwendungsfälle angepasst werden, von Finanzdienstleistungen bis hin zu Logistik und darüber hinaus.

Ressourcen

• Apache Kafka: Offizielle Dokumentation
• Apache Flink: Flink SQL Client Dokumentation
• TimescaleDB: Erste Schritte Anleitung
• Elasticsearch: Python Client Dokumentation
• Verwandte Artikel:
  • Erstellung eines Schlafhypnogramms mit Recharts - Zeitreihen-Schlafdaten visualisieren
  • Echtzeit-Dashboard mit React & Node.js - WebSocket-basierte Live-Datenvisualisierung

Kurzübersicht

Häufig gestellte Fragen

F: Woher weiß ich, ob dies auf mich zutrifft? A: Besprechen Sie Ihre persönliche und familiäre Krankengeschichte mit Ihrem Arzt, um Ihre individuellen Risikofaktoren zu bewerten und geeignete nächste Schritte zu bestimmen.

F: Was sollte ich meinen Arzt fragen? A: Gute Fragen sind: "Was bedeuten meine Ergebnisse?", "Was sind meine Optionen?", "Welche Änderungen des Lebensstils würden helfen?" und "Wann sollte ich nachuntersuchen lassen?"

F: Gibt es Änderungen des Lebensstils, die helfen können? A: Die meisten Gesundheitszustände profitieren von grundlegenden Maßnahmen: ausgewogene Ernährung, regelmäßige körperliche Aktivität (über 150 Minuten/Woche), качественный Schlaf (7-9 Stunden), Stressmanagement und Vermeidung von Tabak.

Haftungsausschluss

Die in diesem Artikel vorgestellten Algorithmen und Techniken dienen ausschließlich technischen Bildungszwecken. Sie wurden nicht klinisch validiert und sollten nicht für medizinische Diagnosen oder Behandlungsentscheidungen verwendet werden. Konsultieren Sie immer qualifizierte Gesundheitsdienstleister für medizinischen Rat.