mkpipe

mkpipe is a Spark-based, modular ETL framework. It provides a Python-first API for building data pipelines with pluggable extractors, loaders, and inline transformations.

Key Features

• PySpark engine — parallel, partitioned JDBC reads/writes for high-throughput data movement
• Modular plugins — pip install mkpipe-extractor-postgres to add a source, pip install mkpipe-loader-postgres to add a destination
• Single YAML config — define connections, pipelines, and tables in one file
• Tag-based execution — assign tags to tables and run only what you need across all pipelines
• Orchestrator-agnostic — use with Dagster, Airflow, cron, or any Python scheduler
• Inline transformations — reference a Python function (df → df) directly in YAML
• Dependency Injection — pass your own SparkSession (Glue, EMR, Dataproc) or let mkpipe create one
• Incremental & full replication — append-only incremental with mkpipe_id for idempotent deduplication

Quick Start

pip install mkpipe mkpipe-extractor-postgres mkpipe-loader-postgres

Create mkpipe_project.yaml:

version: 2
default_environment: prod

prod:
  settings:
    timezone: UTC
    log_dir: /path/to/logs
    backend:
      variant: sqlite

  connections:
    source_pg:
      variant: postgres
      host: localhost
      port: 5432
      database: source_db
      user: ${PG_USER}
      password: ${PG_PASSWORD}
      schema: public

    target_pg:
      variant: postgres
      host: localhost
      port: 5432
      database: dwh_db
      user: ${PG_USER}
      password: ${PG_PASSWORD}
      schema: staging

  pipelines:
    - name: my_pipeline
      source: source_pg
      destination: target_pg
      tables:
        - name: public.users
          target_name: stg_users
          tags: [api, user-domain]
          replication_method: incremental
          iterate_column: updated_at
          iterate_column_type: datetime
          write_strategy: upsert
          write_key: [id]

        - name: public.orders
          target_name: stg_orders
          tags: [api, order-domain]
          replication_method: full

Run it:

# Run all tables
mkpipe run

# Run only tables tagged "api"
mkpipe run --tags api

Python API

import mkpipe

# Run all pipelines
mkpipe.run(config="mkpipe_project.yaml")

# Run a specific pipeline
mkpipe.run(config="mkpipe_project.yaml", pipeline="my_pipeline")

# Run a specific table
mkpipe.run(config="mkpipe_project.yaml", table="stg_users")

# Run by tags — runs matching tables across ALL pipelines
mkpipe.run(config="mkpipe_project.yaml", tags=["api"])
mkpipe.run(config="mkpipe_project.yaml", tags=["api", "order-domain"])

# Combine filters: pipeline + tags
mkpipe.run(config="mkpipe_project.yaml", pipeline="my_pipeline", tags=["api"])

# Pass a custom SparkSession (e.g. AWS Glue, EMR, Dataproc)
mkpipe.run(config="mkpipe_project.yaml", spark=my_spark_session)

# Extract only — returns ExtractResult with a Spark DataFrame
result = mkpipe.extract(config="mkpipe_project.yaml", table="stg_users")
df = result.df

# Load only — pass your own DataFrame
mkpipe.load(config="mkpipe_project.yaml", table="stg_users", df=my_df)

CLI Reference

mkpipe run [OPTIONS]
mkpipe install-jars

Examples:

# Run everything
mkpipe run

# Specific config file
mkpipe run --config /path/to/config.yaml

# Single pipeline
mkpipe run -p my_pipeline

# Single table
mkpipe run -t stg_users

# By tags (OR logic: any matching tag)
mkpipe run --tags api
mkpipe run --tags api,ingestion

# Combine: pipeline + tags
mkpipe run -p my_pipeline --tags api

Tags

Tags let you group tables by business domain, team, priority, or any criteria. When you pass tags, mkpipe runs all matching tables across all pipelines (OR logic).

pipelines:
  - name: pg_to_pg
    source: source_pg
    destination: target_pg
    tables:
      - name: public.users
        target_name: stg_users
        tags: [api, user-domain, critical]

      - name: public.sessions
        target_name: stg_sessions
        tags: [api, user-domain]

  - name: mysql_to_pg
    source: source_mysql
    destination: target_pg
    tables:
      - name: orders
        target_name: stg_orders
        tags: [api, order-domain, critical]

