bizon 0.4.1

Extract and load your data reliably from API Clients with native fault-tolerant and checkpointing mechanism.

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

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.