Metadata-Version: 2.4
Name: killuhub
Version: 0.1.0
Summary: Pluggable data ingestion framework — connectors, Spark/Flink processing, Iceberg storage
Author: Kalleby Ramos
License: MIT
Requires-Python: >=3.11
Description-Content-Type: text/markdown
Requires-Dist: pydantic>=2.0
Provides-Extra: postgres
Requires-Dist: psycopg2-binary>=2.9; extra == "postgres"
Provides-Extra: kafka
Requires-Dist: confluent-kafka>=2.3; extra == "kafka"
Provides-Extra: s3
Requires-Dist: boto3>=1.34; extra == "s3"
Requires-Dist: pyarrow>=15.0; extra == "s3"
Provides-Extra: rest-api
Requires-Dist: requests>=2.31; extra == "rest-api"
Provides-Extra: spark
Requires-Dist: pyspark>=3.5; extra == "spark"
Provides-Extra: flink
Requires-Dist: apache-flink>=1.18; extra == "flink"
Provides-Extra: iceberg
Requires-Dist: pyiceberg>=0.6; extra == "iceberg"
Provides-Extra: delta
Requires-Dist: delta-spark>=3.1; extra == "delta"
Provides-Extra: scheduler
Requires-Dist: apscheduler>=3.10; extra == "scheduler"
Provides-Extra: all
Requires-Dist: killuhub[iceberg,kafka,postgres,rest_api,s3,scheduler,spark]; extra == "all"
Provides-Extra: dev
Requires-Dist: pytest>=8.0; extra == "dev"
Requires-Dist: pytest-mock>=3.12; extra == "dev"
Requires-Dist: ruff>=0.4; extra == "dev"
Requires-Dist: mypy>=1.9; extra == "dev"

# KilluHub

A pluggable lakehouse ingestion framework — connect any source, process with Spark or Flink, land in Apache Iceberg following the medallion architecture (Bronze → Silver → Gold).

---

## Architecture

```
  ┌─────────────────────────────────────────────────────┐
  │                     Data Sources                     │
  │         PostgreSQL · MySQL · Kafka · REST API        │
  └─────────────────────┬───────────────────────────────┘
                        │
                        ▼
  ┌─────────────────────────────────────────────────────┐
  │                    Connectors                        │
  │   BaseConnector — connect() · extract() · close()   │
  │   Server-side cursors · Pagination · Manual commits  │
  └─────────────────────┬───────────────────────────────┘
                        │  yields dict[str, Any] records
                        ▼
  ┌─────────────────────────────────────────────────────┐
  │                Processing Engines                    │
  │         Spark 3.5 (batch + streaming)                │
  │         Flink (Table API, streaming-first)           │
  └─────────────────────┬───────────────────────────────┘
                        │  DataFrame (Spark or Flink)
                        ▼
  ┌─────────────────────────────────────────────────────┐
  │               Medallion Pipelines                    │
  │                                                      │
  │   Bronze ──────────────────────────────────────────  │
  │   Raw data + metadata stamps + data contracts        │
  │   _ingested_at · _source_name · _batch_id            │
  │                                                      │
  │   Silver ──────────────────────────────────────────  │
  │   Dedup · type cast · date dims · upsert             │
  │                                                      │
  │   Gold (roadmap) ──────────────────────────────────  │
  │   Aggregations · business metrics · serving layer    │
  └─────────────────────┬───────────────────────────────┘
                        │
                        ▼
  ┌─────────────────────────────────────────────────────┐
  │                   Storage Layer                      │
  │     Apache Iceberg ✅ · Delta Lake · Apache Hudi     │
  │     append · overwrite · merge (MERGE INTO)          │
  │     S3 · HDFS · local filesystem                     │
  └─────────────────────────────────────────────────────┘
```

---

## Key capabilities