# Runs stg_users + stg_sessions + stg_orders (all have "api")
mkpipe.run(config="config.yaml", tags=["api"])

# Runs stg_users + stg_orders (both have "critical")
mkpipe.run(config="config.yaml", tags=["critical"])

# Runs stg_users + stg_sessions (both have "user-domain")
mkpipe.run(config="config.yaml", tags=["user-domain"])

# OR logic: runs anything tagged "critical" OR "order-domain"
mkpipe.run(config="config.yaml", tags=["critical", "order-domain"])

YAML Configuration Reference

Top-level Structure

version: 2
default_environment: prod    # which environment block to use

prod:                        # environment name
  settings: ...
  connections: ...
  pipelines: ...

staging:                     # you can define multiple environments
  settings: ...
  connections: ...
  pipelines: ...

Settings

settings:
  timezone: UTC              # Spark session timezone (default: UTC)
  log_dir: ./logs            # Log file directory (optional, logs to console if not set)
  ingested_at_column: _ingested_at  # Column name for ingestion timestamp (default: _ingested_at)
  ingestion_id_column: mkpipe_id   # Column name for dedup hash ID (default: mkpipe_id)
  column_name_case: as_is          # Column name casing: lower, upper, as_is (default: as_is)
  if_exists: replace               # Target table behavior on full load: replace (default) or append

  spark:
    master: "local[*]"       # Spark master URL (default: local[*])
    driver_memory: "4g"      # default: auto-detected from system
    executor_memory: "4g"    # default: auto-detected from system
    extra_config:            # any additional Spark config
      spark.sql.shuffle.partitions: "200"
      spark.dynamicAllocation.enabled: "true"

  backend:
    variant: sqlite          # sqlite (default), postgres, duckdb, clickhouse
    host: localhost
    port: 5432
    database: mkpipe_db
    user: mkpipe
    password: ${BACKEND_PASSWORD}

Connections

connections:
  my_postgres:
    variant: postgres
    host: ${PG_HOST}
    port: 5432
    database: mydb
    user: ${PG_USER}
    password: ${PG_PASSWORD}
    schema: public

  my_mongodb:
    variant: mongodb
    mongo_uri: ${MONGO_URI}
    database: mydb

  my_s3:
    variant: file
    extra:
      storage: s3
      format: parquet
      path: s3a://my-bucket/data
    aws_access_key: ${AWS_ACCESS_KEY}
    aws_secret_key: ${AWS_SECRET_KEY}
    region: eu-west-1

Environment variables are referenced with ${VAR_NAME} syntax and resolved at load time.

Connection Parameters

Pipelines & Tables

pipelines:
  - name: my_pipeline        # unique pipeline name
    source: source_pg         # connection name for extraction
    destination: target_pg    # connection name for loading
    pass_on_error: false      # if true, continue on table failure
    tables:
      - name: public.users
        target_name: stg_users
        tags: [api, user-domain]
        replication_method: incremental
        iterate_column: updated_at
        iterate_column_type: datetime
        partitions_column: id
        partitions_count: 10
        fetchsize: 100000
        batchsize: 10000
        write_partitions: 4
        dedup_columns: [id, updated_at]
        write_strategy: upsert
        write_key: [id]
        custom_query: "(SELECT id, name, updated_at FROM users {query_filter}) q"
        transform: transforms/clean_users.py::transform
        pass_on_error: false

Table Parameters

Write Strategy

write_strategy controls how data is written to the destination. If not set, it is inferred automatically: overwrite → replace, append → append.

Usage

tables:
  # Upsert: update existing rows by primary key, insert new ones
  - name: public.users
    target_name: stg_users
    replication_method: incremental
    iterate_column: updated_at
    write_strategy: upsert
    write_key: [id]

  # Replace: full overwrite every run
  - name: public.orders
    target_name: stg_orders
    replication_method: full
    write_strategy: replace

  # Append (default for incremental): just insert
  - name: public.events
    target_name: stg_events
    replication_method: incremental
    iterate_column: created_at
    write_strategy: append

Supported Strategies per Loader

Validation Rules

• write_strategy: upsert or merge requires write_key — raises ConfigError if missing.
• write_key is ignored when strategy is append or replace.
• If write_strategy is not set, the strategy is inferred from the extractor's write mode.

If Exists (Target Table Preservation)

