killuhub 0.1.0

Pluggable data ingestion framework — connectors, Spark/Flink processing, Iceberg storage

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.

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

# 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.

# 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 }

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)

# 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]

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

Streaming bronze (Kafka)

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

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:

# 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 — Abstract interfaces, Registry pattern, Config system
• Connectors — Postgres, MySQL, Kafka, REST API internals
• Processing engines — Spark and Flink, when to use each
• Storage layer — Iceberg deep dive, Delta and Hudi comparison
• Ingestion layer — Pipeline orchestration and Scheduler
• Layers — Bronze and Silver pipeline internals
• Helm chart — Multi-platform deployment
• Usage guide — End-to-end how-to for every scenario