| Capability | Description |
|---|---|
| **Connector abstraction** | Any source (SQL, Kafka, REST) implements the same three-method interface. Swap sources without touching pipeline logic. |
| **Watermark-based ingestion** | Automatic incremental extraction — only records newer than the last saved watermark are fetched. No manual state tracking. |
| **Data contracts** | Schema enforcement at ingestion time: type validation, null checks, min/max bounds. Pipelines fail fast on bad data. |
| **Medallion pipelines** | Bronze stamps raw metadata; Silver deduplicates, casts types, and upserts. One `type: chain` config runs both in sequence. |
| **Multi-engine processing** | The same connector and config works with Spark (batch + structured streaming) and Flink (streaming-first). Switch with one string. |
| **Multi-storage lakehouse** | Iceberg (primary), Delta Lake, and Hudi are drop-in alternatives. Swap with `storage_writer_name`. |
| **Config-driven pipelines** | Everything — connector, engine, writer, watermark, contract — lives in one YAML file. No Python required for standard use cases. |
| **Multi-platform deployment** | One Helm chart targets EKS (Spark Operator), Databricks (Asset Bundles), and AWS EMR. Platform-specific details are injected by the chart, not the config. |

---

## Design principles

- **Engine agnostic** — Connectors yield plain Python dicts. They have no knowledge of Spark or Flink. Engines receive dicts and build their own DataFrames.
- **Storage agnostic** — Writers implement a single `write(df, table, mode)` interface. Iceberg, Delta, and Hudi are plug-in implementations.
- **Config-driven** — A complete pipeline — source, engine, transform, target, contract, schedule — is expressed as a YAML file. Pipeline logic lives in the framework, not in user code.
- **Medallion architecture** — Bronze is the canonical raw layer (immutable, metadata-stamped). Silver is always derived from Bronze, never from the source directly. This makes reprocessing safe and auditable.
- **Stateless pipelines** — Each `Pipeline.run()` creates fresh instances of connector, engine, and writer. No shared state between runs. Scheduled jobs can run in parallel safely.

---

## Data contracts

Every Bronze and Silver pipeline can declare a **data contract** — a schema specification that is validated before data is written. Violations either fail the pipeline or log a warning, depending on `on_violation`.

```yaml
contract:
  on_violation: fail        # fail | warn
  min_row_count: 1          # fail if fewer rows than this were ingested

  columns:
    - name: order_id
      type: long
      nullable: false       # null check enforced

    - name: amount
      type: double
      nullable: false
      min_value: 0          # value range enforced

    - name: created_at
      type: timestamp
      nullable: false

    - name: status
      type: string
      nullable: false
      allowed_values: [pending, confirmed, shipped, cancelled]
```

### What is validated

| Rule | YAML key | Description |
|---|---|---|
| Type check | `type` | Column must match the declared Spark/Iceberg type |
| Null check | `nullable: false` | No nulls allowed in this column |
| Min value | `min_value` | Rejects rows below this numeric threshold |
| Max value | `max_value` | Rejects rows above this numeric threshold |
| Allowed values | `allowed_values` | Enum constraint — only these values are valid |
| Row count | `min_row_count` | Minimum rows the batch must contain |

Contracts make data quality a first-class concern — not an afterthought in a downstream dbt model.

---

## Project structure

