steps-data-system

Containerized health-metric pipeline (synthetic generation -> streaming ingestion -> analytical query API) using Redpanda, FastAPI, and ClickHouse.

This implementation focuses on wearable step/movement telemetry (step, state) as the health signal.

Architecture

flowchart LR
  Seed["seed-init / steps-data-seed<br/>synthetic event generator"] -->|HTTP POST| API["POST /api/v1/step_counter/send<br/>publish step counter event"]
  API -->|JSON event keyed by user_id| RP[(Redpanda topic<br/>step-counter-events)]
  RP --> C["consumer service<br/>batch ingest worker"]
  C -->|malformed or exhausted retries| DLQ[(Redpanda topic<br/>step-counter-events-dlq)]
  C --> DB[(ClickHouse<br/>step_events)]
  Q["POST /api/v1/step_counter/query<br/>read-only SQL"] --> DB

Loading

Requirement-by-requirement mapping

1) Synthetic event stream

Implemented via steps-data-seed (also executed automatically by seed-init in Compose).

Event shape:

{
  "event_id": "123e4567-e89b-12d3-a456-426614174000",
  "activity_id": "123e4567-e89b-12d3-a456-426614174100",
  "user_id": "user-1",
  "timestamp": "2026-07-24T00:00:00Z",
  "step": 120,
  "state": "walking"
}

Generator realism model:

1. Step ranges are state-dependent (running > walking > background).
2. Continuous walking/running stretches receive a shared activity_id; background events leave it null.
3. Events are keyed by user_id in Kafka to preserve per-user ordering.

Chosen default seed profile (defensible baseline):

1. volume=10000
2. user_count=20
3. sampling_rate=2Hz for live mode

2) Event publishing (API endpoint)

Implemented via POST /api/v1/step_counter/send:

1. Client publishes a validated step counter event to the API.
2. Event is validated against the Avro schema registered in Schema Registry.
3. Event is published to the Redpanda topic step-counter-events (keyed by user_id for ordering).
4. API returns acknowledgement with event ID and topic information.

3) Ingestion

Implemented as streaming ingestion:

1. Events from both the API endpoint and seed generator are published to Redpanda.
2. A dedicated consumer process runs the aiokafka consumer.
3. Consumer reads in batches and writes to ClickHouse.

4) Storage shape/format/partitioning

Storage layer: ClickHouse with a write table and a deduplicated read view:

CREATE TABLE step_events_raw (
  event_id UUID,
  activity_id Nullable(UUID),
  user_id String,
  event_timestamp DateTime64(3, 'UTC'),
  step UInt32,
  state LowCardinality(String),
  ingested_at DateTime64(3, 'UTC')
) ENGINE = ReplacingMergeTree(ingested_at)
PARTITION BY toYYYYMM(event_timestamp)
ORDER BY (user_id, event_timestamp, event_id);

CREATE OR REPLACE VIEW step_events AS
SELECT event_id, activity_id, user_id, event_timestamp, step, state
FROM step_events_raw FINAL;

Format/shape choices:

1. Columnar storage in ClickHouse for higher-volume event analytics.
2. activity_id groups one continuous walking/running session without forcing every event to stand alone analytically.
3. ReplacingMergeTree plus a FINAL view keeps read-side idempotency under at-least-once delivery.
4. Monthly partitioning on event_timestamp keeps historical growth bounded, while the sort key favors user/time-based reads. For multi-year datasets, consider partitioning by quarter or year to keep partition sizes manageable and queries responsive.

Tradeoff: the FINAL view keeps query semantics simple, but it is more expensive than a fully pre-deduplicated model. For queries scanning <1M rows, the overhead is negligible; beyond that, consider pre-deduplication or migration to a fully-deduplicated table design.

5) Query engine

Query engine: ClickHouse SQL through HTTP endpoint:

• POST /api/v1/step_counter/query
• Read-only SQL (SELECT / WITH) with positional parameters.
• Guardrails: forbidden write/DDL keywords, single-statement check, row-limit cap (STEPS_DATA_SYSTEM_QUERY_MAX_ROWS).

6) docker compose up on clean machine

Implemented. docker compose up --build starts:

1. clickhouse
2. schema-init (ClickHouse schema)
3. redpanda
4. redpanda-init (topic bootstrap)
5. schema-registry-init
6. app (query API only)
7. consumer (stream ingestion worker)
8. seed-init (initial synthetic publish)
9. console (Redpanda UI)

7) README with architecture, decisions, tradeoffs

Covered in this document:

1. Architecture diagram
2. Key decisions
3. AI usage section
4. Closing “two more weeks” paragraph

Key decisions

Access Pattern: Dashboard with Daily, Weekly, Monthly Aggregations

The primary access pattern is a health dashboard similar to Apple Health, where users can view aggregated step counts at multiple time granularities (daily, weekly, monthly, yearly). This drives all major architectural choices:

Area	Decision	Benefit
Transport	Redpanda (Kafka API)	Real streaming semantics; decouples event production from ingestion, allowing high-frequency events (e.g., 2Hz step samples) to flow at natural rates without overwhelming the API; easy local container
Ingestion mode	Dedicated consumer service	API endpoint scales independently of ingest; consumer can batch writes and handle retries without blocking user-facing requests; enables at-least-once semantics with dead-letter fallback
Storage Layer	ClickHouse (columnar OLAP)	Perfect for time-series aggregations: Pre-aggregating steps by (user_id, date) is nearly free; month/week/day rollups execute in milliseconds on millions of events. Built-in engine (SummingMergeTree, AggregatingMergeTree) makes pre-computed aggregates automatic. Monthly partitioning by event_timestamp keeps partition sizes bounded while the sort key (user_id, event_timestamp) optimizes dashboard queries like "give me total steps for user X from 2026-07-01 to 2026-07-31". Compression (usually 10:1) keeps storage cost low for long-term step history.
Streaming Ingestion	Real-time batch ingest via consumer	Raw events are persisted as soon as they arrive; dashboard sees latest data within milliseconds of event publish. No polling delays. At-least-once semantics with idempotency (via event_id + ReplacingMergeTree) ensures no step loss and no duplicates. High-frequency samples (2Hz) arrive atomically per batch, making dashboards consistent across time zones and sessions.
Schema Model	Flat event table (event_id, activity_id, user_id, event_timestamp, step, state, ingested_at)	Simple, extensible. Each event is a fact: one discrete step sample at a point in time. activity_id groups continuous walking/running sessions (e.g., 5-minute walk) without forcing individual events to carry session context. Columns are few and highly selective, making range scans on (user_id, event_timestamp) extremely efficient. State (walking/running/background) enables filtering "active" vs "passive" step counts on the dashboard.

Why This Stack for Your Use Case

1. Redpanda + Streaming: Wearables emit high-frequency data (2Hz, 50Hz depending on device). A streaming backbone allows the system to ingest at full telemetry rate without the API becoming a bottleneck. At-least-once semantics guarantee no step loss.

2. ClickHouse: Dashboards typically query aggregates over time windows (e.g., "total steps for the week of July 21–27"). ClickHouse's columnar format and native aggregation functions make these queries sub-100ms even on years of data. A relational DB (Postgres, MySQL) would require denormalization or materialized views to achieve similar speed.

3. Monthly Partitioning: Step data grows by ~86.4k events per user per day (2Hz × 86400 seconds). Partitioning by month keeps partition sizes manageable (1–2 GB per user per month on 2Hz) and queries fast. Queries that span a month or less are usually single-partition lookups.

Bonus status

Bonus 1 — Query endpoint

Implemented (POST /api/v1/step_counter/query).

cURL examples:

Count events by state with step threshold:

curl -X POST http://localhost:8000/api/v1/step_counter/query \
  -H "Content-Type: application/json" \
  -d '{
    "query": "SELECT state, COUNT(*) AS total FROM step_events WHERE step >= ? GROUP BY state",
    "params": [4]
  }'

Get step distribution per user:

curl -X POST http://localhost:8000/api/v1/step_counter/query \
  -H "Content-Type: application/json" \
  -d '{
    "query": "SELECT user_id, state, AVG(step) AS avg_steps FROM step_events GROUP BY user_id, state ORDER BY user_id",
    "params": []
  }'

With authentication:

curl -X POST http://localhost:8000/api/v1/step_counter/query \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer YOUR_TOKEN_HERE" \
  -d '{
    "query": "SELECT COUNT(*) FROM step_events",
    "params": []
  }'

