bizon ⚡️

Extract and load your largest data streams with a framework you can trust for billions of records.

Bizon is a lightweight Python ETL framework built around a checkpointed producer → queue → consumer pipeline. It extracts data from API clients, databases and event streams, optionally transforms records in flight with Python, and loads them into your warehouse — with native fault-tolerance, resumable checkpoints, and very high throughput, while staying small enough to read end to end.

Table of Contents

• Features
• Architecture
• Installation
• Quickstart
• CLI Reference
• Configuration Reference
• Connectors
  • Sources
  • Destinations
• Sync Modes
  • Incremental Sync
  • Stream Reset
• Engine Configuration
  • Backends
  • Queues
  • Runners
• Transforms
• Secrets & References
• Monitoring & Alerting
• Multi-Stream Routing
• Documentation & Contributing
• License

Features

• Natively fault-tolerant — a checkpointing mechanism tracks progress so a pipeline resumes from its last committed cursor after a crash or restart.
• High throughput — designed to process billions of records, using Polars DataFrames for memory-efficient, vectorized buffering and Parquet for batch loads.
• Queue-system agnostic — run on an in-process Python queue, RabbitMQ, or Kafka/Redpanda behind a single Queue interface; adapters can be written for any broker.
• Pluggable connectors — 10 built-in sources and 5 destinations, all behind clean AbstractSource / AbstractDestination interfaces. Sources are auto-discovered — no registration needed.
• Multiple sync modes — full_refresh, incremental (append-only), and continuous stream.
• Secret resolution — keep secrets out of YAML with gsm:// (Google Secret Manager) and env:// references, resolved before validation so connectors only ever see plain strings.
• In-pipeline transforms — apply user-defined Python to records in flight.
• Pipeline metrics — exhaustive metrics with Datadog & OpenTelemetry tracing: ETAs, records processed, completion %, and source ↔ destination latency.
• Lightweight & lean — a minimal codebase with few core dependencies (requests, pyyaml, pydantic, sqlalchemy, polars, pyarrow).

Architecture

Bizon uses a producer–consumer pattern with pluggable components:

YAML Config → RunnerFactory → Producer → Queue → Consumer → Destination
                                  ↑                    ↓
                                Source              Backend (checkpoints)

• The Producer pulls records from the Source in iterations and pushes them onto the Queue as Polars DataFrames.
• The Consumer pulls from the queue, applies Transforms, and writes batches to the Destination.
• The Backend persists cursors so syncs are resumable.

Checkpointing & recovery. The producer writes its source cursor to the backend every syncCursorInDBEvery iterations, and the consumer records a destination cursor after each successful write. On restart, the pipeline reads the last destination cursor and resumes from the next iteration. Bizon's delivery contract is at-least-once — on recovery a batch may be re-written, so destinations are designed to tolerate duplicate writes.

Installation

Requires Python ≥ 3.9, < 3.13.

pip install bizon

Optional features are installed via extras:

Combine extras as needed, e.g. pip install 'bizon[bigquery,kafka,secretmanager]'.

For Development

# Install uv (if not already installed)
pip install uv

# Clone and install with all extras and dependency groups
git clone https://github.com/bizon-data/bizon-core.git
cd bizon-core
uv sync --all-extras --all-groups

# Run tests
uv run pytest tests/

# Format & lint (Ruff, line length 120)
uv run ruff format .
uv run ruff check --fix .

Quickstart

This example needs zero external dependencies — it uses the in-process Python queue and a SQLite backend (both defaults), reading from the built-in dummy source and writing to logger.

Create config.yml:

name: demo-creatures-pipeline

source:
  name: dummy
  stream: creatures
  authentication:
    type: api_key
    params:
      token: dummy_key

destination:
  name: logger
  config:
    dummy: dummy

Run it:

bizon run config.yml

CLI Reference

The CLI entry point is bizon (bizon.cli.main:cli).

bizon run