```
killuhub/
├── core/
│   ├── connector_interface.py   # BaseConnector ABC
│   ├── engine_interface.py      # BaseEngine ABC
│   ├── storage_interface.py     # BaseStorageWriter ABC
│   ├── config.py                # ConnectorConfig, PipelineConfig
│   ├── registry.py              # Registry + default_registry singleton
│   ├── batch.py                 # BatchConfig, BatchMode, StreamingConfig
│   ├── contract.py              # ContractSpec, ContractValidator
│   └── exceptions.py            # KilluHubError hierarchy
│
├── connectors/
│   ├── postgres/connector.py    # Server-side cursor, incremental watermark
│   ├── mysql/connector.py       # Streaming cursor, fetchmany, optional TLS
│   ├── kafka/connector.py       # confluent-kafka, batch + stream modes
│   └── rest_api/connector.py    # Page / cursor / offset pagination
│
├── processing/
│   ├── spark_engine.py          # PySpark 3.5 + Iceberg catalog wiring
│   └── flink_engine.py          # PyFlink Table API + Iceberg catalog
│
├── storage/
│   ├── iceberg/
│   │   ├── writer.py            # append / overwrite / merge via MERGE INTO
│   │   └── schema_manager.py    # schema evolution, time travel, compaction
│   ├── delta/writer.py          # Delta Lake (drop-in alternative)
│   └── hudi/writer.py           # Apache Hudi (drop-in alternative)
│
├── ingestion/
│   ├── pipeline.py              # Low-level connector → engine → writer loop
│   └── scheduler.py             # APScheduler cron + interval jobs
│
└── layers/
    ├── bronze/pipeline.py       # BronzePipeline — metadata stamping + contract
    ├── silver/pipeline.py       # SilverPipeline — dedup, cast, partition, upsert
    ├── streaming/pipeline.py    # StreamingBronzePipeline — Spark Structured Streaming
    └── gold/                    # (roadmap)

config/
├── bronze_postgres.yaml         # Batch bronze from Postgres
├── bronze_kafka.yaml            # Streaming bronze from Kafka
├── silver_orders.yaml           # Silver from bronze
├── chain_orders.yaml            # Bronze + Silver in one file (Postgres)
└── chain_api_orders.yaml        # Bronze + Silver in one file (REST API)

helm/killuhub/                   # Helm chart — deploys to EKS / Databricks / EMR
├── Chart.yaml
├── values.yaml                  # Single user-facing interface
└── templates/
    ├── configmap.yaml           # Renders pipeline YAML — used on all platforms
    ├── spark-application.yaml   # Spark Operator CRD (EKS only)
    ├── rbac.yaml                # ServiceAccount + IRSA (EKS only)
    ├── databricks-job.yaml      # Databricks Asset Bundle (Databricks only)
    └── emr-step.yaml            # EMR step + cluster JSON (EMR only)

docs/
├── core/core.md
├── connectors/connectors.md
├── processing/processing.md
├── storage/storage.md
├── ingestion/ingestion.md
├── layers/layers.md
├── helm/helm.md
└── usage/usage.md              # End-to-end how-to for every scenario
```

---

## Quick start

### Install

```bash
# Minimum install
pip install -e .

# With extras for your use case
pip install -e ".[postgres,spark,iceberg]"
pip install -e ".[kafka,spark,iceberg]"
pip install -e ".[mysql,spark,iceberg]"
pip install -e ".[all]"
```

### Bronze + Silver in one config (recommended)

The standard pattern is a **chain** config: one file runs bronze first, then silver automatically.
`silver.bronze_table` is auto-injected from the bronze stage — you never repeat the table name.

```yaml
# config/chain_orders.yaml
type: chain

engine:
  name: spark
  warehouse: ${WAREHOUSE:-/tmp/killuhub-warehouse}
  catalog_name: ${CATALOG:-local}
  catalog_type: hadoop

stages:
  - name: bronze-orders
    type: bronze
    mode: batch
    batch:
      strategy: incremental
      watermark_column: updated_at
      initial_watermark: "2024-01-01T00:00:00"
    connector:
      name: postgres
      config:
        host: ${PG_HOST:-localhost}
        database: shop
        user: postgres
        password: ${PG_PASSWORD}
        query: "SELECT * FROM orders"
    bronze:
      table: local.bronze.orders
      source_name: postgres.shop.orders
      partition_by: [_ingestion_date]

  - name: silver-orders
    type: silver
    mode: batch
    batch:
      strategy: incremental
      watermark_column: _ingested_at
      initial_watermark: "1970-01-01T00:00:00"
    silver:
      # bronze_table is auto-injected from the bronze stage above
      silver_table: local.silver.orders
      key_columns: [order_id]
      date_columns: [created_at, updated_at]
      type_map: { amount: double, quantity: int }
      null_check_columns: [order_id, customer_id]
      partition_by: [created_date]
      state_store: json
    contract:
      on_violation: fail
      columns:
        - { name: order_id, type: long, nullable: false }
        - { name: amount, type: double, nullable: false, min_value: 0 }
```