if_exists controls whether the target table is dropped and recreated or preserved on full load (replace strategy).

When if_exists: append is combined with write_strategy: replace, the loader truncates (clears all data from) the existing table before inserting — the table structure is preserved but data is refreshed. This is useful when you want to keep indexes, constraints, and grants intact across runs.

Behavior Matrix

Global Setting

settings:
  if_exists: replace    # default: drop and recreate target tables

Per-Table Override

tables:
  - name: public.users
    target_name: stg_users
    if_exists: append     # preserve table structure, truncate + insert on replace

  - name: public.orders
    target_name: stg_orders
    # inherits settings.if_exists (replace)

Table-level if_exists takes priority over the settings-level default. If the table does not specify it, the settings value is used.

Note: if_exists only affects the replace write strategy (full load). It has no effect on append, upsert, or merge strategies, which never drop the target table.

Column Name Casing

Some databases (Snowflake, Oracle) return uppercase column names by default. When writing to a case-sensitive target like PostgreSQL, this results in uppercase quoted identifiers that are inconvenient to query.

Use column_name_case to normalize column names before writing:

settings:
  column_name_case: lower    # lower, upper, or as_is (default: as_is)

This can also be overridden per table:

tables:
  - name: SNOWFLAKE_SCHEMA.USERS
    target_name: stg_users
    column_name_case: lower   # override for this table only

Table-level column_name_case takes priority over the settings-level default. If the table does not specify it, the settings value is used.

Incremental Replication

mkpipe uses an append-only strategy for incremental replication:

1. Extract: reads rows where iterate_column >= last_point (inclusive, no boundary loss)
2. Load: appends to the target table (never overwrites or deletes)
3. Dedup: if dedup_columns is set, a mkpipe_id (xxhash64 hash) is generated for downstream deduplication

An ingestion timestamp column is always added to every row. The column name defaults to _ingested_at and can be customized via settings.ingested_at_column.

- name: public.users
  target_name: stg_users
  replication_method: incremental
  iterate_column: updated_at
  iterate_column_type: datetime
  dedup_columns: [id, updated_at]  # mkpipe_id = xxhash64(id, updated_at)

Filter Bounds (Static Range Extraction)

By default, incremental replication uses the backend's last checkpoint (last_point) to determine which rows to extract. You can override this with static bounds using filter_lower_bound and filter_upper_bound:

tables:
  # Extract only rows where created_at >= '2024-01-01' AND created_at < '2024-07-01'
  - name: public.orders
    target_name: stg_orders
    replication_method: incremental
    iterate_column: created_at
    iterate_column_type: datetime
    filter_lower_bound: "2024-01-01"
    filter_upper_bound: "2024-07-01"

  # Lower bound only — extract everything from 2024 onwards
  - name: public.events
    target_name: stg_events
    replication_method: incremental
    iterate_column: event_id
    iterate_column_type: int
    filter_lower_bound: "1000000"

  # Upper bound only — extract everything before a cutoff
  - name: public.legacy_data
    target_name: stg_legacy
    replication_method: incremental
    iterate_column: created_at
    iterate_column_type: datetime
    filter_upper_bound: "2023-01-01"

When static bounds are set:

• The last_point from the backend is ignored — bounds take full control
• write_mode is set to overwrite (the target table is replaced with the bounded slice)
• The last_point_value is still derived from max(iterate_column) in the result set

This is useful for backfills, data migrations, or extracting a specific time window without affecting the incremental checkpoint.

Multi-Column Incremental

When a table has separate createdDate and updatedDate columns, you can pass iterate_column as a YAML list. mkpipe will use OR logic — a row is extracted if any of the columns meets the threshold:

tables:
  - name: product
    target_name: stg_product
    replication_method: incremental
    iterate_column:
      - createdDate
      - updatedDate
    iterate_column_type: datetime

Each extractor translates the list into its native OR syntax:

The last_point_value is computed as the greatest max across all listed columns, so the next run picks up any change in either column.

If you prefer a single SQL expression instead (e.g. for JDBC), you can still use a string:

iterate_column: "greatest(createdDate, updatedDate)"

Both approaches are valid. The list form has the advantage of letting each extractor use index-optimized OR queries (e.g. MongoDB IXOR index scan).

NULL Handling in iterate_column

