MapReduce on Kubernetes

INF-419 Principles of Distributed Systems — Class Project 2026
Team: Dimas Christos, Giannakakos Alexandros, Lamprakis Michalis
Repo: distributed-systems-inf419

A distributed MapReduce engine on Kubernetes. Users submit a mapper function, a reducer function, and a JSONL dataset via CLI. The system splits the input, runs map and reduce tasks as ephemeral Kubernetes Jobs, and produces final output in MinIO — recovering automatically from worker or manager failures.

---

Prerequisites

• Docker Engine
• Minikube
• kubectl
• Python 3.11+

---

Setup

make setup

Builds all Docker images inside Minikube's Docker context and applies all Kubernetes manifests in dependency order (storage → auth → manager → ui).

make purge   # tear everything down

---

Project Structure

├── services/
│   ├── ui/          # API gateway — the only externally exposed service
│   ├── manager/     # Orchestrator — job state machine, spawns K8s Jobs
│   └── worker/      # Ephemeral task runner — one pod per map/reduce task
├── cli/             # mr CLI client
├── shared/          # Pydantic schemas shared across services
├── k8s-manifests/   # Kubernetes manifests for all services and infrastructure
├── benchmarks/      # Word count and inverted index mapper/reducer pairs + datasets
├── tests/           # Unit and integration tests
└── scripts/         # test_fault_tolerance.sh, test_r6_recovery.sh

---

Services

Service	K8s Resource	Port	Role
UI Service	Deployment (2 replicas)	NodePort 30080	Auth, file upload, job routing
Manager	StatefulSet (2 replicas)	ClusterIP	Job orchestration, K8s Job creation
Worker	Job (ephemeral)	—	Executes one map or reduce task
PostgreSQL	StatefulSet + PVC	ClusterIP	Job and task state
MinIO	StatefulSet + PVC	ClusterIP	Input, intermediate, and output data
Keycloak	StatefulSet + PVC	NodePort 30081	JWT token issuance (OAuth2/OIDC)

All inter-service traffic uses ClusterIP DNS (e.g. manager-service:8000). Config via ConfigMaps, credentials via K8s Secrets — nothing hardcoded in images.

---

CLI Usage

mr login --username alice

mr jobs upload-data --file benchmarks/wordcount/data/input.jsonl
mr jobs upload-code --mapper benchmarks/wordcount/mapper.py \
                    --reducer benchmarks/wordcount/reducer.py

mr jobs submit --input-id <input_id> --code-id <code_id> --reducers 2
mr jobs status <job_id>
mr jobs results <job_id>
mr jobs abort <job_id>

mr admin create-user --username bob

---

How a Job Runs

1. CLI POSTs to UI Service → UI Service forwards to Manager.
2. Manager computes byte-range splits, inserts task rows into PostgreSQL, spawns one Kubernetes Job per map task.
3. Each map worker downloads mapper.py from MinIO at startup, processes its byte range, writes R partitioned objects to intermediate/, calls back the Manager with status: done.
4. When all map tasks finish, Manager spawns one Kubernetes Job per reduce partition.
5. Each reduce worker reads its partition from intermediate/, runs reducer.py, writes to outputs/, calls back the Manager.
6. Manager marks the job completed and deletes all worker Jobs.

---

Fault Tolerance

Worker failure is detected two ways:

• Heartbeat timeout: background loop runs every 30 s. Any task stuck in running for more than 120 s is retried.
• Startup recovery scan: on Manager restart, recover_jobs() checks every in-progress task against live K8s state. Tasks whose Job object is gone are retried immediately.

Recovery: partial map_outputs from the failed attempt are deleted, attempt increments, a new K8s Job is spawned. After MAX_TASK_ATTEMPTS (3) the job is marked failed.

Manager failure: Manager is a StatefulSet. On restart, all state is rebuilt from PostgreSQL — no in-memory state is lost.

---

Run Tests

Prerequisites

• minikube running
• System deployed via make setup from the src/ directory
• All tests run from src/scripts/
• CLI runs from src/cli/

Setup

cd src
make setup

Wait until all pods are Running:

kubectl get pods

Expected pods: keycloak-0, manager-0, manager-1, minio-0, postgres-0, ui-service-* (x2)

Test data

All test data lives in src/cli/test-data/ (see notes for files):

File	Purpose
/benchmark_2/corpus.jsonl	Full-size word count input
small_test.jsonl	Small input for fault tolerance tests
mapper.py	Normal word count mapper
reducer.py	Word count reducer
one_fail_mapper.py	Attempt-aware mapper — hangs on attempt 0, succeeds on attempt 1

Run everything

cd src/scripts
bash run_all_tests.sh

Run individually

bash test_happy_path.sh
bash test_worker_ft.sh all      # or 1|2|3|4
bash manager_tolerance_test.sh all   # or 1|2|3
bash test_postgres_ft.sh
bash test_committed_output.sh
bash test_abort.sh
bash test_concurrent.sh

Test descriptions

Script	What it tests
test_happy_path.sh	End-to-end job submission, execution, and result retrieval
test_worker_ft.sh 1	Worker killed mid-execution — heartbeat detects, retries, completes
test_worker_ft.sh 2	Worker always fails — job marked FAILED after max attempts
test_worker_ft.sh 3	No duplicate map outputs after retry
test_worker_ft.sh 4	Reduce worker killed — retries and completes
manager_tolerance_test.sh 1	One manager down — other handles new submissions
manager_tolerance_test.sh 2	Manager dies mid-job — recovers from DDS and completes
manager_tolerance_test.sh 3	Both managers down — each recovers only its own job
test_postgres_ft.sh	PostgreSQL pod killed — connection pool reconnects, job completes
test_committed_output.sh	Manager killed after map output committed — reused without duplication
test_abort.sh	Job abort cancels all running workers cleanly
test_concurrent.sh	Two concurrent jobs complete without interference

Notes

• test_worker_ft.sh 2 requires manually setting MAX_TASK_ATTEMPTS=2 in the manager and uploading an always-hang mapper — follow the prompt when running
• Heartbeat timeout is set to 45s (HEARTBEAT_TIMEOUT=45) — tests account for this wait
• to have your own corpus.jsonl version:

python3 -c "
import json, random, sys

words = ['the','and','is','in','it','of','to','a','that','was','he','she','they',
         'map','reduce','kubernetes','distributed','system','cluster','worker','job',
         'data','file','input','output','stream','process','node','task','parallel']

with open('corpus.jsonl', 'w') as f:
    for i in range(2000000):
        length = random.randint(5, 20)
        text = ' '.join(random.choices(words, k=length))
        f.write(json.dumps({'id': i, 'text': text}) + '\n')
        if i % 100000 == 0:
            print(f'{i} lines written...', file=sys.stderr)
"
ls -lh corpus.jsonl

---

Benchmarks

Two MapReduce programs in benchmarks/:

• Word Count — emits (word, 1) per token, sums by key.
• Inverted Index — emits (word, doc_id) per token, collects document lists per word.

---

Tech Stack

Component	Technology
Services	Python 3.12 + FastAPI
Data format	JSONL
Object storage	MinIO (S3-compatible)
Database	PostgreSQL
Auth	Keycloak (OAuth2 / OIDC)
Orchestration	Kubernetes / Minikube
Schema validation	Pydantic