Request structure:

{
  "query": "SELECT state, COUNT(*) AS total FROM step_events WHERE step >= ? GROUP BY state",
  "params": [500]
}

Response example:

{
  "columns": ["state", "total"],
  "rows": [["running", 12], ["walking", 40]],
  "row_count": 2
}

Error shape:

{
  "error": {
    "code": "QUERY_POLICY_VIOLATION",
    "message": "only SELECT/WITH queries are allowed"
  }
}

Common error codes:

• QUERY_POLICY_VIOLATION: Read-only policy, DDL/write keyword, disallowed table, or disallowed data source detected
• QUERY_VALIDATION_FAILED: Malformed SQL or single-statement check failed
• QUERY_EXECUTION_TIMEOUT: Query exceeded STEPS_DATA_SYSTEM_QUERY_TIMEOUT_SECONDS
• QUERY_ROW_LIMIT_EXCEEDED: Result set exceeded STEPS_DATA_SYSTEM_QUERY_MAX_ROWS
• UNAUTHORIZED: Missing or invalid bearer token when auth is enabled

Table access restriction: Only the step_events table is accessible via the query endpoint. Queries referencing other tables (including step_events_raw, system tables, or external data sources) will be rejected.

Optional auth: if STEPS_DATA_SYSTEM_API_AUTH_TOKEN is set, requests require Authorization: Bearer <token>.

Bonus 2 — Natural-language layer

Not implemented in this version. The API is SQL-first.

Going deeper (areas emphasized)

1) Query safety and bounded execution

• Read-only SQL enforcement via policy checks.
• Single-statement restriction.
• Disallowed source protection (rejects system tables and external table functions).
• Maximum row-return guardrail to prevent accidental large responses.
• Query execution-time cap enforced at ClickHouse query settings.
• Structured error responses for policy, validation, and execution failures.

2) Schema registry and enforcement

• Schema evolution: Events are validated against Avro schemas registered in Confluent Schema Registry before ingestion.
• Producer side: Serialization via Avro ensures all published events conform to a versioned schema.
• Consumer side: The ingestion consumer deserializes events using the registered schema; any deviation (missing fields, type mismatch) is rejected and sent to the dead-letter topic.
• Backward/forward compatibility: Schema versions can be registered with compatibility mode (BACKWARD, FORWARD, FULL) to prevent breaking changes and enable safe evolution.
• Schema enforcement: Malformed or incompatible payloads are caught early and not persisted to ClickHouse, maintaining data integrity.

3) Retry mechanism and dead-letter queue (DLQ)

• Automatic retries with exponential backoff: When the consumer encounters transient errors (e.g., ClickHouse temporary unavailability, network glitches), it retries the batch with backoff to maximize delivery.
• Configurable retry policy: Max retries and backoff intervals are tunable via environment variables to balance resilience and latency.
• Dead-letter topic (step_events_dlq): After exhausting retries, failures—including malformed events, schema violations, and persistent ClickHouse errors—are published to the DLQ for:
  • Manual intervention: Operators can inspect and replay DLQ messages after fixing the underlying issue.
  • Monitoring and alerting: DLQ depth is tracked; high queue depth triggers alerts.
  • Offset safety: Only after successful DLQ publish (or explicit skip) does the consumer commit the original offset, ensuring no message loss.
• Event triage: The DLQ preserves the original payload + metadata (error reason, retry count, timestamp) to aid debugging and remediation.

Prerequisites

• Docker + Docker Compose plugin
• (Optional local dev/test) uv: https://docs.astral.sh/uv/

Run

docker compose up --build

Useful endpoints:

• API docs: http://localhost:8000/docs
• ReDoc: http://localhost:8000/redoc
• OpenAPI: http://localhost:8000/openapi.json
• ClickHouse HTTP: http://localhost:8123
• Redpanda Console: http://localhost:8080
• Redpanda Admin API: http://localhost:9644
• Schema Registry: http://localhost:8081

Local development

Common workflows:

make install
make check
make run
make consumer
make seed
make compose-up

Useful overrides:

• make seed SEED_MODE=live SEED_VOLUME=120 SEED_USER_COUNT=1
• make api-test API_TEST_ARGS='--query "SELECT COUNT(*) FROM step_events"'
• make compose-logs SERVICE=consumer

