Streaming Worker Config Reference

Reference for the YAML config consumed by pycharter stream run. Fields are grouped by location in the document.

CLI companions: pycharter stream validate <config> runs the config-side preflight (contract resolution, sink/source key checks — no Kafka I/O); stream run --var KEY=VALUE overrides variables:; stream run --dry-run [--max-records N] previews validation without writing sinks (ephemeral consumer group for kafka sources); stream list / stream inspect discover and summarise configs. See the streaming validation guide.

api_version: pycharter.io/v2
kind: StreamingWorker
metadata:
  name: orders
  namespace: example
  version: "1.0.0"
  title: orders
spec:
  variables: {...}
  streams: [...]
  observability: {...}

Worker files carry the unified manifest envelope (KM): api_version + kind + identity in metadata (namespace/name/version) + the body under spec:. Files at the older formats (the TH envelope with top-level format_version/id/ version, or the pre-TH streaming: wrapper) upgrade on read or via pycharter migrate, which rewrites them in place. API payloads and store documents remain section-based (streams, variables, observability, metadata) — the envelope is the file format.

The schema is validated by pycharter.streaming.config.StreamingConfig (source). Unknown keys are rejected; stream names must be unique within a config.

Envelope: api_version, kind, and metadata

Field	Type	Default	Notes
api_version	string	—	pycharter.io/v2 — the manifest envelope version (KM-2). Required.
kind	string	—	StreamingWorker. Required.
metadata.name	string	—	Identity slug (^[a-z0-9_-]+$). With metadata.namespace it forms the stored artifact's name key (namespace/name). Required.
metadata.namespace	string	none	Identity scope.
metadata.version	string	—	Config content version (semver); the registry keys on (namespace/name, version). Required.
variables	mapping[string, string]	{}	Default values for ${VAR} placeholder substitution, declared under spec:. Declared values take priority over the worker's environment; ${VAR:-default} and ${VAR:?error} modifiers behave identically to batch. Resolved at worker load time and by the API's source-tap; editable in the UI's Variables tab.
metadata	mapping	none	Cataloguing metadata, mirroring a batch pipeline config's metadata block. Optional; the store derives title/namespace defaults from id when absent.

metadata fields (when present):

Field	Type	Default	Notes
title	string	—	Human-readable config title. Required within metadata.
description	string	none	Longer description of the config's purpose.
owner	string	none	Owning team or person.
tags	list[string]	[]	Free-form tags for search/filtering.
namespace	string	none	Slash-delimited browse folder (e.g. market-data/massive).

metadata.version was deleted in format 2 (TH-1): the artifact version lives at the top level only. A document still carrying it is rejected with a pointer to pycharter migrate.

format_version: "2"
id: market-data/massive
version: "1.0.0"
metadata:
  title: Massive market data
  description: Normalize and validate massive.com market-data events.
  owner: data-platform
  tags: [streaming, market-data, massive]
  namespace: market-data/massive
variables:
  KAFKA_BROKERS: localhost:9092   # default for ${KAFKA_BROKERS}; wins over the env
streams: [...]

streams[]

One stream = one source-to-sink validation pipeline. A worker can run any number concurrently.

streams:
  - name: orders-validator
    source: {...}
    contract: {...}        # batch-aligned: ref + validate/coerce/annotate
    transform: {...}       # optional: rename / convert / add / select / drop
    mode: enforce
    delivery: at_least_once
    load: { target: {...} }
    dlq:  { target: {...} }
    backpressure: {...}
    quality: {...}

The contract / transform / load blocks mirror a batch pipeline step so the "contract specification" reads the same in both engines.

Required fields

Field	Type	Default	Notes
name	string	—	Used in metrics labels (stream=...), logs, and the readiness probe. Must be unique within a worker.
source	mapping	—	See source.
contract	mapping	—	See contract.
load	mapping	—	{ target: {...} }. See Sinks.
dlq	mapping	—	{ target: {...} }. See Sinks.

Optional fields

