Egress, Batching and Delivery
An egress connector, or sink, writes the records a run publishes: to the console, Kafka, Redis, PostgreSQL, Parquet or JSON Lines files, or Iceberg tables. This page shows each sink, how to choose which records each one gets, and how records travel from a task to a sink.
What It Is
A run publishes three kinds of record: events, such as a task's queued, started and finished; telemetry, single values such as the statistics; and labels, the ground-truth records of what a scenario changed. Every sink receives every event and every telemetry record, unless its when says otherwise, and labels go only to sinks that ask for them. Records and Telemetry shows each record's shape and what each sink writes.
How to Use It
Each sink is attached with add_egress in the builder, passed to env.setup_egress in the low-level API, or listed under egress in a blueprint with its type and config. In the builder snippets below, app is a SimulationContext.
Console
ConsoleEgress logs each record at INFO level: [EVT] for an event, [TEL] for telemetry and [LBL] for a label.
Kafka
KafkaEgress sends events to event_topic and telemetry to telemetry_topic, keyed by the record's key or path_id, so all events of one entity land on one partition, in order. KafkaEgress creates both topics when the run starts, if they do not exist; with a topic_router it creates none. Complex Routing sends records to other topics.
app.add_egress(
KafkaEgress(
bootstrap_servers="localhost:9092",
event_topic="sim-events",
telemetry_topic="sim-telemetry",
)
)
Parquet and JSON Lines
ParquetStorageEgress and JsonlStorageEgress write each batch as a file of its own, on a local disk or S3-compatible storage, through PyArrow. Without a path_router, each event is one flat row in default_path, with its value mapping unpacked into columns, and telemetry is left out.
# Parquet files on the local disk
app.add_egress(ParquetStorageEgress(default_path="output/events"))
# JSON Lines files on S3-compatible storage, here SeaweedFS from odctl
seaweedfs = fs.S3FileSystem(
access_key="user",
secret_key="password",
endpoint_override="127.0.0.1:8333",
scheme="http",
)
app.add_egress(
JsonlStorageEgress(default_path="warehouse/events", filesystem=seaweedfs)
)
Redis
RedisEgress appends each record to a Redis Stream with XADD, as one payload field holding the record as JSON. A __stream__ key in the record's value picks the stream, and is removed before writing.
# XADD to the events stream, unless a record's value names another in __stream__
app.add_egress(
RedisEgress("redis://user:password@localhost:6379/0", stream_name="events")
)
PostgreSQL
PostgresEgress inserts records into a table in batches. Only records whose value is a mapping are written, and the labels on a labels sink. A label is written as sim_ts, timestamp, path, old, new and label. The record's sim_ts, timestamp, key (as event_id) and path_id are added to the value and replace fields of the same name. Its task and instance are added too, but a value field of the same name keeps its own value. Keys with no matching column are dropped. A __table__ key in the value sends the record only to the sink whose table_name matches.
dsn = "postgresql://user:password@localhost:5432/odctl"
app.add_egress(PostgresEgress(dsn, table_name="orders"))
By default a record whose key already exists is skipped. upsert_keys updates that row instead, for a model that changes rows it wrote earlier, such as an order moving from packed to shipped. The key columns need a unique index; an update sets only the columns the record carries, and within one flush the last record for a key wins.
app.add_egress(PostgresEgress(dsn, table_name="orders", upsert_keys=["order_id"]))
Iceberg
IcebergStorageEgress appends records to an Apache Iceberg table through a REST catalog. One flush is one commit, so this sink wants a large batch_size of its own: every commit writes new metadata, and many small commits slow down queries.
catalog = RestCatalog(
"odctl",
uri="http://localhost:8181",
warehouse="s3://warehouse/",
**{
"s3.endpoint": "http://localhost:8333",
"s3.access-key-id": "user",
"s3.secret-access-key": "password",
"s3.region": "us-east-1",
},
)
# One commit per flush, so this sink gets a large batch of its own
app.add_egress(
IcebergStorageEgress(catalog=catalog, default_table="sim.events"),
batch_size=200_000,
)
upsert_keys upserts on key columns instead of appending, so rerunning a seeded model over the same window does not duplicate rows. An upsert reads the matching rows first, replaces whole rows, never deletes, and needs pyiceberg 0.9.0 or later.
app.add_egress(
IcebergStorageEgress(
catalog=catalog,
default_table="sim.orders",
upsert_keys={"sim.orders": ["order_id"]},
),
batch_size=200_000,
)
Connectors in a Blueprint
In a blueprint, config holds the constructor's arguments, and settings that Python takes as objects are plain values: a Parquet or JSON Lines filesystem is a mapping with type: local or type: s3, an Iceberg catalog is a mapping of PyIceberg properties, and a Postgres tables mapping creates its tables when the run starts. YAML Format lists them all.
egress:
- type: Kafka
config:
bootstrap_servers: ${KAFKA_BOOTSTRAP_SERVERS:-localhost:9092}
event_topic: sim-events
telemetry_topic: sim-telemetry
batch_size: 50
flush_interval: 0.5
- type: Parquet
config:
default_path: warehouse/events.parquet
# A mapping becomes PyArrow's S3FileSystem; type: local is the local disk
filesystem:
type: s3
endpoint_override: 127.0.0.1:8333
scheme: http
access_key: user
secret_key: password
batch_size: 200000
- type: Jsonl
config:
path_router: !python routing_logic.daily_path
- type: Postgres
config:
connection_dsn: postgresql://user:password@localhost:5432/odctl
table_name: orders
upsert_keys: [order_id]
- type: Redis
config:
url: redis://user:password@localhost:6379/0
stream_name: events
Choose What Each Sink Gets
Every sink receives every event and telemetry record. when narrows that down:
- A function taking one record and returning True to send it to this sink, in the builder or as
!pythonin a blueprint. historyandlive, in a blueprint: the records stamped beforesimulation.go_live_at, and the rest, as Time and Pacing describes.labels: the ground-truth labels, and nothing else. No other sink receives a label, so data and labels always reach different sinks; Live Parameters describes them.
app.add_egress(stream).add_egress(lake) # both receive every record
# A record carries its timestamp as an ISO string, so compare with one
hot_from = datetime.now().isoformat(timespec="milliseconds")
app.add_egress(stream, when=lambda r: r["timestamp"] >= hot_from) # hot tail
app.add_egress(lake, when=lambda r: r["timestamp"] < hot_from) # cold history
app.add_egress(audit) # no predicate: everything
app.add_egress(stream) # events and telemetry, never a label
app.add_egress(audit, when="labels") # ground-truth labels, and nothing else
egress:
# The data under test: events and telemetry, and no label
- type: Console
# The labels only, for a scorer that checks what a detector found
- type: Jsonl
config: {default_path: labels/labels.jsonl}
when: labels
A run's warmup applies to every sink: none receives a record or telemetry stamped before it. Time and Pacing describes it.
Batch Size and Flush Interval
A sink receives records in batches. A batch is handed over when it holds batch_size records, or when flush_interval simulation seconds have passed, whichever comes first. At factor: 0 the timer is not started, because simulation time is not tied to the wall clock, until a go_live_at switches the run to real time.
| Setting | Default | Set with |
|---|---|---|
batch_size |
500 | with_batching, or per sink on add_egress |
flush_interval |
1.0 simulation seconds | with_batching, or per sink on add_egress |
max_queued_batches |
2000 | with_batching |
drain_stall_seconds |
30.0 seconds | with_batching |
with_batching sets the defaults, and each sink can have its own size and interval: a stream sink wants small, frequent batches to keep latency down, and a file or lake sink wants large ones, because the batch is the file, and for Iceberg the commit.
app.with_batching(
batch_size=5_000, # records per batch
flush_interval=1.0, # simulation seconds; not used at factor=0
max_queued_batches=2_000, # a full queue makes the simulation wait
drain_stall_seconds=30.0, # how long teardown waits for a queue that stopped
)
app.with_batching(batch_size=500, flush_interval=1.0) # the default for every sink
app.add_egress(stream, batch_size=50, flush_interval=0.5) # small and often
app.add_egress(lake, batch_size=200_000) # large files, the default interval
The same settings in a blueprint, and in the low-level API, where the lists are matched to the sinks by position and None means the default:
# The delivery settings of the Egress page, as a blueprint.
# Run it with: desflow run batching.yaml
simulation:
sim_id: Shop
factor: 0
# when: history and when: live split the records at this instant
logical_start_time: -1d
go_live_at: now
# Defaults for every egress, the arguments of with_batching
batching:
batch_size: 500
flush_interval: 1.0
max_queued_batches: 2000
drain_stall_seconds: 30
egress:
# Small and often for this sink; the other uses the defaults
- type: Console
batch_size: 50
flush_interval: 0.5
# Only the records stamped before go_live_at
- type: Console
when: history
resources:
packers: {current_cap: 1, max_cap: 1}
run:
until: 60
env = DynamicRealtimeEnvironment(factor=0.0)
env.setup_egress(
[stream, lake],
batch_size=500, # defaults for every provider
flush_interval=1.0,
predicates=[None, lambda r: r["stream_type"] == "event"], # by position
batch_sizes=[50, None], # None: the default
flush_intervals=[0.5, None],
labels=[False, False], # True makes a provider a labels sink
warmup=None, # simulation seconds before which no provider gets a record
)
setup_egress also takes lag_monitor_interval, the simulation seconds between system.simulation.lag_seconds records, 1.0 by default; None or 0 turns it off.
How It Works
A published record goes through the same steps for each sink:
publish_event,publish_telemetryorpublish_labelbuilds the record as a plain dictionary.- Each sink tests it once: a label only for a labels sink, and nothing else for one; nothing stamped before the warm-up; then its
when. - The record goes into that sink's buffer. When the buffer reaches
batch_size, or the sink's timer fires, the buffer is handed to the sink's queue as one batch. - The egress thread takes batches from each queue and writes them with the sink's connector.
Each sink has its own buffer and its own queue, so a slow sink does not hold up a fast one, and memory is the sum of the buffers. A queue holds at most max_queued_batches batches. When it is full, the simulation waits for room, so a sink that cannot keep up slows the run instead of building a backlog. When a queue stays full for drain_stall_seconds, the sink has stopped consuming, and the run raises RuntimeError rather than losing records.
Teardown flushes every buffer, then waits for the queues to drain. It keeps waiting while they shrink, and gives up only when they have not shrunk for drain_stall_seconds; records still queued then are lost, and a warning names the sinks that held them.
Stability and Limits
- A slow sink slows the run. That is on purpose: the alternative is a backlog that teardown would have to throw away.
- Size or interval, whichever comes first. The effective batch is the smaller of
batch_sizeand what arrives in oneflush_interval, so a large size with a short interval never fills; a warning says so once, after one sink has flushed on its timer 20 times in a row with less than a quarter of itsbatch_size. - A labels sink receives only labels, and a data sink never receives one. A label sent to a Kafka sink goes to its
event_topicwith no key. - Records matching no sink are not written anywhere, which is how
whendrops them. - At
factor: 0, records leave a buffer only when it is full and at teardown.