Data Engineering Jobs in Telecom: Role and Responsibilities

# CDR Aggregation — the core pipeline in telecom data engineering.
# Billion-scale data, minute-level latency, strict SLAs.

from pyspark.sql import SparkSession
from pyspark.sql.functions import (
    col, count, sum as _sum, to_date, window, from_unixtime,
)

spark = (
    SparkSession.builder
    .appName("cdr_aggregation")
    .getOrCreate()
)

# Read the raw CDRs from the data lake
# Partitioned by date for efficient scan
cdr = spark.read.parquet("s3://telecom-lake/cdr/date=2026-09-12/")

# Compute per-subscriber daily usage
usage = (
    cdr
    .filter(col("call_type").isin(["voice", "sms", "data"]))
    .groupBy("subscriber_id", to_date(from_unixtime("event_ts")).alias("date"))
    .agg(
        count("*").alias("event_count"),
        _sum("duration_seconds").alias("total_duration"),
        _sum("bytes_up").alias("bytes_up"),
        _sum("bytes_down").alias("bytes_down"),
    )
)

# Write to the warehouse in a partitioned table
usage.write.mode("overwrite").partitionBy("date").parquet(
    "s3://telecom-warehouse/fact_usage_daily/"
)

# What this does:
# - Reads billions of CDR rows from partitioned parquet
# - Aggregates per subscriber per day
# - Writes a summary table partitioned by date
#
# What a telecom data engineer has to think about:
# - Skew: some subscribers have 1000x the events of others
# - Late data: a CDR might arrive 2 hours after the event
# - Duplicates: retries can cause the same event twice
# - Idempotency: reruns must not double-count
#
# These are the interview topics. Anyone can write a groupBy.
# The job is making it correct, reliable, and cost-effective
# at telecom volume.

# Kafka consumer — network events at high throughput.
# Telecom streams are often 1M+ events/second.

from kafka import KafkaConsumer
import json
import time
import boto3

consumer = KafkaConsumer(
    "network-events",
    bootstrap_servers="kafka-broker:9092",
    group_id="network-events-processor",
    value_deserializer=lambda m: json.loads(m.decode("utf-8")),
    auto_offset_reset="earliest",
    enable_auto_commit=False,  # commit manually after processing
    max_poll_records=5000,
)

s3 = boto3.client("s3")
buffer = []
batch_size = 1000

def flush():
    if not buffer:
        return
    # Write as JSONL to S3, partitioned by hour
    ts = int(time.time())
    key = f"network-events/hour={ts // 3600}/batch-{ts}.jsonl"
    body = "\n".join(json.dumps(e) for e in buffer)
    s3.put_object(
        Bucket="telecom-lake",
        Key=key,
        Body=body.encode("utf-8"),
    )
    buffer.clear()

for msg in consumer:
    event = msg.value
    buffer.append(event)

    if len(buffer) >= batch_size:
        flush()
        consumer.commit()

# What this does:
# - Consumes events from Kafka at scale
# - Batches them into ~1000-event chunks
# - Writes to S3 as JSONL, partitioned by hour
# - Commits offsets only after a successful write

# What a telecom data engineer has to think about:
# - Backpressure: what happens if Kafka produces faster than
#   we can write to S3?
# - Exactly-once vs at-least-once semantics
# - Ordering guarantees within a partition
# - Failure recovery: if the process dies, does it resume
#   from where it left off without duplicating events?
#
# These are the questions in a telecom DE interview. They are
# not about writing a consumer — they are about making it
# reliable at scale.

# Airflow DAG — daily reconciliation between billing and CDRs.
# Every telecom operator runs something like this nightly.

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.apache.spark.operators.spark_submit import (
    SparkSubmitOperator,
)
from datetime import datetime, timedelta

def validate_counts(**context):
    """Check that today's billing and CDR counts are within tolerance."""
    ds = context["ds"]  # YYYY-MM-DD
    # In a real DAG, this queries the warehouse
    billing_count = 42_500_000   # example
    cdr_count = 42_512_000       # example

    diff_pct = abs(billing_count - cdr_count) / billing_count * 100
    if diff_pct > 0.5:
        raise ValueError(f"Reconciliation failed: {diff_pct:.2f}% drift")
    print(f"OK: {diff_pct:.2f}% drift")

with DAG(
    dag_id="telecom_daily_reconciliation",
    start_date=datetime(2026, 1, 1),
    schedule="0 3 * * *",  # every day at 3 AM
    catchup=False,
    default_args={
        "retries": 2,
        "retry_delay": timedelta(minutes=15),
    },
) as dag:

    aggregate_cdr = SparkSubmitOperator(
        task_id="aggregate_cdr",
        application="jobs/aggregate_cdr.py",
        conn_id="spark_default",
    )

    aggregate_billing = SparkSubmitOperator(
        task_id="aggregate_billing",
        application="jobs/aggregate_billing.py",
        conn_id="spark_default",
    )

    reconcile = PythonOperator(
        task_id="reconcile",
        python_callable=validate_counts,
    )

    aggregate_cdr >> aggregate_billing >> reconcile

# What this does:
# - Runs nightly at 3 AM
# - Aggregates CDRs and billing separately
# - Reconciles the counts, failing if drift is above 0.5%
# - Retries twice with a delay if anything fails
#
# What a telecom data engineer has to think about:
# - SLA: what time does the pipeline need to finish?
# - Cost: how many Spark clusters does this start, and for how long?
# - Alerting: who gets paged if reconciliation fails at 4 AM?
# - Backfill: how do we recover a week of data after an outage?
#
# The "pipelines" in telecom are this: reliable, repeatable,
# monitored batch and streaming jobs that keep the business
# running. That is the job.