telecom-cdr-pipeline-ai/README.md

Telecom Raw CDR Ingestion & Analytics Pipeline

A high-throughput, fault-tolerant streaming data pipeline designed to ingest, buffer, normalize, and store raw telecom Call Detail Records (CDRs) using pure Python (framework-less asyncio), Kafka (Redpanda), and ClickHouse.

Built for at-least-once delivery, backpressure protection, deterministic deduplication, and operational observability.

---

1. Quickstart

Prerequisites

• Docker & Docker Compose v2+
• Python 3.12+ (for running the local verification script)

1. Start the Pipeline

docker compose up --build

This single command brings up:

• Redpanda (Kafka-compatible message broker on port 9092)
• Topic Provisioner (topic-init: creates cdr.raw and cdr.raw.dlq with exact partitions)
• ClickHouse (Analytics database on port 8123 HTTP / 9000 Native with auto-initialized schema)
• Ingestion API (Port 8000)
• Processing Worker (Port 8001)

All dependent services declare health checks with depends_on: { condition: service_healthy } to eliminate startup race conditions.

2. Verify Pipeline Behaviors

In a separate terminal, run the automated traffic generator:

python scripts/generate_traffic.py

Expected verification summary:

=== Pipeline Verification Summary ===
Records sent:               1570
Records in ClickHouse:      1000    ✅
Records in DLQ:             50      ✅
API 400 errors:             10      ✅
Duplicate raw_ids resolved: 20      ✅
=====================================

---

2. Architecture & Data Flow

flowchart LR
    NE([Network Element]) -->|POST /api/v1/cdr/raw| API[Ingestion API :8000]
    API -->|Bounded Async Queue| BP[Backpressure Gate]
    BP -->|Key: raw_id| Kafka[(Redpanda: cdr.raw)]

    Kafka -->|Consume| Worker[Processing Worker :8001]

    subgraph WorkerLogic [Worker Pipeline]
        direction TB
        V[Validate & Clean] --> N[Normalize Schema]
        N --> B[Micro-Batch Buffer]
        B --> OC[Commit Offsets after ClickHouse ACK]
    end
    Worker --- WorkerLogic

    WorkerLogic -->|Batch Insert| CH[(ClickHouse: cdr_records)]
    WorkerLogic -.->|Malformed / Errors| DLQ[(Redpanda: cdr.raw.dlq)]
    WorkerLogic --> PromW[Worker /metrics]

    API --> PromA[API /metrics]
    API -->|Query Internal Stats| Worker
    API --> Stats[Public GET /stats]

Component Roles

• Ingestion API: A framework-less HTTP/1.1 server built with standard library asyncio.start_server. Enforces bounded in-memory async queue backpressure (429 Too Many Requests), validates JSON syntax, and publishes to Kafka using raw_id as the partition key.
• Message Broker (Redpanda/Kafka):
  • cdr.raw: 3 partitions, LZ4 compression, 24-hour retention.
  • cdr.raw.dlq: 1 partition, 7-day retention.
• Processing Worker: Consumes cdr.raw, normalizes subscriber telephone numbers to E.164 (phonenumbers), maps radio access technology (RAT), derives deterministic UUIDv5 cdr_ids, micro-batches into ClickHouse, and manages Kafka offset commits.
• ClickHouse: Columnar storage using ReplacingMergeTree(processed_at) partitioned by month and ordered by subscriber metadata and timestamp.

---

3. Micro-Batching Strategy

Inserting row-by-row into ClickHouse causes MergeTree merge storms and performance degradation. The processing worker implements micro-batching:

Flush Conditions (Whichever Occurs First)

1. Count Threshold: Accumulated in-memory buffer reaches BATCH_SIZE (default: 1,000 records).
2. Time Threshold: Elapsed time since last flush reaches BATCH_FLUSH_INTERVAL_SECONDS (default: 2 seconds).

Graceful Shutdown (SIGTERM / SIGINT)

On container stop or deployment:

1. Worker stops polling new messages from Kafka.
2. Any remaining records buffered in memory are flushed to ClickHouse.
3. Kafka offsets for the flushed batch are committed.
4. Worker exits cleanly with exit code 0. Zero records buffered in memory are lost on shutdown.

---

4. Offset Management & At-Least-Once Delivery

Kafka consumer offsets are committed only after ClickHouse acknowledges the batch insert.

Failure Scenario Handling

Failure Scenario	Resolution Mechanism
ClickHouse Write Timeout	Worker retries the insert with exponential backoff (CH_RETRY_MAX_ATTEMPTS=3, initial delay 500ms, factor 2.0). If all retries fail, the entire batch is wrapped in DLQ envelopes, published to cdr.raw.dlq, and offsets are committed to prevent pipeline deadlock.
Worker Crash After Write, Before Commit	Upon worker restart, the batch is re-consumed from Kafka and re-inserted into ClickHouse. Idempotency is guaranteed by ReplacingMergeTree(processed_at) and deterministic UUIDv5 cdr_ids. Duplicate rows sharing the ORDER BY key are merged.
Partial Batch Failure	All-or-nothing transactional semantics. ClickHouse batch inserts are atomic per block; consumer offsets are never partially committed.

