Scaling Wellness Analytics: A Practical Guide to the CQRS Pattern

In der Welt der Gesundheits- und Wellness-Anwendungen sind Daten der König. Benutzer protokollieren Workouts, verfolgen Mahlzeiten und überwachen ihren Schlaf, was einen ständigen Datenstrom erzeugt. Auf der anderen Seite erwarten sie schöne, aufschlussreiche Dashboards, die ihren Fortschritt über Wochen, Monate oder sogar Jahre hinweg zeigen. Dies stellt eine klassische architektonische Herausforderung dar: Ihre Datenbank muss sowohl für häufige, einfache Schreibvorgänge (z.B. „Mein 10-km-Lauf protokollieren") als auch für komplexe, analytische Lesevorgänge (z.B. „Meinen durchschnittlichen Lauf-Trend, segmentiert nach Wochentag, für das vergangene Jahr anzeigen") optimiert sein.

Eine traditionelle monolithische Architektur, bei der eine einzelne Datenbank beide Aufgaben übernimmt, wird unweigerlich auf einen Engpass stoßen. Das Datenmodell, das perfekt für schnelle Schreibvorgänge ist (normalisiert), ist oft schlecht für schnelle analytische Abfragen, die komplexe Joins und Aggregation erfordern können. Hier kommt das Command Query Responsibility Segregation (CQRS)-Muster ins Spiel.

CQRS ist ein architektonisches Muster, das die Modelle für das Lesen und Schreiben von Daten trennt. Anstatt eines Modells, das alles bearbeitet, haben wir zwei:

• Ein Command-Modell für die Behandlung von Zustandsänderungen (Schreiben, Aktualisieren, Löschen).
• Ein Query-Modell für das Abrufen von Daten.

In diesem Tutorial werden wir ein vereinfachtes Wellness-Analytik-System erstellen, das die Leistungsfähigkeit von CQRS demonstriert. Wir werden eine „Write"-Seite mit einer PostgreSQL-Datenbank und eine „Read"-Seite mit einer denormalisierten ClickHouse-Datenbank erstellen, die alle durch einen Kafka-Event-Bus verbunden sind. Diese Architektur ermöglicht es uns, jeden Teil unseres Systems unabhängig zu skalieren und sowohl schnelles Workout-Protokollieren als auch blitzschnelle Analytik-Dashboards zu gewährleisten. ✨

Das Problem verstehen

Stellen Sie sich eine typische Wellness-App vor. Die Schreiboperationen sind einfach und häufig:

• LogWorkout (userId, type: 'run', duration: 3600, distance: 10)
• AddMeal (userId, name: 'Salad', calories: 450)

Diese werden am besten von einer transaktionalen, normalisierten Datenbank wie PostgreSQL behandelt, um die Datenintegrität zu gewährleisten.

Die Leseoperationen sind jedoch ein anderes Biest. Ein Benutzer möchte sein Dashboard mit Abfragen wie diesen sehen:

• „Wie ist meine insgesamt gelaufene Distanz pro Monat für das letzte Jahr?`
• „Zeigen Sie mir einen wöchentlichen Durchschnitt meiner Kalorienzufuhr.`
• „Vergleichen Sie meine Workout-Dauern an Wochentagen vs. Wochenenden.`

Das Ausführen dieser Abfragen gegen eine normalisierte transaktionale Datenbank kann langsam und ressourcenintensiv sein und erfordert oft mehrere Joins und Berechnungen in Echtzeit. Mit dem Wachstum Ihrer Benutzerbasis und des Datenvolumens werden diese Abfragen Ihre Anwendung zum Stillstand bringen.

Durch die Trennung dieser Bedenken ermöglicht CQRS uns, das beste Werkzeug für jede Aufgabe zu verwenden. Wir können ein Write-Modell haben, das für transaktionale Konsistenz optimiert ist, und ein Read-Modell, das denormalisiert und auf die spezifischen Abfragen zugeschnitten ist, die unser Analytik-Dashboard stellen wird.

CQRS-Architekturmuster

Das folgende Diagramm zeigt, wie Daten durch unser CQRS-basiertes Analytik-System fließen:

Diese Trennung ermöglicht 100x schnellere Analytik-Abfragen im Vergleich zur direkten Abfrage der transaktionalen Datenbank.

Voraussetzungen

Bevor wir beginnen, sollten Sie ein grundlegendes Verständnis von Folgendem haben:

• REST-APIs (wir werden Node.js mit Express für unsere Beispiele verwenden)
• Docker und Docker Compose (für die Ausführung unserer Infrastruktur)
• Grundlegendes SQL

Sie benötigen Folgendes installiert:

• Node.js (v18 oder höher)
• Docker

Die Write-Seite mit PostgreSQL und Kafka erstellen

Unsere Write-Seite oder „Command"-Seite wird aus einer einfachen API zum Protokollieren neuer Workouts und einer PostgreSQL-Datenbank zu deren Speicherung bestehen.

Was wir tun

Wir definieren ein einfaches, normalisiertes Schema in PostgreSQL für unsere workouts-Tabelle und erstellen einen API-Endpunkt zum Einfügen neuer Datensätze.

Implementierung

Zunächst erstellen wir eine docker-compose.yml-Datei zum Ausführen unserer Infrastruktur: PostgreSQL, Kafka und ClickHouse.

# docker-compose.yml
version: '3.8'
services:
  postgres:
    image: postgres:15
    restart: always
    environment:
      - POSTGRES_USER=user
      - POSTGRES_PASSWORD=password
      - POSTGRES_DB=wellness_write_db
    ports:
      - '5432:5432'
    volumes:
      - postgres_data:/var/lib/postgresql/data

  zookeeper:
    image: confluentinc/cp-zookeeper:7.3.0
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181

  kafka:
    image: confluentinc/cp-kafka:7.3.0
    depends_on: [zookeeper]
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,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

  clickhouse:
    image: clickhouse/clickhouse-server:latest
    ports:
      - "8123:8123"
      - "9000:9000"
    volumes:
      - clickhouse_data:/var/lib/clickhouse

volumes:
  postgres_data:
  clickhouse_data:

Führen Sie docker-compose up -d aus, um alle Dienste zu starten.

Jetzt richten wir eine einfache Node.js-Anwendung für unsere Command-API ein.

mkdir wellness-app
cd wellness-app
npm init -y
npm install express pg kafkajs

Hier ist der Code für unseren „Write"-Dienst.

// command-service/index.js
const express = require('express');
const { Pool } = require('pg');
const { Kafka } = require('kafkajs');

const app = express();
app.use(express.json());

// PostgreSQL setup
const pool = new Pool({
  user: 'user',
  host: 'localhost',
  database: 'wellness_write_db',
  password: 'password',
  port: 5432,
});

// Kafka setup
const kafka = new Kafka({
  clientId: 'wellness-app',
  brokers: ['localhost:9092']
});
const producer = kafka.producer();

// Tabelle erstellen, falls nicht vorhanden
const initDb = async () => {
  await pool.query(`
    CREATE TABLE IF NOT EXISTS workouts (
      id SERIAL PRIMARY KEY,
      user_id VARCHAR(50) NOT NULL,
      type VARCHAR(50) NOT NULL,
      duration_seconds INT NOT NULL,
      distance_km FLOAT,
      logged_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP
    );
  `);
};

app.post('/log-workout', async (req, res) => {
  try {
    const { userId, type, durationSeconds, distanceKm } = req.body;
    // 1. In die transaktionale Datenbank schreiben
    const result = await pool.query(
      'INSERT INTO workouts(user_id, type, duration_seconds, distance_km) VALUES($1, $2, $3, $4) RETURNING *',
      [userId, type, durationSeconds, distanceKm]
    );
    const newWorkout = result.rows;
    console.log('Workout in PostgreSQL protokolliert:', newWorkout);

    // 2. Event an Kafka senden
    await producer.send({
      topic: 'workout-logged',
      messages: [{ value: JSON.stringify(newWorkout) }],
    });
    console.log('Event an Kafka-Topic gesendet: workout-logged');

    res.status(201).json(newWorkout);
  } catch (error) {
    console.error('Workout-Protokollierung fehlgeschlagen:', error);
    res.status(500).send('Internal Server Error');
  }
});

const start = async () => {
  await initDb();
  await producer.connect();
  app.listen(3000, () => {
    console.log('Command-Dienst lauscht auf Port 3000');
  });
};

start();

Wie es funktioniert

1. API-Endpunkt: Wir haben einen einzelnen Endpunkt /log-workout, der einen neuen Workout-Datensatz akzeptiert.
2. Transaktionaler Schreibvorgang: Die Daten werden zuerst in unsere PostgreSQL-workouts-Tabelle geschrieben. Dies ist unsere „Quelle der Wahrheit". Die Operation ist atomar und gewährleistet die Datenintegrität.
3. Event-Veröffentlichung: Nach einem erfolgreichen Datenbankschreibvorgang veröffentlichen wir ein Event in einem Kafka-Topic namens workout-logged. Dieses Event enthält die vollständigen Details des neu erstellten Workouts. Dies ist ein entscheidender Schritt für die Entkopplung unserer Write- und Read-Seiten.

Die Read-Seite mit ClickHouse Materialisierten Sichten erstellen

Jetzt zum lustigen Teil: der „Read"-Seite. Unser Ziel ist es, ein Datenmodell in ClickHouse zu erstellen, das für die genauen Abfragen optimiert ist, die unser Analytik-Dashboard benötigen wird.

Was wir tun

Wir werden eine Tabelle in ClickHouse erstellen, die für schnelle analytische Abfragen konzipiert ist. ClickHouse ist eine spaltenorientierte Datenbank, die sie außergewöhnlich schnell für OLAP-Workloads (Online Analytical Processing) macht. Wir werden auch einen Mechanismus einrichten, um Daten aus unserem Kafka-Topic in diese ClickHouse-Tabelle aufzunehmen.

Implementierung

ClickHouse hat eine native Kafka-Engine, die direkt ein Kafka-Topic abonnieren kann. Dies macht die Integration unglaublich nahtlos.

Zunächst verbinden Sie sich mit der ClickHouse-Instanz. Sie können den Kommandozeilen-Client über Docker verwenden:

docker-compose exec -it clickhouse clickhouse-client

Führen Sie im ClickHouse-Client die folgenden SQL-Befehle aus:

-- 1. Zieltabelle für unsere denormalisierten Analytik-Daten erstellen.
-- Hier werden die Daten dauerhaft für schnelle Abfragen gespeichert.
CREATE TABLE IF NOT EXISTS wellness_analytics.workouts_view
(
    `id` UInt32,
    `user_id` String,
    `type` String,
    `duration_seconds` UInt32,
    `distance_km` Float32,
    `logged_at` DateTime,
    `day_of_week` String,
    `is_weekend` UInt8
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(logged_at)
ORDER BY (user_id, logged_at);

-- 2. Kafka-Engine-Tabelle erstellen.
-- Diese Tabelle fungiert als Live-Stream von unserem Kafka-Topic. Sie speichert selbst keine Daten.
CREATE TABLE IF NOT EXISTS wellness_analytics.workouts_queue
(
    `id` UInt32,
    `user_id` String,
    `type` String,
    `duration_seconds` UInt32,
    `distance_km` Float32,
    `logged_at` DateTime
)
ENGINE = Kafka()
SETTINGS
    kafka_broker_list = 'kafka:9092',
    kafka_topic_list = 'workout-logged',
    kafka_group_name = 'clickhouse_wellness_consumer',
    kafka_format = 'JSONEachRow';

-- 3. Materialisierte Sicht erstellen.
-- Dies ist die Magie, die den Kafka-Stream mit unserer Ziel-Analytik-Tabelle verbindet.
-- Sie hört automatisch auf neue Nachrichten in workouts_queue und fügt sie in workouts_view ein,
-- während sie Transformationen durchführt.
CREATE MATERIALIZED VIEW IF NOT EXISTS wellness_analytics.workouts_consumer TO wellness_analytics.workouts_view
AS SELECT
    id,
    user_id,
    type,
    duration_seconds,
    distance_km,
    logged_at,
    formatDateTime(logged_at, '%A') AS day_of_week,
    if(in(toDayOfWeek(logged_at),), 1, 0) AS is_weekend
FROM wellness_analytics.workouts_queue;

Wie es funktioniert

1. workouts_view-Tabelle: Dies ist unsere Haupt-Analytik-Tabelle. Wir haben die MergeTree-Engine verwendet, die ClickHouses Kraftwerk für Analytik ist. Beachten Sie, dass wir vorberechnete, denormalisierte Spalten wie day_of_week und is_weekend hinzugefügt haben. Dies vermeidet diese Berechnungen zur Abfragezeit und macht unsere Dashboards schneller.
2. workouts_queue-Tabelle: Diese Tabelle verwendet die Kafka-Engine, um sich mit unserem Kafka-Broker zu verbinden und das workout-logged-Topic zu abonnieren. Sie bietet im Wesentlichen eine Live-Streaming-Ansicht der Nachrichten im Topic.
3. workouts_consumer-Materialisierte Sicht: Dies ist die ETL-Pipeline (Extract, Transform, Load). Sie zieht automatisch neue Datensätze aus workouts_queue, wendet einfache Transformationen an (wie die Berechnung von day_of_week) und lädt sie in unsere workouts_view-Tabelle.

Analytik aus dem optimierten Read-Modell abfragen

Mit unseren Write- und Read-Seiten, die eingerichtet sind, testen wir den gesamten Ablauf.

Ein neues Workout posten

Verwenden Sie ein Tool wie curl oder Postman, um eine POST-Anfrage an unseren Command-Dienst zu senden:

curl -X POST http://localhost:3000/log-workout \
-H "Content-Type: application/json" \
-d '{
    "userId": "user123",
    "type": "run",
    "durationSeconds": 3600,
    "distanceKm": 10.5
}'

Sie sollten Protokolle in Ihrer Command-Dienst-Konsole sehen, die anzeigen, dass das Workout in PostgreSQL protokolliert und ein Event an Kafka gesendet wurde.

Daten in ClickHouse verifizieren

Jetzt fragen wir unser ClickHouse Read-Modell ab. Gehen Sie zurück zu Ihrem ClickHouse-Client:

SELECT * FROM wellness_analytics.workouts_view;

Innerhalb von Augenblicken sollten Sie den neuen Workout-Datensatz sehen, komplett mit den hinzugefügten day_of_week- und is_weekend-Feldern!

Komplexe analytische Abfragen ausführen

Hier glänzt unsere Architektur. Führen wir eine Abfrage aus, die auf unserer transaktionalen Datenbank langsam gewesen wäre. Zum Beispiel: „Wie hoch ist die Gesamtdistanz und durchschnittliche Dauer der Workouts pro Benutzer, gruppiert nach Workout-Typ?"

SELECT
    user_id,
    type,
    sum(distance_km) as total_distance,
    avg(duration_seconds) / 60 as avg_duration_minutes
FROM wellness_analytics.workouts_view
GROUP BY user_id, type;

Diese Abfrage wird in ClickHouse unglaublich schnell sein, selbst mit Millionen oder Milliarden von Zeilen, weil die Daten in einem spaltenorientierten Format gespeichert sind, das für solche Aggregation optimiert ist.

Leistungsüberlegungen

• Unabhängige Skalierung: Der größte Leistungsgewinn ist die Fähigkeit, Lese- und Schreibvorgänge unabhängig zu skalieren. Wenn Ihre Analytik-Dashboards beliebt werden, können Sie weitere ClickHouse-Knoten hinzufügen, ohne die Leistung Ihres Workout-Protokollierungsdienstes zu beeinträchtigen.
• ClickHouse-Optimierung: Die MergeTree-Engine in ClickHouse ist hochgradig abstimmbar. Die Partitionierung Ihrer Daten (wie wir es nach Monat getan haben) ist entscheidend für die Leistung, da sie ClickHouse ermöglicht, irrelevante Daten nicht zu lesen.
• Eventuelle Konsistenz: Es ist wichtig zu verstehen, dass diese Architektur eventuell konsistent ist. Es wird eine leichte Verzögerung (normalerweise Millisekunden bis Sekunden) zwischen dem Protokollieren eines Workouts und seinem Erscheinen im Analytik-Dashboard geben. Für die meisten Wellness-Analytik-Anwendungsfälle ist dies ein völlig akzeptabler Kompromiss für die massiven Leistungsgewinne.

Alternative Ansätze

• Apache Druid: Für Anwendungsfälle, die eine noch geringere Latenz bei der Echtzeit-Aufnahme und Dashboards erfordern, ist Apache Druid eine starke Alternative zu ClickHouse. Druids Architektur ist stärker verteilt und kann komplexer zu verwalten sein, aber es eignet sich hervorragend für Echtzeit-, ereignisgesteuerte Szenarien.
• Change Data Capture (CDC): Anstatt dass die Anwendung Events veröffentlicht, könnten Sie ein CDC-Tool wie Debezium verwenden, um Änderungen direkt aus dem Write-Ahead Log (WAL) von PostgreSQL nach Kafka zu streamen. Dies entkoppelt den Write-Dienst weiter vom Eventing-System.

Fazit

Wir haben erfolgreich ein skalierbares Analytik-Backend unter Verwendung des CQRS-Musters erstellt. Durch die Trennung unseres Write-Modells (PostgreSQL) von unserem Read-Modell (ClickHouse) und deren Verbindung mit einem Event-Bus (Kafka) haben wir ein System geschaffen, das:

• Leistungsstark: Optimierte Datenbanken sowohl für transaktionale Schreibvorgänge als auch für analytische Lesevorgänge.
• Skalierbar: Wir können die Read- und Write-Seiten unabhängig skalieren, um die Nachfrage zu erfüllen.
• Wartbar: Die Trennung der Bedenken macht das System einfacher zu verstehen, zu entwickeln und zu warten.

Leistungsauswirkung: CQRS-Implementierungen in Gesundheits-Analytik-Systemen zeigen 50-100x Verbesserung der Abfrageleistung für komplexe Aggregation. Organisationen berichten über 90% Reduzierung der Dashboard-Ladezeiten und die Fähigkeit, 10x mehr gleichzeitige Benutzer auf der gleichen Infrastruktur zu bewältigen. Die spaltenorientierte Speicherung von ClickHouse ermöglicht Antwortzeiten unter einer Sekunde auf Milliarden-Datensätze, die mit traditionellen SQL-Datenbanken Minuten dauern würden.

Dieses Muster ist unglaublich leistungsstark für jede Anwendung mit mismatched Read- und Write-Workloads, von E-Commerce-Plattformen bis zu IoT-Systemen und natürlich Wellness-Analytik.

Nächste Schritte

• Query-API erstellen: Erstellen Sie einen neuen Dienst, der Endpunkte zum Abfragen der ClickHouse-workouts_view bereitstellt und Daten an ein Frontend-Dashboard liefert.
• Erweiterte Analytik erkunden: Nutzen Sie ClickHouses leistungsstarken SQL-Dialekt und Funktionen, um komplexere Analytik zu erstellen, wie Zeitreihenanalyse oder Trichteranalyse.
• Fehlerbehandlung: Implementieren Sie eine Dead-Letter-Queue in Kafka, um Events zu behandeln, die vom ClickHouse-Consumer nicht verarbeitet werden können.

Ressourcen

• Offizielle CQRS-Dokumentation von Martin Fowler
• ClickHouse-Dokumentation für die Kafka-Engine
• Offizielle Apache Kafka Website
• Offizielle PostgreSQL Website
• Verwandte Artikel:
  • Echtzeit-Pipeline mit Kafka & Flink - Stream-Verarbeitung im großen Maßstab
  • Monolith zu Microservices mit DDD refaktorieren - Systemarchitekturmuster

Kurzübersicht

Häufig gestellte Fragen

F: Wie 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 Veränderungen des Lebensstils würden helfen?" und „Wann sollte ich nachuntersuchen?"

F: Gibt es Verä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 (150+ Minuten/Woche), качественнй Schlaf (7-9 Stunden), Stressbewältigung und Vermeidung von Tabak.

Haftungsausschluss

Die Algorithmen und Techniken, die in diesem Artikel vorgestellt werden, 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 Angehörige der Gesundheitsberufe für medizinischen Rat.