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.

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.

Abstraction	Base Class	Location
Source	`AbstractSource`	`bizon/source/source.py`
Destination	`AbstractDestination`	`bizon/destination/destination.py`
Queue	`AbstractQueue`	`bizon/engine/queue/queue.py`
Backend	`AbstractBackend`	`bizon/engine/backend/backend.py`
Runner	`AbstractRunner`	`bizon/engine/runner/runner.py`

Installation

Requires Python ≥ 3.9, < 3.13.

pip install bizon

Optional features are installed via extras:

Extra	`pip install`	Enables
`postgres`	`bizon[postgres]`	PostgreSQL backend
`bigquery`	`bizon[bigquery]`	BigQuery backend & destinations (incl. Storage Write API)
`kafka`	`bizon[kafka]`	Kafka/Redpanda queue and Kafka source (Avro/Schema Registry)
`rabbitmq`	`bizon[rabbitmq]`	RabbitMQ queue
`gsheets`	`bizon[gsheets]`	Google Sheets source
`datadog`	`bizon[datadog]`	Datadog metrics & tracing
`secretmanager`	`bizon[secretmanager]`	`gsm://` Google Secret Manager references

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

Command	Description
`bizon run <config.yml>`	Run a pipeline from a YAML config
`bizon source list`	List available sources and their streams
`bizon stream list <source>`	List a source's streams, flagged `[Supports incremental]` / `[Full refresh only]`
`bizon secrets check <config.yml>`	Dry-run every `gsm://` / `env://` reference and report (masked) results
`bizon destination`	Subcommand group (no subcommands yet)

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

Key	Required	Description
`name`	✅	Unique name identifying this sync
`source`	✅	Source connector config (see below)
`destination`	✅	Destination connector config; routed by its `name` field
`transforms`	—	List of in-pipeline transforms (default `[]`)
`engine`	—	Backend, queue and runner config (sensible defaults if omitted)
`secrets`	—	Provider defaults for `gsm://` / `env://` resolution
`monitoring`	—	Datadog metrics & tracing
`alerting`	—	Slack alerting
`streams`	—	Multi-table routing (requires `source.sync_mode: stream`)

`source` keys

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

Key	Default	Description
`name`	—	Connector name (e.g. `hubspot`, `notion`, `kafka`)
`stream`	—	Stream to sync
`sync_mode`	`full_refresh`	`full_refresh` \| `incremental` \| `stream`
`cursor_field`	`None`	Timestamp field for incremental filtering (e.g. `updated_at`)
`authentication`	`None`	Auth block (`type` + `params`); connector-specific
`force_ignore_checkpoint`	`false`	Ignore existing checkpoints and restart from iteration 0
`max_iterations`	`None`	Cap iterations per run (default: run until source is exhausted)
`api_config.retry_limit`	`10`	Retries before giving up on an API call
`source_file_path`	`None`	Path to a custom source file (same as `--custom-source`)

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.

Source	Connects to	Example streams	Auth	Incremental	Extra
`hubspot`	HubSpot CRM (v3)	`contacts`, `companies`, `deals`	api_key, oauth	—	—
`notion`	Notion API	`pages`, `databases`, `data_sources`, `blocks`, `blocks_markdown`, `users` (+ `all_*`)	api_key	✅	—
`kafka`	Kafka / Redpanda	`topic` (Avro + Schema Registry)	basic (SASL)	stream	`kafka`
`gsheets`	Google Sheets	`worksheet`	service account, ADC	—	`gsheets`
`cycle`	Cycle (GraphQL)	`customers`	api_key	—	—
`periscope`	Periscope / Sisense	`charts`, `dashboards`, `views`, `users`, `databases`	cookies	—	—
`sana_ai`	Sana AI Insight Reports	`insight_report`	oauth	—	—
`pokeapi`	PokéAPI (public)	`pokemon`, `berry`, `item`	none	—	—
`gbif`	GBIF biodiversity (public)	`occurrence`	none	—	—
`dummy`	Mock source (testing/demos)	`creatures`, `plants`	api_key, oauth	—	—

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