```bash
python main.py --config config/chain_orders.yaml

# Dry-run — validate config without executing
python main.py --config config/chain_orders.yaml --dry-run
```

### Standalone bronze (batch)

```yaml
# config/bronze_postgres.yaml
type: bronze
mode: batch
batch:
  strategy: incremental
  watermark_column: updated_at
  initial_watermark: "2024-01-01T00:00:00"
connector:
  name: postgres
  config:
    host: localhost
    database: shop
    user: postgres
    password: ${PG_PASSWORD}
    query: "SELECT * FROM orders"
engine:
  name: spark
  warehouse: /tmp/killuhub-warehouse
  catalog_name: local
  catalog_type: hadoop
bronze:
  table: local.bronze.orders
  source_name: postgres.shop.orders
  partition_by: [_ingestion_date]
```

```bash
python main.py --config config/bronze_postgres.yaml
```

### Streaming bronze (Kafka)

```yaml
type: bronze
mode: streaming
streaming:
  trigger: processingTime
  trigger_interval: "30 seconds"
  checkpoint_location: ${CHECKPOINT_PATH:-/tmp/checkpoints}
  output_mode: append
connector:
  name: kafka
  stream_format: kafka
  stream_options:
    kafka.bootstrap.servers: ${KAFKA_BROKERS:-localhost:9092}
    subscribe: ${KAFKA_TOPIC:-orders}
    startingOffsets: latest
engine:
  name: spark
  warehouse: /tmp/killuhub-warehouse
  catalog_name: local
  catalog_type: hadoop
bronze:
  table: local.bronze.orders
  source_name: kafka.orders
  partition_by: [_ingestion_date]
```

### Custom connector

```python
from killuhub.core import BaseConnector, ConnectorConfig, default_registry

class MongoConnector(BaseConnector):
    def connect(self): ...
    def extract(self): yield from my_collection.find()
    def close(self): ...

default_registry.register_connector("mongo", MongoConnector)

# Now usable in any config: connector.name: mongo
```

---

## Pipeline types

| Type | Description |
|------|-------------|
| `bronze` | Ingest raw data from a source, stamp metadata columns, write to Iceberg. Supports `batch` and `streaming` mode. |
| `silver` | Read from Bronze Iceberg table, deduplicate, cast types, add date dimensions, write to Silver Iceberg. Always `batch`. |
| `chain` | Run multiple stages in order (typically bronze → silver) from a single config file. |

### Batch strategies

Both `bronze` and `silver` support:

| `batch.strategy` | Description |
|-----------------|-------------|
| `incremental` | Read only records newer than the last saved watermark. Efficient for daily/hourly runs. |
| `full` | Read everything from the source on every run. Use for small tables or full reprocessing. |

---

## Available connectors

| Connector  | Source type | Supports incremental | Mode |
|------------|-------------|---------------------|------|
| `postgres` | PostgreSQL  | Yes (watermark column) | batch |
| `mysql`    | MySQL       | Yes (watermark column) | batch |
| `kafka`    | Kafka topic | Yes (offset tracking) | batch + streaming |
| `rest_api` | HTTP/REST   | Yes (watermark column) | batch |

> S3 is a **destination** in KilluHub (Iceberg tables stored on S3), not a source connector.
> Data in S3 files is read directly by Spark via the engine's `warehouse` path.

---

## Storage format comparison

| Feature                  | Iceberg ✅ (primary) | Delta Lake | Hudi       |
|--------------------------|----------------------|------------|------------|
| Multi-engine             | Spark, Flink, Trino, Presto, Snowflake | Spark, Databricks-first | Spark, Flink |
| ACID transactions        | Yes (v2)             | Yes        | Yes        |
| Schema evolution         | Yes                  | Yes        | Yes        |
| Time travel              | Yes (snapshots)      | Yes (versions) | Yes (timeline) |
| Hidden partitioning      | Yes                  | No         | No         |
| Open format              | Yes (no vendor lock) | Partially  | Yes        |
| Databricks compatible    | Unity Catalog v2     | Native     | Yes        |
| Upsert strategy          | MERGE INTO SQL       | MERGE INTO | Copy-on-Write / Merge-on-Read |

