bluesky-zerobus

bluesky-zerobus reads Bluesky Jetstream's v2 WebSocket with jetstream-lakehouse and sends normalized events to a Unity Catalog Delta table. The sink uses either the official Databricks Zerobus Python SDK over gRPC or Zerobus' Kafka-compatible producer API. The connector follows live data and resumes from the last locally saved Jetstream sequence after a restart. Archived backfill can be added later.

Prepare Databricks

Zerobus Ingest must be available in your workspace region. Create a table with the same column names and types as the mapper below; change the fully qualified table name as needed:

CREATE TABLE main.bronze.bluesky_events (
  source_id STRING NOT NULL,
  seq BIGINT NOT NULL,
  event_time STRING,
  witnessed_at STRING,
  did STRING,
  kind STRING,
  operation STRING,
  collection STRING,
  rkey STRING,
  cid STRING,
  rev STRING,
  is_resync BOOLEAN,
  record STRING,
  event_payload STRING
) USING DELTA;

record and event_payload contain JSON text. Identity, account, sync, create, update, and delete events all land as separate rows. Build a downstream view or table to fold events into current record state if needed.

Create a Databricks service principal and grant USE CATALOG, USE SCHEMA, SELECT, and MODIFY on the destination. Its application ID and secret belong in .secrets.toml. Follow the Databricks setup guide for the endpoint and grants.

Configure and run

Install dependencies with uv, then copy the example settings into the ignored settings.local.toml. Set your workspace, Zerobus endpoint, table, source ID, and batch limits there. On Azure, the endpoint is https://<workspace-id>.zerobus.<region>.azuredatabricks.net; on AWS, it is https://<workspace-id>.zerobus.<region>.cloud.databricks.com. Get a retained start cursor from the same WebSocket subscription, and set jetstream.start_cursor to the printed number before the first run. Cursor discovery needs no Databricks credentials. Copy the credential template into a local, ignored file:

uv sync --locked
cp settings.toml settings.local.toml
uv run bluesky-zerobus --config settings.local.toml --discover-cursor
# Set jetstream.start_cursor in settings.local.toml to the printed number.
cp credentials.example.toml .secrets.toml
chmod 600 .secrets.toml
uv run bluesky-zerobus --config settings.local.toml --secrets .secrets.toml

Edit .secrets.toml before running. Dynaconf reads the two explicit TOML files. The connector does not load application settings or credentials from environment variables. You can pass other file paths with --config and --secrets.

Choose [sink].type = "grpc" (the default) or "kafka" in the settings TOML. Both sinks use the same [zerobus] endpoint, workspace URL, destination table, and service principal credentials. For a side-by-side comparison, use separate destination tables, pipeline IDs, and checkpoint files so each sink processes its own source window. The Kafka-compatible API is Beta and may need to be enabled in the workspace Previews page.

A new checkpoint requires an explicit start_cursor, so a failure before the first acknowledged batch cannot silently restart from a later live tip. The discovery command subscribes at the current tip, waits for one matching event, and prints its sequence minus one. A saved checkpoint takes precedence on subsequent runs. If the WebSocket can no longer serve a saved cursor, the connector fails rather than skip data. Choose a recovery or archive backfill procedure before changing that checkpoint.

The SQLite checkpoint is stored at pipeline.checkpoint_path, relative to the settings file unless absolute. Put it on persistent storage when deploying. A process lock prevents simultaneous consumers using the same checkpoint. The checkpoint is tied to pipeline.id and a fingerprint of source, filters, source ID, and destination table; changing those settings requires a new pipeline ID. The Databricks credentials are excluded from the fingerprint. source_id identifies a Jetstream serving instance for downstream deduplication.

Stop with SIGINT or SIGTERM. The reader stops, pending rows are flushed, the confirmed cursor is saved, and the Zerobus stream closes.

Set [logging].level in the settings TOML to DEBUG, INFO, WARNING, ERROR, or CRITICAL (INFO by default). This controls only the connector's loggers. Logs go to stderr and report startup, cursor origin, confirmed batches, shutdown, and failure types. Batch confirmation is logged only after the Zerobus acknowledgment and SQLite checkpoint commit. The connector's own log messages omit event payloads and credentials. When running as a Databricks Python task, these messages appear in the task output; Zerobus service activity is available separately in its system tables.

Delivery contract

The source runs in a separate thread with a bounded queue. The WebSocket frame limit matches the configured batch byte limit, and the default queue holds at most 32 events. The producer groups rows by count, UTF-8 byte size, or time. The SDK's ingest_records_offset() queues a batch; flush() waits until Zerobus confirms it is durable. Only then does the connector save the last event sequence in SQLite. The default 4 MiB batch ceiling stays below Zerobus' 10 MB JSON message limit; the ceiling is an estimate of row JSON bytes and leaves room for SDK framing. A record larger than the configured ceiling stops the connector without advancing its cursor.

This provides at-least-once delivery. A crash between Zerobus' confirmation and SQLite's checkpoint commit can repeat a row. Use (source_id, seq) for deduplication downstream. A durability acknowledgment precedes query visibility in the Delta table. See Databricks' acknowledgment guide.

Sink.write_batch(rows) defines the destination contract: it returns only after all rows are durable, or raises. ZerobusGrpcSink uses the official SDK's JSON stream and recovery. ZerobusKafkaSink reuses a long-lived Kafka producer, obtains a fresh table-scoped OAuth token on reconnect, and sends UTF-8 JSON values to a topic named after the full table name. It uses SASL_SSL, OAUTHBEARER, acks=all, and no compression. It checks every produce future before the source checkpoint can advance. A failed batch may have written some rows; replay can therefore duplicate them. Kafka's consumer, admin, and transaction APIs are not available on this endpoint. See the Azure Databricks Kafka-compatible API guide.

Development

uv sync --locked
uv run ruff check .
uv run pytest

Unit tests use a fake WebSocket iterator and sink. A live end-to-end run requires your Jetstream endpoint and a Databricks workspace and is not part of the default test suite.

License

Apache License 2.0. See LICENSE.