---

5. Deduplication Strategy

To guarantee idempotency during network retries and crash replays:

1. Deterministic cdr_id Generation: Derived using UUIDv5 with the DNS namespace and raw_id:

   import uuid
   cdr_id = uuid.uuid5(uuid.NAMESPACE_DNS, raw_id)
   

   Identical raw_ids always generate the exact same cdr_id.
2. ReplacingMergeTree(processed_at) Engine:

   ORDER BY (service_type, radio_access_type, cell_id, session_start_time, msisdn)
   

   ClickHouse collapses duplicate rows sharing the compound order key, keeping only the most recent processed_at timestamp. Queries can also use FINAL for immediate deduplicated views.

---

6. Observability & Monitoring

Prometheus Metrics

• Ingestion API: http://localhost:8000/metrics
  • cdr_api_received_total: Total CDRs received at the API gate.
• Processing Worker: http://localhost:8001/metrics
  • cdr_processed_total{service_type="...", radio_access_type="..."}
  • cdr_dlq_total{reason="..."}
  • cdr_kafka_to_clickhouse_latency_seconds (Histogram)
  • cdr_batch_flush_duration_seconds (Histogram)
  • cdr_batch_size_at_flush (Histogram)
  • cdr_in_flight_buffer_size (Gauge)
  • cdr_kafka_consumer_lag{partition="..."} (Gauge)

Unified /stats Endpoint

Exposed publicly by the Ingestion API at http://localhost:8000/stats:

{
  "pipeline": {
    "total_received": 1570,
    "total_processed": 1500,
    "total_dlq": 50,
    "in_flight_buffer": 0,
    "last_flush_at": "2026-09-19T10:00:00Z"
  },
  "breakdown": {
    "by_service_type": { "4G_DATA": 1000, "5G_DATA": 500 },
    "by_radio_access_type": { "LTE_4G": 1000, "5G_NR": 500 },
    "dlq_by_reason": {
      "missing_required_field": 20,
      "invalid_rat_value": 10,
      "invalid_epoch": 10,
      "normalization_error": 10
    }
  }
}

Analytical Queries (ClickHouse)

Connect via HTTP http://localhost:8123 or CLI:

-- 1. Total count by service type
SELECT service_type, count() AS total
FROM telecom.cdr_records
GROUP BY service_type;

-- 2. Average session duration by Radio Access Type
SELECT radio_access_type, avg(duration_seconds) AS avg_duration
FROM telecom.cdr_records
GROUP BY radio_access_type;

-- 3. Top cells by total traffic volume
SELECT cell_id, sum(total_bytes) AS total_traffic
FROM telecom.cdr_records
GROUP BY cell_id
ORDER BY total_traffic DESC
LIMIT 10;

---

7. Configuration Reference

All application behavior is driven by environment variables:

Variable	Default	Component	Description
API_HOST	0.0.0.0	API	HTTP host interface to bind
API_PORT	8000	API	Ingestion API port
API_QUEUE_MAX_SIZE	10000	API	Bounded queue capacity (returns 429 when full)
WORKER_STATS_URL	http://worker:8001/internal/stats	API	Internal worker stats endpoint for aggregation
WORKER_INSTANCE	worker-1	Worker	Worker instance identifier for DLQ tracing
WORKER_STATS_HOST	0.0.0.0	Worker	Worker internal HTTP host
WORKER_STATS_PORT	8001	Worker	Worker internal HTTP port
KAFKA_BOOTSTRAP_SERVERS	localhost:9092	All	Kafka / Redpanda broker address
KAFKA_RAW_TOPIC	cdr.raw	All	Raw CDR topic name (3 partitions)
KAFKA_DLQ_TOPIC	cdr.raw.dlq	Worker	Dead Letter Queue topic name
KAFKA_CONSUMER_GROUP	cdr-processing-group	Worker	Kafka consumer group ID
CLICKHOUSE_HOST	localhost	Worker	ClickHouse host address
CLICKHOUSE_PORT	8123	Worker	ClickHouse HTTP port
CLICKHOUSE_DATABASE	telecom	Worker	Target database name
CLICKHOUSE_TABLE	cdr_records	Worker	Target table name
BATCH_SIZE	1000	Worker	Number of records before triggering a micro-batch flush
BATCH_FLUSH_INTERVAL_SECONDS	2.0	Worker	Maximum seconds before triggering a micro-batch flush
CH_RETRY_MAX_ATTEMPTS	3	Worker	Maximum ClickHouse insert retry attempts
CH_RETRY_INITIAL_DELAY_MS	500	Worker	Initial retry backoff delay in milliseconds
CH_RETRY_BACKOFF_FACTOR	2.0	Worker	Exponential retry multiplier
LOG_LEVEL	INFO	All	Logging level (DEBUG, INFO, WARN, ERROR)