**Iceberg is the primary choice** because it was designed for multi-engine access. Write with Spark today, query with Trino tomorrow, move to Databricks later — Iceberg handles it without migration.

---

## Bronze metadata columns

Every row written by `BronzePipeline` gets these columns stamped automatically:

| Column | Type | Description |
|--------|------|-------------|
| `_ingested_at` | TIMESTAMP | When this batch was written |
| `_source_name` | STRING | Human label from `bronze.source_name` |
| `_batch_id` | STRING | UUID for this run (idempotency key) |
| `_batch_mode` | STRING | `"full"` or `"incremental"` |
| `_ingestion_date` | DATE | Date partition column |

---

## Deployment

KilluHub runs anywhere via the Helm chart. Set `platform` and fill the matching section:

```bash
# EKS (Spark Operator)
helm install killuhub-orders ./helm/killuhub \
  -f helm/killuhub/values.yaml \
  --set platform=eks \
  --set pipeline.bronze.table=prod.bronze.orders

# Databricks
helm install killuhub-orders ./helm/killuhub \
  -f helm/killuhub/values.yaml \
  --set platform=databricks \
  --set databricks.warehouse=s3://my-lake/warehouse

# AWS EMR
helm install killuhub-orders ./helm/killuhub \
  -f helm/killuhub/values.yaml \
  --set platform=emr \
  --set emr.configBucket=s3://my-bucket/killuhub
```

The chart injects the correct `engine:` block (catalog type, warehouse path, Unity Catalog name, etc.) into the rendered `pipeline.yaml` ConfigMap. Your pipeline config never contains platform-specific details.

---

## Roadmap

- [ ] Gold pipeline — aggregations, business metrics, serving layer
- [ ] CDC connectors — Debezium (Postgres/MySQL), DynamoDB Streams
- [ ] Schema registry integration — Avro/Protobuf deserialization from Confluent Schema Registry
- [ ] Iceberg compaction service — scheduled file compaction and snapshot expiry
- [ ] Data quality checks — row-level anomaly detection, freshness SLAs
- [ ] OpenTelemetry tracing — spans for each pipeline stage, exportable to Datadog/Grafana
- [ ] Great Expectations integration — contract DSL backed by GE expectation suites
- [ ] REST API for pipeline management — trigger, status, history via FastAPI

---

## Dependencies

| Extra       | Package              | Purpose                        |
|-------------|----------------------|--------------------------------|
| `postgres`  | psycopg2-binary      | PostgreSQL driver              |
| `mysql`     | mysql-connector-python | MySQL driver                 |
| `kafka`     | confluent-kafka      | Kafka consumer                 |
| `rest_api`  | requests             | HTTP client                    |
| `spark`     | pyspark              | Spark processing engine        |
| `flink`     | apache-flink         | Flink processing engine        |
| `iceberg`   | pyiceberg            | Iceberg Python client          |
| `delta`     | delta-spark          | Delta Lake writer              |
| `scheduler` | apscheduler          | Cron + interval job scheduling |

---

## Study guides

- [Core layer](docs/core/core.md) — Abstract interfaces, Registry pattern, Config system
- [Connectors](docs/connectors/connectors.md) — Postgres, MySQL, Kafka, REST API internals
- [Processing engines](docs/processing/processing.md) — Spark and Flink, when to use each
- [Storage layer](docs/storage/storage.md) — Iceberg deep dive, Delta and Hudi comparison
- [Ingestion layer](docs/ingestion/ingestion.md) — Pipeline orchestration and Scheduler
- [Layers](docs/layers/layers.md) — Bronze and Silver pipeline internals
- [Helm chart](docs/helm/helm.md) — Multi-platform deployment
- [Usage guide](docs/usage/usage.md) — End-to-end how-to for every scenario