Destinations

Destination	Writes to	Sync modes	Notable features
`bigquery`	BigQuery	full / incremental / stream	Batch loads via GCS + Parquet; atomic table swaps via free copy jobs; async/batched load jobs; partitioning; optional schema unnesting
`bigquery_streaming`	BigQuery	full / incremental / stream	Legacy streaming insert API; dynamic schema evolution; large-row fallback to load jobs
`bigquery_streaming_v2`	BigQuery	full / incremental / stream	Storage Write API (protobuf); higher throughput; multi-threaded appends; schema caching
`file`	Local NDJSON file	full / incremental / stream	Atomic finalize on full refresh; append on incremental; optional unnesting
`logger`	stdout (loguru)	full / incremental / stream	Logs records — for testing & debugging

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

Mode	Behavior
`full_refresh`	Re-syncs all data from scratch on each run (default)
`incremental`	Syncs only new/updated records since the last successful run (append-only)
`stream`	Continuous streaming for real-time data (e.g. Kafka)

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:

Source	Cursor Field	Notes
`notion`	`last_edited_time`	Supports `pages`, `databases`, `blocks`, `blocks_markdown` streams
(others)	Varies	Check source docs or implement `get_records_after()`

Supported Destinations

Destinations must implement finalize() with incremental logic:

Destination	Support	Notes
`bigquery`	✅	Append-only via temp table
`bigquery_streaming_v2`	✅	Append-only via temp table
`file`	✅	Appends to existing file
`logger`	✅	Logs completion

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

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.

Type	Use case
`sqlite`	File-based SQLite — local development (default)
`sqlite_in_memory`	Memory-only SQLite — unit tests
`postgres`	PostgreSQL — production, frequent cursor updates (`bizon[postgres]`)
`bigquery`	BigQuery — lightweight production state storage (`bizon[bigquery]`)

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.

Type	Use case
`python_queue`	In-process — local development & testing (default)
`rabbitmq`	RabbitMQ — production, high throughput (`bizon[rabbitmq]`)
`kafka`	Kafka / Redpanda — production, high throughput, strong persistence (`bizon[kafka]`)

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.

Type	Description	Best for
`thread`	`ThreadPoolExecutor` (default)	I/O-bound sources (APIs, DBs)
`process`	`ProcessPoolExecutor`, true parallelism	CPU-bound sources/transforms
`stream`	Single-threaded, inline	Real-time `stream` sync & multi-stream routing

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.

Name		Name	Last commit message	Last commit date
Latest commit History 186 Commits
.claude/skills		.claude/skills
.github		.github
.vscode		.vscode
bizon		bizon
docs		docs
scripts		scripts
tests		tests
.env.example		.env.example
.gitignore		.gitignore
.pre-commit-config.yaml		.pre-commit-config.yaml
AGENTS.md		AGENTS.md
CHANGELOG.md		CHANGELOG.md
CLAUDE.md		CLAUDE.md
CONTRIBUTING.md		CONTRIBUTING.md
LICENSE		LICENSE
Makefile		Makefile
README.md		README.md
docker-compose.yml		docker-compose.yml
pyproject.toml		pyproject.toml
uv.lock		uv.lock

Uh oh!

Folders and files

Latest commit

History

Repository files navigation

bizon ⚡️

Table of Contents

Features

Architecture

Installation

For Development

Quickstart

CLI Reference

bizon run

bizon secrets check

Configuration Reference

source keys

Annotated example

Connectors

Sources

Destinations

Sync Modes

Incremental Sync

Configuration

How It Works

Supported Sources

Supported Destinations

Example: Notion Incremental Sync

First Run Behavior

Engine Configuration

Backends (state storage)

Queues

Runners

Transforms

Secrets & References

Monitoring & Alerting

Multi-Stream Routing

Documentation & Contributing

License

About

Resources

License

Contributing

Uh oh!

Stars

Watchers

Forks

Releases 21

Packages 0

Uh oh!

Uh oh!

Contributors

Uh oh!

Languages

`bizon run`

`bizon secrets check`

`source` keys

Packages