Run tests:

uv run pytest

Coverage:

uv run pytest --cov=steps_data_system --cov-report=term-missing

Generate sample data manually:

uv run steps-data-seed --mode init --volume 10000 --sampling-rate 2

Live simulation example:

uv run steps-data-seed --mode live --volume 120 --sampling-rate 2 --user-count 1

Test the local query API:

uv run steps-data-test-api
uv run steps-data-test-api --query "SELECT state, COUNT(*) AS total FROM step_events WHERE step >= ? GROUP BY state" --param 500

Tester knobs:

• --base-url: root URL for the running API (default http://localhost:8000)
• --query: read-only SQL query to execute (default smoke test counts all events)
• --param: repeatable positional query parameter, JSON-parsed for scalars when possible
• --token: optional bearer token; falls back to STEPS_DATA_SYSTEM_API_AUTH_TOKEN when set
• --timeout: request timeout in seconds (default 10)

Seeder knobs:

• --mode: init or live
• --sampling-rate: events/second in live mode
• --user-count: rotating synthetic users (default 20)
• --seed: deterministic random seed (default 42)
• --start-time: ISO-8601 start timestamp for init mode
• --bootstrap-servers, --topic: Redpanda overrides

Environment variables

Note: Docker Compose automatically sets these for containerized runs. Override them only for local development outside containers.

• STEPS_DATA_SYSTEM_CLICKHOUSE_HOST (default localhost)
• STEPS_DATA_SYSTEM_CLICKHOUSE_PORT (default 8123)
• STEPS_DATA_SYSTEM_CLICKHOUSE_DATABASE (default steps_data_system)
• STEPS_DATA_SYSTEM_CLICKHOUSE_USERNAME (default default)
• STEPS_DATA_SYSTEM_CLICKHOUSE_PASSWORD (default empty)
• STEPS_DATA_SYSTEM_REDPANDA_BOOTSTRAP_SERVERS (default localhost:9092)
• STEPS_DATA_SYSTEM_REDPANDA_TOPIC (default step-counter-events)
• STEPS_DATA_SYSTEM_REDPANDA_DEAD_LETTER_TOPIC (default step-counter-events-dlq)
• STEPS_DATA_SYSTEM_REDPANDA_CONSUMER_GROUP (default steps-data-system-step-counter)
• STEPS_DATA_SYSTEM_REDPANDA_AUTO_OFFSET_RESET (default earliest)
• STEPS_DATA_SYSTEM_REDPANDA_CONSUMER_BATCH_SIZE (default 1000)
• STEPS_DATA_SYSTEM_REDPANDA_CONSUMER_BATCH_WAIT_MS (default 100)
• STEPS_DATA_SYSTEM_REDPANDA_BATCH_MAX_RETRIES (default 3)
• STEPS_DATA_SYSTEM_REDPANDA_BATCH_RETRY_DELAY_MS (default 1000)
• STEPS_DATA_SYSTEM_QUERY_MAX_ROWS (default 10000)
• STEPS_DATA_SYSTEM_QUERY_TIMEOUT_SECONDS (default 10)
• STEPS_DATA_SYSTEM_API_AUTH_TOKEN (default unset; auth disabled when unset)

AI usage notes

I used AI tooling for:

1. Documenting architecture and tradeoffs in this README.
2. SQL query policy enforcement and error-shape design.
3. Looking for different tools to implement the ingestion and query engine.
4. Boilerplate code for the FastAPI query endpoint and the aiokafka consumer.
5. Unit tests

Where I overrode/corrected AI output:

1. Missing docstring and type hints in Python code.
2. Missing error handling in Python code.

What I would not trust AI alone to do here:

1. Reliability and failure-mode decisions

If you had two more weeks, what would you build next?

I would prioritize security hardening across the API, tables, and topic:

1. API Security: Implement row-level access control (RLAC) to limit user queries to their own user_id data without creating database-level users.

2. Table Security: Lock down table access via row-level policies enforced in the query layer (no new database user accounts required). Restrict materialized views to pre-computed safe aggregations. Separate read views from ingestion tables with immutable access control.

3. Topic Security: Encrypt topic data at rest and in transit. Implement producer/consumer ACLs in Redpanda to prevent unauthorized publishes to step-counter-events and prevent external consumers from reading the topic.