Field	Type	Default	Notes
transform	mapping	none	Per-message transform applied before validation (rename / convert / add / select / drop, same operators as the batch transform step). A transform error routes the record to the DLQ.
mode	enum	enforce	detect / enforce / shadow. See Operating modes.
delivery	enum	at_least_once	at_least_once / at_most_once / exactly_once. exactly_once is a transactional Kafka read-process-write path: it requires kafka source and sinks, supports transform and the contract toggles (incl. annotate), but rejects window — window state cannot participate in the Kafka transaction; use at_least_once with window.checkpoint for durable windows.
window	mapping	none	Windowed aggregation (tumbling / sliding / session). When set, the stream emits window rows instead of individual records. See window.
backpressure	mapping	defaults shown below	See below.
quality	mapping	{windows: [1m, 5m, 1h]}	See below.

streams[].source

The source carries connection details only — the contract lives in the contract block, not on the source. kafka is the first-class type (auth fields mirror KafkaExtractor's constructor); every other registered extractor type is built via the shared extractor factory with its own keys: websocket, sse, rabbitmq, sqs, redis (Redis Streams: stream + consumer_group), mongodb, file_watcher, plus the batch-ish http / graphql / file / database / cloud_storage. pycharter stream validate reports an unregistered type with the full list.

source:
  type: kafka
  topic: orders
  bootstrap_servers: kafka:9092
  consumer_group: orders-validator
  auto_offset_reset: latest             # latest | earliest
  poll_timeout: 1.0                     # seconds per poll
  security_protocol: PLAINTEXT          # PLAINTEXT | SASL_PLAINTEXT | SASL_SSL | SSL
  sasl_mechanism: SCRAM-SHA-512
  sasl_username: ...
  sasl_password: ...                    # never inline; read via env interpolation
  ssl_cafile: /etc/certs/ca.pem

Field	Type	Default	Notes
type	string	kafka	kafka, or any registered extractor type (websocket / sse / rabbitmq / sqs / redis / mongodb / file_watcher / …) via the extractor factory.
topic	string	—	Source topic (kafka).
bootstrap_servers	string	—	Comma-separated broker addresses.
consumer_group	string	stream name	Kafka group id.
auto_offset_reset	enum	latest	Where to start when no committed offset exists.
poll_timeout	float	1.0	Seconds the consumer blocks per poll cycle.
security_protocol	enum	PLAINTEXT	PLAINTEXT / SASL_PLAINTEXT / SASL_SSL / SSL.
sasl_mechanism	string	none	e.g. SCRAM-SHA-256, SCRAM-SHA-512.
sasl_username	string	none	SASL username.
sasl_password	string	none	Never inline. Use env interpolation.
ssl_cafile	path	none	CA bundle for TLS.

Sinks: load and dlq

Each sink is a { target: {...} } block (same shape as a batch load step's target). The valid sink writes successfully validated records, the DLQ sink writes the §3.7 envelope.

load:
  target:
    type: kafka
    topic: orders-validated
    bootstrap_servers: kafka:9092
    # plus the same auth fields as source

dlq:
  target:
    type: kafka
    topic: orders-dlq
    bootstrap_servers: kafka:9092

target.type	Required keys	Optional keys	Notes
kafka	topic, bootstrap_servers	same auth fields as source	The valid sink re-stamps pycharter-contract / pycharter-contract-version headers; the DLQ sink stamps pycharter-dlq-reason: <reason>. Idempotent producer, acks=all.
file	path	—	Records appended as JSONL, flushed per write.
postgres / database / sqlite	connection_string (or url / database.url), table	schema (default public), valid sink: record_column (default record), contract_column (contract_id), version_column (contract_version); DLQ sink: envelope_column (envelope)	Records inserted as JSON, committed per write. Table/schema/column identifiers are allowlist-validated.

streams[].contract

Batch-aligned contract specification — the same shape as a pipeline step's contract: block, plus the streaming-only resolution knobs.

contract:
  ref:
    type: store                          # store | file
    name: orders                         # store: contract name
    version: "2"                         # store: version (omit for latest)
    # path: /app/contracts/orders        # file: directory of split artifacts
  store:                                 # required when ref.type=store
    type: postgres                       # sqlite | postgres | mongodb | redis | in_memory
    connection_string: ${DATABASE_URL}
  resolution: static                     # static | header
  on_missing_header: strict              # strict | fallback_to_static
  validate: true
  coerce: true
  annotate: false
  privacy:                               # optional — see docs/guides/privacy-flags.md
    mode: detect                         # off | detect | enforce
    clearance: []                        # flags this stream's sink may receive in clear

Field	Type	Default	Notes
ref.type	enum	store	store (registry lookup) or file (single contract from ref.path).
ref.name / ref.version	string	—	Store lookup key. name required for static / fallback resolution.
ref.path	path	—	Directory of split artifacts (schema/coercion/validation). Required when ref.type=file.
store	mapping	none	Contract-store connection; required when ref.type=store.
resolution	enum	static	static uses ref; header reads the pycharter-contract headers per message (requires ref.type=store).
on_missing_header	enum	strict	When resolution=header: strict → DLQ; fallback_to_static → use ref (requires ref.name).
validate / coerce / annotate	bool	true / true / false	Mirror the batch toggles with batch semantics (order coerce → validate → annotate). validate: false passes every record through without validation-failure DLQ routing (decode / unknown-contract routing is unchanged). coerce: false skips the contract's coercion_rules. annotate: true stamps _concept_annotations from the contract's field bindings. For identical coercion across modes, author rules with the shared coerce_to_* operation names.
privacy	mapping	{mode: detect}	How contract-flagged fields (field_bindings entries with pii: true / restricted: true) are treated at the valid sink and the DLQ — the same block as the batch step contract. detect never mutates and counts flagged values reaching the sink in clear; enforce tokenizes pii (salted HMAC, equal to the KB match_token) and redacts restricted to null immediately before each valid-sink write (after the KB-emitter / extraction-sink tees) unless the flag is listed in clearance; under enforce a quarantined payload has every flagged field redacted and an unclassifiable payload (no contract / undecodable) is withheld as {"_withheld": true, "sha256", "bytes"}; if tokenization is needed and no PYCHARTER_KB_ORG_SECRET is set, records are withheld to the DLQ (privacy_error) instead of written in clear. off restores pre-feature behaviour. Requires ref.type: store for enforce: the split-directory ref.type: file form carries no field_bindings, so the worker refuses enforce at build time and warns that detect reports nothing.

streams[].window

Windowed aggregation: validated records are aggregated per key and the stream emits window rows (one dict per closed window) to the valid sink instead of individual records.

window:
  type: tumbling                # tumbling | sliding | session
  size: 60                      # tumbling/sliding; seconds
  slide: 30                     # sliding only
  gap: 300                      # session only
  key: [symbol]                 # group-by fields (string or list)
  time_field: ts                # event-time field (epoch seconds); omit for processing time
  allowed_lateness: 0           # seconds a late record may still join its window
  aggregates:                   # at least one required
    - {name: trades, op: count}
    - {name: total, op: sum, field: qty}
  checkpoint:                   # optional durable state (survives restarts)
    type: file                  # file | sqlite (default file)
    path: /var/lib/pycharter/{stream_name}.json   # file: JSON path; sqlite: db path
    interval_seconds: 30        # persist at most this often (plus on shutdown)
    on_mismatch: discard        # discard | fail — see below

Aggregate op values: count, sum, min, max, mean, first, last, vwap (requires weight). Non-count ops require field.

The window operates on the validated record — fields referenced by key, time_field, or aggregates must be part of the contract's schema (validation output drops fields the schema doesn't declare), or records silently land in a keyless window.

window.checkpoint

Without a checkpoint, window state is in-memory and lost on restart. With one, open-window state is persisted at most every interval_seconds (triggered on the record path — state cannot change while idle) plus a final persist at shutdown, and restored at startup. {stream_name} in path is substituted; multiple streams may share one sqlite database (one row per stream). Run one worker per stream name — two workers sharing a checkpoint race.

Every checkpoint is stamped with a fingerprint of the window spec. If the window: config changed since the checkpoint was written, on_mismatch: discard (default) logs a warning and starts with empty state; fail refuses to start so an operator can decide.

Durability contract: with delivery: at_least_once + a checkpoint, window rows are emitted at-least-once. After a crash, replayed records may double-count into aggregates restored from a checkpoint that is up to interval_seconds stale — pair with idempotent sinks or tolerate approximate aggregates. exactly_once delivery rejects window entirely (state cannot participate in the Kafka transaction).

streams[].backpressure

Engine-wide tunables.

backpressure:
  max_in_flight: 1000                       # bound on the internal queue
  registry_outage_timeout_seconds: 60.0     # circuit breaker open-after threshold
  registry_retry:                           # in-loop retry for transient registry errors
    max_attempts: 3
    base_delay_seconds: 0.2
    max_delay_seconds: 5.0

Field	Type	Default	Notes
max_in_flight	int	1000	Bound on the queue between source and validator. Source pauses partition assignment when this saturates.
registry_outage_timeout_seconds	float	60.0	How long to tolerate consecutive registry failures before tripping the breaker. Once tripped, /readyz returns 503.
registry_retry.max_attempts	int	3	Total tries per record, including the first (1 disables retry). Covers both contract resolution and validator lookup: a store outage in either stage retries, then dead-letters the record as registry_unavailable and the stream keeps consuming. A store that answers "contract absent" dead-letters immediately as unknown_contract (no retry).
registry_retry.base_delay_seconds	float	0.2	Delay before the first retry; doubles per attempt with full jitter. Retries block the stream's consume loop (head-of-line), so worst-case added latency per record is bounded by the sum of delays (~0.2+0.4=0.6s at defaults).
registry_retry.max_delay_seconds	float	5.0	Ceiling for the exponential backoff.

streams[].quality

Windows for the in-process WindowedQualityReducer. Lower-cardinality than Prometheus histograms; useful for dashboards that show "violation rate over the last 5 minutes".

quality:
  windows: ["1m", "5m", "1h"]

Recognised window labels: 1m, 5m, 15m, 1h. Custom labels are rejected at config-load time.

observability

Worker-level settings shared by every stream.

observability:
  health_port: 8080
  metrics_port: 9090       # reserved; metrics currently surface on health_port
  log_format: json         # json | text

Field	Type	Default	Notes
health_port	int	8080	Port for /healthz, /readyz, /metrics.
metrics_port	int	9090	Reserved for split metrics-vs-health ports; unused in v1.
log_format	enum	json	json for production (one JSON object per record), text for local development.

Full example

format_version: "2"
id: example/orders
version: "1.0.0"

streams:
  - name: orders-validator
    source:
      type: kafka
      topic: orders
      bootstrap_servers: kafka.prod.svc.cluster.local:9092
      consumer_group: orders-validator
      auto_offset_reset: latest
      security_protocol: SASL_SSL
      sasl_mechanism: SCRAM-SHA-512
      sasl_username: ${KAFKA_USERNAME}
      sasl_password: ${KAFKA_PASSWORD}
      ssl_cafile: /etc/certs/ca.pem
    contract:
      ref: { type: store, name: orders, version: "2" }
      store: { type: postgres, connection_string: ${DATABASE_URL} }
      resolution: header
      on_missing_header: fallback_to_static
      validate: true
      coerce: true
    mode: enforce
    delivery: at_least_once
    load:
      target:
        type: kafka
        topic: orders-validated
        bootstrap_servers: kafka.prod.svc.cluster.local:9092
        security_protocol: SASL_SSL
        sasl_mechanism: SCRAM-SHA-512
        sasl_username: ${KAFKA_USERNAME}
        sasl_password: ${KAFKA_PASSWORD}
        ssl_cafile: /etc/certs/ca.pem
    dlq:
      target:
        type: kafka
        topic: orders-dlq
        bootstrap_servers: kafka.prod.svc.cluster.local:9092
        security_protocol: SASL_SSL
        sasl_mechanism: SCRAM-SHA-512
        sasl_username: ${KAFKA_USERNAME}
        sasl_password: ${KAFKA_PASSWORD}
        ssl_cafile: /etc/certs/ca.pem
    backpressure:
      max_in_flight: 5000
      registry_outage_timeout_seconds: 30
    quality:
      windows: ["1m", "5m", "1h"]
observability:
  health_port: 8080
  log_format: json