On initial load (no previous checkpoint), the extractor automatically includes rows where iterate_column evaluates to NULL. This prevents data loss when using expressions like greatest(cdate, udate) where both source columns can be NULL.

• Initial load: WHERE (col >= min AND col <= max) OR col IS NULL
• Incremental load: WHERE col >= last_point AND col <= max (NULL rows already captured)

If all rows have a NULL iterate_column, the extractor falls back to a full extraction.

Partition Column Type Behavior

The partitions_column_type parameter controls how partition bounds are converted for Spark JDBC partitioning:

Scenario 1: No partition column specified (default)

- name: public.orders
  replication_method: incremental
  iterate_column: created_at
  iterate_column_type: datetime
  # partitions_column defaults to created_at
  # partitions_column_type inherits datetime from iterate_column_type

Scenario 2: Integer partition column specified

- name: public.customers
  replication_method: incremental
  iterate_column: updated_at
  iterate_column_type: datetime
  partitions_column: customer_id
  # partitions_column_type defaults to 'int' (most partition keys are integers)
  # Handles PostgreSQL NUMERIC/DECIMAL types correctly

Scenario 3: Explicit datetime partition column

- name: public.events
  replication_method: incremental
  iterate_column: event_id
  iterate_column_type: int
  partitions_column: event_timestamp
  partitions_column_type: datetime  # Must be explicit if different from iterate_column_type

Partition Bounds on Large Tables

Before reading, the JDBC extractor needs lowerBound/upperBound for the partition column. mkpipe resolves these with two separate queries so that databases can use their min/max index shortcuts:

-- 1. iterate bounds (index shortcut on the iterate_column index)
SELECT min(greatest(cdate,udate)), max(greatest(cdate,udate))
FROM apld_bill_rt_tax WHERE greatest(cdate,udate) >= '2026-07-28 01:38:06'

-- 2. partition bounds (unfiltered → index-only min/max on the partition key)
SELECT min(acct_bill_id), max(acct_bill_id) FROM apld_bill_rt_tax

Keeping them in one SELECT would force a full aggregate scan, which on very large tables (hundreds of GB) can take hours.

The partition bounds query is unfiltered by default. This is safe — Spark's first and last partitions are unbounded, so rows outside the range are never lost — but it can make partitions uneven, since recently inserted rows tend to cluster at one end of the key range.

Three ways to control this:

1. Static bounds (zero cost)

- name: apld_bill_rt_tax
  replication_method: incremental
  iterate_column: "greatest(cdate,udate)"
  iterate_column_type: datetime
  partitions_column: acct_bill_id
  partitions_lower_bound: 800000000
  partitions_upper_bound: 900000000

No bounds query is issued at all. A rough estimate is enough.

2. Partition on the iterate expression (recommended)

When partitions_column normalizes to the same expression as iterate_column, mkpipe reuses the iterate bounds and issues no partition bounds query. Partitions also spread evenly, because the filter and the partition key are the same expression:

- name: apld_bill_rt_tax
  target_name: raw_i2i__apld_bill_rt_tax
  replication_method: incremental
  iterate_column: "greatest(cdate,udate)"
  iterate_column_type: datetime
  dedup_columns: [acct_bill_id, seq_num]
  partitions_column: "greatest(cdate,udate) as _part_ts"
  partitions_column_type: datetime
  partitions_count: 128
  fetchsize: 10000

The as _part_ts alias is required because Spark quotes partitionColumn as an identifier and cannot use a raw SQL expression. mkpipe injects the alias into the generated query and drops it from the DataFrame after loading, so it never reaches the target table:

SELECT *, greatest(cdate,udate) AS _part_ts FROM apld_bill_rt_tax
WHERE greatest(cdate,udate) >= 'min' AND greatest(cdate,udate) <= 'max'

Each partition predicate then becomes a range scan on _part_ts, which maps directly onto a (greatest(cdate,udate)) index. If you supply your own custom_query, the alias is not injected — expose it yourself.

Note that lowerBound/upperBound support TimestampType and DateType in addition to numeric types. Timestamp values are parsed using spark.sql.session.timeZone, so keep it aligned with the database timezone.

3. Filtered bounds (accurate but slow)

partitions_bounds_filtered: true

Restores the pre-optimization behaviour: the partition bounds query carries the incremental filter. Only use this when the filter is cheap to evaluate.

Downstream dedup query example:

SELECT * FROM (
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY mkpipe_id ORDER BY _ingested_at DESC) AS rn
  FROM stg_users
) WHERE rn = 1

Inline Transformations

Add a transform field to any table:

tables:
  - name: public.products
    target_name: stg_products
    replication_method: full
    transform: transforms/clean_products.py::transform

The transform function receives and returns a PySpark DataFrame:

# transforms/clean_products.py
def transform(df):
    df = df.filter(df.status != "deleted")
    return df

Orchestrator Integration

Dagster

from dagster import asset, Definitions
import mkpipe

@asset
def api_tables():
    mkpipe.run(config="mkpipe_project.yaml", tags=["api"])

@asset
def critical_tables():
    mkpipe.run(config="mkpipe_project.yaml", tags=["critical"])

defs = Definitions(assets=[api_tables, critical_tables])

Airflow

from airflow.decorators import task

@task
def sync_api_tables():
    import mkpipe
    mkpipe.run(config="/path/to/mkpipe_project.yaml", tags=["api"])

@task
def sync_user_domain():
    import mkpipe
    mkpipe.run(config="/path/to/mkpipe_project.yaml", tags=["user-domain"])

Backend (State Tracking)

mkpipe tracks pipeline state (last sync point, status) in a manifest database. Default is SQLite (zero-config). PostgreSQL, DuckDB, and ClickHouse are also supported:

settings:
  backend:
    variant: postgres
    host: localhost
    port: 5432
    database: mkpipe_db
    user: mkpipe
    password: ${BACKEND_PASSWORD}

Install optional backend dependencies:

pip install mkpipe[postgres-backend]
pip install mkpipe[duckdb-backend]
pip install mkpipe[clickhouse-backend]
pip install mkpipe[all-backends]

Custom Exceptions

mkpipe provides specific exception classes for clean error handling:

from mkpipe import (
    MkpipeError,          # base exception
    ConfigError,          # YAML or configuration issues
    ExtractionError,      # data extraction failures
    LoadError,            # data loading failures
    TransformError,       # transformation failures
    PluginNotFoundError,  # missing plugin
    BackendError,         # backend manifest failures
)

JAR Management

mkpipe plugins that depend on JDBC drivers or Spark connectors need JAR files. mkpipe handles this automatically — no manual steps required for most users.

Online (Default) — Lazy Download

When mkpipe starts, it detects which plugins are installed and resolves their Maven dependencies via spark.jars.packages. JARs are downloaded on first run and cached by Spark's Ivy resolver.

# Nothing to do — just run your pipeline
mkpipe run

Offline / Docker — Pre-download JARs

For air-gapped or on-premise environments without internet access, pre-download all JARs during the Docker build:

mkpipe install-jars

This command:

1. Discovers all installed plugins and their Maven dependencies
2. Downloads JARs via Spark's Ivy resolver into a temp directory
3. Copies them into each plugin's jars/ directory
4. Cleans up the temp Ivy cache

Dockerfile example:

FROM python:3.11-slim

# Install Java (required for PySpark)
RUN apt-get update && apt-get install -y default-jdk && rm -rf /var/lib/apt/lists/*

# Install mkpipe and plugins
RUN pip install mkpipe mkpipe-extractor-postgres mkpipe-loader-clickhouse

# Pre-download JARs (no internet needed at runtime)
RUN mkpipe install-jars

COPY mkpipe_project.yaml .
CMD ["mkpipe", "run"]

How It Works

CLI Reference

mkpipe install-jars    # Download all Maven JARs for installed plugins

Available Plugins

For the full list, visit the mkpipe-hub.

Extractors

• PostgreSQL, MySQL, MariaDB, SQL Server, Oracle, SQLite, Redshift, ClickHouse, MongoDB, Snowflake, BigQuery, Cassandra, TimescaleDB, DynamoDB, Elasticsearch, InfluxDB, Redis, File (S3/GCS/local/Iceberg/Delta)

Loaders

• PostgreSQL, MySQL, MariaDB, SQL Server, Oracle, SQLite, Redshift, ClickHouse, MongoDB, Snowflake, BigQuery, Cassandra, TimescaleDB, DynamoDB, Elasticsearch, InfluxDB, Redis, File (S3/GCS/local/Iceberg/Delta)

License

Apache 2.0 — see LICENSE.