streaming-project

A hands-on learning project for building a streaming data pipeline end to end — Kafka, Flink, distributed storage/search (OpenSearch), and observability (Grafana) — while actually planning the architecture and choosing components, rather than having them handed over working. Kubernetes, Drone (CI/CD), and Terraform (IaC) are opportunistic additions if a natural chance to use them comes up.

See .specs/RFC.md for the full motivation and learning goals, and .specs/STATE.md for project-wide decisions and the current handoff state.

Architecture

Today: ingestion/ polls a source's public events API (GitHub wired, GitLab config-only), formats payloads into a RawEvent envelope (Pydantic, shared/models.py), and publishes them to a single Kafka topic (events-raw, shared across sources — the source field is what tells events apart downstream). flink/normalization/ consumes events-raw, applies a per-source declarative contract (YAML, no source-specific Python), and publishes NormalizedEvents, keyed by partition_key, to events-normalized. flink/analytics/ runs a Flink SQL job (Table API) over events-normalized, windowing counts into events-analytics. Local infra (3 controllers + 3 brokers + Kafka UI on localhost:8080, Flink job/task managers per stage) runs via infra/docker/docker-compose.yml.

Both ingestion sources and the normalization contract for each source are declared as data, not Python — see interface/ below and .specs/PLATFORM.md for why.

Planned, not yet built: OpenSearch as the metrics/search store, Grafana dashboards on top. See .specs/features/ for each stage's exact scope — streaming-ingestion (ingestion), flink-normalization (normalization), flink-aggregation (analytics), interface-layout (the interface/ contract layout itself).

GitHub/GitLab Events API --> ingestion (client -> formatter -> producer) --> Kafka (events-raw)
                                                                                  |
                                                          flink/normalization (contract-driven, per-source YAML)
                                                                                  |
                                                                       Kafka (events-normalized)
                                                                                  |
                                                              flink/analytics (Flink SQL, windowed counts)
                                                                                  |
                                                                       Kafka (events-analytics)
                                                                                  |
                                                                    [planned] OpenSearch --> Grafana

Repository layout

• ingestion/ — the ingestion service: ports.py (EventClientPort, EventProducerPort), adapters/ (RequestsClientAdapter, KafkaProducerAdapter), domain/ (ValidatingRawEventFormatter, YamlSourceConfigRepository, InMemoryDuplicateTracker), models.py (SourceConfig/EventModel), wired by use_case.py's IngestionPipeline and run via app.py
• flink/normalization/ — the normalization job: ports.py (TransformerPort), domain/ (EventNormalizer, NormalizationRulesEventEvaluator, YamlContractRepository), models.py (FieldRule/NormalizationContract/NormalizedEvent), adapters/transformer.py (FlinkTransformerAdapter, the PyFlink FlatMapFunction), run via app.py
• flink/analytics/ — the aggregation job: reads a .sql file from interface/analytics/ and runs it as Flink SQL/Table API statements (no Python transform logic — the SQL file is the contract, per AD-009)
• flink/common/ — Kafka source/sink building shared by both Flink jobs: ports.py (EventSourcePort, EventSinkPort), adapters/ (KafkaSourceAdapter, KafkaSinkAdapter), factory.py (KafkaFactory)
• interface/ — declarative contracts, not Python: sources/<name>/ingestion.yml (endpoints, auth, id/type fields) and sources/<name>/normalization.yml (field-mapping rules per source) for ingestion/normalization, analytics/*.sql for the aggregation stage
• shared/ — cross-cutting utilities: models.py (RawEvent, the domain-neutral envelope), logger.py (per-module logging setup)
• tests/ — mirrors ingestion//flink//shared/ 1:1, one test file per source module
• infra/docker/ — local infra: docker-compose.yml (Kafka cluster, Kafka UI, ingestion/normalization/analytics containers, Flink job/task managers), docker-compose.dev.yml (lighter single-broker overlay), scripts/create-topics.sh
• .specs/ — the spec-driven workflow: RFC.md (why this project exists), STATE.md (decisions + handoff), PLATFORM.md (the contract-driven-authoring decision), features/[name]/ (spec, design, tasks per feature)

Requirements

• Python 3.12
• Docker (with Docker Compose) for the local Kafka + Flink stack

Setup

python -m venv .venv
source .venv/bin/activate
pip install -r requirements/dev.txt

Dependencies are split by target under requirements/: base.txt (shared runtime), ingestion.txt and flink.txt (per-service runtime, what each Docker image installs), and dev.txt (everything plus the make test / make neat tooling).

All make targets below assume .venv is activated — the Makefile calls bare pytest/python/autoflake/isort/black, which resolve to .venv/bin only while it's active.

Usage

Start the local stack (Kafka + Flink):

make kafka-up

Kafka UI is then available at localhost:8080. Pass MODE=dev for a lighter single-broker/single-controller footprint (docker-compose.dev.yml overlay):

make kafka-up MODE=dev

Run the ingestion service against GitHub:

make ingestion-default

Stop the stack:

make kafka-down

Development

make test    # full test suite with coverage
make neat    # format/clean: autoflake + isort + black
make clean   # remove __pycache__ / .pyc files

Conventions

• Abstract base classes + constructor injection for testability (EventClientPort/EventProducerPort in ingestion/ports.py, TransformerPort in flink/normalization/ports.py, EventSourcePort/EventSinkPort in flink/common/ports.py) — new components follow this shape instead of hard-wiring dependencies.
• Pydantic models for anything crossing a boundary (RawEvent, NormalizedEvent, SourceConfig).
• One logger per class via getLogger(self.__class__.__name__) (see shared/logger.py) — never the bare root logging module.
• Tests are unittest.TestCase + unittest.mock, one file per source module, named test_<module>.py under a mirrored path in tests/.
• Conventional Commits (feat:, fix:, ...), one atomic commit per logical change.

Project workflow

This project follows a spec-driven workflow under .specs/: each feature has a spec.md (requirements and decisions, authoritative for that feature), plus design and tasks artifacts. Cross-feature decisions and the paused-work snapshot live in .specs/STATE.md.

Note for contributors: per project decision AD-008 in STATE.md, an AI agent working in this repository has its authorship over production/test code decided per task (own/paired/deliver), derived from the project author's tracked knowledge state — it is not a blanket "agent never writes code" rule. See .claude/rules/interaction-protocol.md for the exact scope.