bizon run config.yml \
  --custom-source ./my_source.py \   # Custom Python file implementing a Bizon source
  --runner thread \                  # thread | process | stream (default: thread)
  --log-level INFO \                 # DEBUG | INFO | WARNING | ERROR | CRITICAL
  --env-file .env                    # Load env vars from a .env file (auto-detected if omitted)

The --runner flag overrides engine.runner.type in the config; --log-level overrides engine.runner.log_level.

bizon secrets check

bizon secrets check config.yml [--env-file .env]

Resolves every reference in the config (without running the pipeline) and prints each one with a masked ✓/✗ status. Exits non-zero if any reference fails — handy as a pre-deploy gate.

Configuration Reference

A pipeline is defined by a single YAML file validated against BizonConfig (bizon/common/models.py). Unknown top-level keys are rejected.

source keys

Common SourceConfig fields (bizon/source/config.py); each connector adds its own:

Annotated example

name: hubspot contacts to bigquery

source:
  name: hubspot
  stream: contacts
  properties:
    strategy: all          # connector-specific: fetch all properties
  authentication:
    type: api_key
    api_key: ${env://HUBSPOT_API_KEY}   # inline secret reference

destination:
  name: bigquery
  config:
    project_id: my-gcp-project-id
    dataset_id: bizon_test
    dataset_location: US
    create_dataset: false
    gcs_buffer_bucket: bizon-buffer
    gcs_buffer_format: parquet
    buffer_size: 10            # in MB
    buffer_flush_timeout: 300  # in seconds

engine:
  backend:
    type: bigquery
    config:
      database: my-gcp-project
      schema_name: bizon_backend
    syncCursorInDBEvery: 10

Full, copy-pasteable examples live next to each connector under bizon/connectors/**/config/*.example.yml.

Connectors

Sources

All sources are auto-discovered. Run bizon source list to see what's installed and bizon stream list <source> for a source's streams.

notion is the reference incremental source — see bizon/connectors/sources/notion/src/source.py for get_records_after().

Destinations

Adding connectors. Sources are auto-discovered — drop a connector under bizon/connectors/sources/{name}/src/ and it appears. Destinations register in three places (see the guide). Use the /new-source and /new-destination Claude skills, or read docs/contributing/adding-sources.md and docs/contributing/adding-destinations.md.

Sync Modes

Incremental Sync

Incremental sync fetches only new or updated records since the last successful run, using an append-only strategy.

Configuration

source:
  name: your_source
  stream: your_stream
  sync_mode: incremental
  cursor_field: updated_at  # The timestamp field to filter records by

How It Works

┌─────────────────────────────────────────────────────────────────────┐
│                        INCREMENTAL SYNC FLOW                        │
├─────────────────────────────────────────────────────────────────────┤
│                                                                     │
│  1. Producer checks for last successful job                         │
│     └─> Backend.get_last_successful_stream_job()                    │
│                                                                     │
│  2. If found, creates SourceIncrementalState:                       │
│     └─> last_run = previous_job.created_at                          │
│     └─> cursor_field = config.cursor_field (e.g., "updated_at")     │
│                                                                     │
│  3. Calls source.get_records_after(source_state, pagination)        │
│     └─> Source filters: WHERE cursor_field > last_run               │
│                                                                     │
│  4. Records written to temp table: {table}_incremental              │
│                                                                     │
│  5. finalize() appends temp table to main table                     │
│     └─> BigQuery: free copy job (WRITE_APPEND), creating the         │
│         main table on first run; other destinations: INSERT INTO     │
│     └─> Deletes temp table                                          │
│                                                                     │
│  FIRST RUN: No previous job → falls back to get() (full refresh)    │
│                                                                     │
└─────────────────────────────────────────────────────────────────────┘

Supported Sources

Sources must implement get_records_after() to support incremental sync:

Supported Destinations

Destinations must implement finalize() with incremental logic:

Example: Notion Incremental Sync

name: notion pages incremental sync

source:
  name: notion
  stream: pages           # Options: databases, data_sources, pages, blocks, users
  sync_mode: incremental
  cursor_field: last_edited_time
  authentication:
    type: api_key
    params:
      token: secret_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
  database_ids:
    - "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"
  page_size: 100

destination:
  name: bigquery
  config:
    project_id: my-gcp-project
    dataset_id: notion_data
    dataset_location: US
    gcs_buffer_bucket: my-gcs-bucket
    gcs_buffer_format: parquet

engine:
  backend:
    type: bigquery
    config:
      database: my-gcp-project
      schema_name: bizon_backend
    syncCursorInDBEvery: 2

First Run Behavior

On the first incremental run (no previous successful job):

• Falls back to the get() method (full-refresh behavior)
• All data is fetched and loaded
• The job is marked successful
• Subsequent runs use get_records_after() with the last_run timestamp

Stream Reset

A reset re-fetches the whole stream once and replaces the destination table, then resumes incremental from that run. Use it when the table has drifted — a backfill, a bug in a transform, or records the source changed without bumping its cursor field.

There are three ways to ask for one; all do the same thing:

# One-shot, for a run you launch yourself
bizon run config.yml --reset

# Queued in the backend, for a pipeline whose command line is fixed by a scheduler.
# The next `bizon run config.yml` picks it up — no change to the cron/Airflow job.
bizon stream reset config.yml
bizon stream reset config.yml --cancel   # changed your mind

# One config templated across streams? Pick which stream to reset.
bizon stream reset config.yml --stream deals

source:
  sync_mode: incremental
  reset: true      # same effect, set in the config

What the run does differently:

• Producer ignores the watermark and calls get() instead of get_records_after()
• Destination materializes the run as a full refresh, replacing the table rather than appending to it (for bigquery: staging into {table}_temp, then a WRITE_TRUNCATE copy job)
• The job row stays incremental, so the next run uses this run as its new watermark

Notes:

• Scoped to a single stream. The request is keyed on (name, source_name, stream_name) — the same triple as the watermark it overrides — so resetting one stream never affects another, even under the same pipeline name. bizon stream reset takes the stream from the config; --stream overrides it.
• Only meaningful with sync_mode: incremental; ignored (with a warning) for the other modes.
• Supported by every destination with a working full-refresh path, since that is what a reset materializes as. The exception is bigquery_streaming, which has no staging table and appends even on a full refresh; a reset there is rejected at config validation rather than duplicating your data.
• A reset that crashes is retried as a reset — the request stays bound to its job, so a rerun cannot silently degrade into an append.
• bizon stream reset writes to the backend, so it must reach the same one the pipeline uses. With the default file-based sqlite backend that means the same machine and file.

Engine Configuration

The engine block configures three pluggable subsystems. All have defaults, so engine is optional for local runs.

Backends (state storage)

The backend stores Bizon's state (jobs and cursors). Configured under engine.backend.

syncCursorInDBEvery (default 2) controls how often the source cursor is flushed to the backend — lower values mean finer-grained recovery, higher values mean less write overhead.

engine:
  backend:
    type: postgres
    config:
      host: localhost
      port: 5432
      database: bizon
      schema_name: bizon
      username: ${env://PG_USER}
      password: ${env://PG_PASSWORD}
    syncCursorInDBEvery: 10

Queues

The queue carries records from the producer to the consumer. Configured under engine.queue.

Spin up a broker locally with the provided compose files:

docker compose --file ./scripts/queues/kafka-compose.yml up      # Kafka
docker compose --file ./scripts/queues/redpanda-compose.yml up   # Redpanda
docker compose --file ./scripts/queues/rabbitmq-compose.yml up    # RabbitMQ

# Kafka
engine:
  queue:
    type: kafka
    config:
      queue:
        bootstrap_server: localhost:9092   # Kafka:9092 / Redpanda:19092

# RabbitMQ
engine:
  queue:
    type: rabbitmq
    config:
      queue:
        host: localhost
        queue_name: bizon

Runners

The runner controls how the producer and consumer execute. Configured under engine.runner.

Transforms

Transforms apply user-defined Python to each record as it flows through the pipeline. Each transform receives the record's parsed payload as a data dict, mutates it, and the result is re-serialized. Transforms are applied sequentially, in order.

transforms:
  - label: debezium
    python: |
      from datetime import datetime

      cluster = data['value']['source']['name'].replace('_', '-')
      partition = data['partition']
      offset = data['offset']
      kafka_timestamp = datetime.utcfromtimestamp(
          data['value']['source']['ts_ms'] / 1000
      ).strftime('%Y-%m-%d %H:%M:%S.%f')

      deleted = False
      if data['value']['op'] == 'd':
          data = data['value']['before']
          deleted = True

A TransformModel has just two fields: label (a display name) and python (the code). There are no built-in transforms — the logic is entirely yours.

Secrets & References

Keep secrets out of YAML by referencing them with a URI scheme. Resolution runs once over the raw config before validation (bizon/engine/resolvers/), so connectors need no changes — they always read plain strings.

• gsm://<id> → Google Secret Manager, latest version (ADC auth). Pin with gsm://<id>/versions/<N>, or pass a full gsm://projects/<p>/secrets/<id>/versions/<N> path. Requires bizon[secretmanager].
• env://<VAR> → environment variable.
• Inline form — embed in a larger string with ${...}, e.g. dsn: "postgres://u:${gsm://db-pw}@host/db" (multiple tokens allowed).
• An optional secrets: block holds provider defaults (e.g. secrets.gsm.project_id).

Validate every reference before running:

bizon secrets check config.yml

Add a provider by dropping one adapter in bizon/engine/resolvers/adapters/ and one entry in _SCHEME_FACTORIES (bizon/engine/resolvers/resolver.py).

Monitoring & Alerting

Monitoring (bizon/monitoring/) — emit pipeline metrics and traces to Datadog. Install bizon[datadog].

monitoring:
  type: datadog
  config:
    enable_tracing: true
    datadog_agent_host: localhost   # or datadog_host_env_var: DD_AGENT_HOST
    datadog_agent_port: 8125
    tags:
      env: production
      team: data

Alerting (bizon/alerting/) — post alerts to Slack on configured log levels.

alerting:
  type: slack
  log_levels: [ERROR]              # defaults to [ERROR]
  config:
    webhook_url: ${env://SLACK_WEBHOOK_URL}

Multi-Stream Routing

For one-to-many pipelines (common with Kafka CDC), the streams block maps multiple source streams to their own destination tables and schemas in a single run. It requires source.sync_mode: stream and the stream runner.

source:
  name: kafka
  stream: topic
  sync_mode: stream
  # topics are auto-extracted from the streams block below
  bootstrap_servers: your-kafka-broker:9092
  group_id: your-consumer-group

destination:
  name: bigquery_streaming_v2
  config:
    project_id: your-gcp-project
    dataset_id: your_dataset
    dataset_location: US
    unnest: true

streams:
  - name: users
    source:
      topic: cdc.public.users
    destination:
      table_id: your-gcp-project.your_dataset.users
      clustering_keys: [id]
      record_schema:
        - { name: id, type: INTEGER, mode: REQUIRED }
        - { name: email, type: STRING, mode: NULLABLE }
  - name: orders
    source:
      topic: cdc.public.orders
    destination:
      table_id: your-gcp-project.your_dataset.orders
      clustering_keys: [id, user_id]

engine:
  runner:
    type: stream

See bizon/connectors/sources/kafka/config/kafka_streams.example.yml for a complete example.

Documentation & Contributing

• Adding sources — docs/contributing/adding-sources.md
• Adding destinations — docs/contributing/adding-destinations.md
• AI-assisted connector generation — docs/ai-connector-guide.md, docs/ai-destination-guide.md
• Annotated reference connector — docs/reference-connector.md
• Contribution guidelines — CONTRIBUTING.md
• Release notes — CHANGELOG.md

Claude Code skills are available for common workflows: /new-source, /new-destination, and /run-checks.

License

Bizon is released under the GNU General Public License v3.0. See LICENSE.