Data Sets

Seven data sets, each generated by one blueprint: a retail set written to three sinks, a film rental store, auction streams, a group-by table, taxi rides, labelled text and a point-in-time training set. Each entry is a complete file in three forms, so you can run it, read how it is built and change it.

---

Before You Start

The commands run from the root of a clone of the repository, where the files are in examples/gallery/:

git clone https://github.com/jaehyeon-kim/desflow.git
cd desflow

An entry that needs a service or an extra says so. Services for the Examples shows how to start the services. A file with a fixed random_seed and factor: 0 prints the same records on every run, and each output block below holds the first events of the YAML file. The Builder and Low-level files publish the same records.

---

Entries

Retail records go to PostgreSQL, Kafka and Parquet

A shop has 20 customers and 10 products at the start and takes about one order every two seconds. Each order picks a customer and a product at random and is followed by its invoice. The reference data goes to PostgreSQL tables, the orders to a Kafka topic and the invoices to a Parquet file, each sink taking the records of its own tasks (the when expression of the entry).

It needs a broker and a database, started with odctl up kafka-lite and odctl up postgres, and the kafka, postgres, parquet and fake extras. KAFKA_BOOTSTRAP_SERVERS sets the broker address and DEST_PATH the folder of the Parquet file (data by default), so this entry shows no output block here.

YAMLBuilderLow-level

Run it with the sample option, which stops after the first events and prints them:

desflow run examples/gallery/retail_records_to_three_sinks.yaml --sample 5

# Retail records go to PostgreSQL, Kafka and Parquet.
#
# A shop has 20 customers and 10 products at the start and takes about one order every
# two seconds. Each order picks a customer and a product at random and is followed by
# its invoice. The reference data goes to PostgreSQL tables, the orders to a Kafka
# topic and the invoices to a Parquet file, each sink taking the records of its own
# tasks (the `when` expression of the entry).
#
# Needs: a broker and a database, with odctl up kafka-lite and odctl up postgres, and
# the kafka, postgres, parquet and fake extras. KAFKA_BOOTSTRAP_SERVERS sets the broker
# address, and DEST_PATH the folder of the Parquet file (data by default).
# Run it with: desflow run examples/gallery/retail_records_to_three_sinks.yaml

simulation:
  sim_id: Retail
  factor: 0
  random_seed: 42
  logical_start_time: 2026-01-01T00:00:00

egress:
  - type: Postgres
    when: {expr: "record.task == 'customer'"}
    config:
      connection_dsn: postgresql://user:password@127.0.0.1:5432/odctl
      tables:
        customers:
          columns: {customer_id: INT, name: TEXT, country: TEXT}
          primary_key: customer_id
  - type: Postgres
    when: {expr: "record.task == 'product'"}
    config:
      connection_dsn: postgresql://user:password@127.0.0.1:5432/odctl
      tables:
        products:
          columns: {product_id: INT, title: TEXT, price: REAL}
          primary_key: product_id
  - type: Kafka
    when: {expr: "record.task == 'order'"}
    config:
      bootstrap_servers: ${KAFKA_BOOTSTRAP_SERVERS:-localhost:9092}
      event_topic: retail-orders
  - type: Parquet
    when: {expr: "record.task == 'invoice'"}
    config: {default_path: "${DEST_PATH:-data}/invoices.parquet"}

arrivals:
  customers: {initial: 20, spawn: customer}
  products: {initial: 10, spawn: product}
  orders: {dist: exponential, rate: 0.5, spawn: order}

tasks:
  customer:
    payload:
      customer_id: {gen: sequence, start: 1}
      name: {gen: fake, method: name}
      country:
        gen: weighted
        options: {AU: 5, US: 3, GB: 2}

  product:
    payload:
      product_id: {gen: sequence, start: 1}
      title:
        gen: one_of
        choices: [Desk lamp, Notebook, Water bottle, Backpack, Headphones, Keyboard]
      price: {gen: uniform, low: 2, high: 200, decimals: 2}

  order:
    vars:
      product: {gen: lookup, task: product}
      price: {gen: var, name: product, path: price}
      quantity: {gen: integer, bounds: [1, 5]}
    payload:
      order_id: {gen: sequence, start: 1}
      customer_id: {gen: lookup, task: customer, path: customer_id}
      product_id: {gen: var, name: product, path: product_id}
      quantity: {gen: var, name: quantity}
      total: {gen: expr, expr: "round(price * quantity, 2)"}
    next: invoice

  invoice:
    payload:
      order_id: {gen: upstream, path: order_id}
      amount: {gen: upstream, path: total}

run: {until: 10 min}

The Builder file builds the same model with the builder API and prints every record of its run:

python examples/gallery/retail_records_to_three_sinks_builder.py

"""Retail records go to PostgreSQL, Kafka and Parquet, builder API.

The builder version of `retail_records_to_three_sinks.yaml`: 20 customers and 10
products at the start, an order about every two seconds and an invoice for each order.
The customers and products go to PostgreSQL tables, the orders to a Kafka topic and the
invoices to a Parquet file.

Needs: a broker and a database, with odctl up kafka-lite and odctl up postgres, and the
kafka, postgres, parquet and fake extras. KAFKA_BOOTSTRAP_SERVERS sets the broker
address, and DEST_PATH the folder of the Parquet file (data by default).
Run it with: python examples/gallery/retail_records_to_three_sinks_builder.py
"""

import logging
import os
from datetime import datetime, timezone

from desflow import (
    KafkaEgress,
    ParquetStorageEgress,
    PostgresEgress,
    SimulationContext,
)
from desflow.values import Template

UNTIL = 600
DSN = "postgresql://user:password@127.0.0.1:5432/odctl"


def build() -> SimulationContext:
    """Builds the shop, with a sink for each kind of record."""
    app = SimulationContext(
        sim_id="Retail",
        factor=0,
        random_seed=42,
        logical_start_time=datetime(2026, 1, 1, tzinfo=timezone.utc),
    )
    app.add_egress(
        PostgresEgress(
            DSN,
            tables={
                "customers": {
                    "columns": {
                        "customer_id": "INT",
                        "name": "TEXT",
                        "country": "TEXT",
                    },
                    "primary_key": "customer_id",
                }
            },
        ),
        when=lambda record: record.get("task") == "customer",
    )
    app.add_egress(
        PostgresEgress(
            DSN,
            tables={
                "products": {
                    "columns": {"product_id": "INT", "title": "TEXT", "price": "REAL"},
                    "primary_key": "product_id",
                }
            },
        ),
        when=lambda record: record.get("task") == "product",
    )
    app.add_egress(
        KafkaEgress(
            bootstrap_servers=os.environ.get(
                "KAFKA_BOOTSTRAP_SERVERS", "localhost:9092"
            ),
            event_topic="retail-orders",
        ),
        when=lambda record: record.get("task") == "order",
    )
    app.add_egress(
        ParquetStorageEgress(
            default_path=f"{os.environ.get('DEST_PATH', 'data')}/invoices.parquet"
        ),
        when=lambda record: record.get("task") == "invoice",
    )
    app.add_initial_arrival("customers", 20, spawn="customer")
    app.add_initial_arrival("products", 10, spawn="product")
    app.add_arrival("orders", "exponential", rate=0.5, spawn="order")
    customer_payload = Template(
        {
            "customer_id": {"gen": "sequence", "start": 1},
            "name": {"gen": "fake", "method": "name"},
            "country": {"gen": "weighted", "options": {"AU": 5, "US": 3, "GB": 2}},
        },
        name="tasks.customer",
    )
    product_payload = Template(
        {
            "product_id": {"gen": "sequence", "start": 1},
            "title": {
                "gen": "one_of",
                "choices": [
                    "Desk lamp",
                    "Notebook",
                    "Water bottle",
                    "Backpack",
                    "Headphones",
                    "Keyboard",
                ],
            },
            "price": {"gen": "uniform", "low": 2, "high": 200, "decimals": 2},
        },
        name="tasks.product",
    )
    order_payload = Template(
        {
            "order_id": {"gen": "sequence", "start": 1},
            "customer_id": {"gen": "lookup", "task": "customer", "path": "customer_id"},
            "product_id": {"gen": "var", "name": "product", "path": "product_id"},
            "quantity": {"gen": "var", "name": "quantity"},
            "total": {"gen": "expr", "expr": "round(price * quantity, 2)"},
        },
        name="tasks.order",
        vars={
            "product": {"gen": "lookup", "task": "product"},
            "price": {"gen": "var", "name": "product", "path": "price"},
            "quantity": {"gen": "integer", "bounds": [1, 5]},
        },
    )
    invoice_payload = Template(
        {
            "order_id": {"gen": "upstream", "path": "order_id"},
            "amount": {"gen": "upstream", "path": "total"},
        },
        name="tasks.invoice",
    )

    @app.task(name="customer", history=1_000_000)
    def customer(task_id):
        """Gives one customer."""
        return customer_payload

    @app.task(name="product", history=1_000_000)
    def product(task_id):
        """Gives one product."""
        return product_payload

    @app.task(name="order", next="invoice")
    def order(task_id):
        """Takes one order."""
        return order_payload

    @app.task(name="invoice")
    def invoice(task_id):
        """Bills the order."""
        return invoice_payload

    return app


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
    build().run(until=UNTIL)

The Low-level file builds the model on DynamicRealtimeEnvironment, with plain Python processes, and prints every record of its run:

python examples/gallery/retail_records_to_three_sinks_low_level.py

The Low-level file draws each value with Sampler.generator(<place>), where the place is the key path of the blueprint's generator that makes the same value, such as tasks.customer.payload.name, tasks.customer.payload.country and tasks.product.payload.title and 4 more. Each draw therefore comes from the random sequence the blueprint uses at the same place, so the draws are the blueprint's. The gaps between arrivals come from Sampler.sample on a DistributionConfig whose stream is arrival.orders, the sequence the blueprint uses for its arrivals.

"""Retail records go to PostgreSQL, Kafka and Parquet, low-level API.

The low-level version of `retail_records_to_three_sinks.yaml`: plain SimPy on a
`DynamicRealtimeEnvironment`, which publishes each record with `publish_event`. Each
draw is made with `Sampler.generator(<name>)`, from the random sequence the blueprint's
generator at that place uses, so the draws are the blueprint's. A name is drawn with a
Faker that is seeded from that sequence.

Needs: a broker and a database, with odctl up kafka-lite and odctl up postgres, and the
kafka, postgres, parquet and fake extras. KAFKA_BOOTSTRAP_SERVERS sets the broker
address, and DEST_PATH the folder of the Parquet file (data by default).
Run it with: python examples/gallery/retail_records_to_three_sinks_low_level.py
"""

import itertools
import logging
import os
from datetime import datetime, timezone

from faker import Faker

from desflow import (
    DistributionConfig,
    DynamicRealtimeEnvironment,
    KafkaEgress,
    ParquetStorageEgress,
    PostgresEgress,
    Sampler,
)

UNTIL = 600
START = datetime(2026, 1, 1, tzinfo=timezone.utc)
ENVIRONMENT = {"logical_start_time": START, "timezone": "UTC"}
DSN = "postgresql://user:password@127.0.0.1:5432/odctl"
COUNTRIES = {"AU": 5, "US": 3, "GB": 2}
TITLES = ["Desk lamp", "Notebook", "Water bottle", "Backpack", "Headphones", "Keyboard"]


def weighted(rng, options):
    """One key of `options`, chosen by its weight, as the `weighted` generator does."""
    point = rng.random() * sum(options.values())
    running = 0.0
    for value, weight in options.items():
        running += weight
        if point < running:
            return value
    return value


def build(env: DynamicRealtimeEnvironment) -> None:
    """Starts the arrivals on `env`."""
    sampler = Sampler(seed=42)
    faker = Faker("en_US")
    faker.seed_instance(
        int(sampler.generator("tasks.customer.payload.name#fake").integers(2**32))
    )
    country = sampler.generator("tasks.customer.payload.country")
    title = sampler.generator("tasks.product.payload.title")
    price = sampler.generator("tasks.product.payload.price")
    product_row = sampler.generator("tasks.order.vars.product")
    quantity = sampler.generator("tasks.order.vars.quantity")
    customer_row = sampler.generator("tasks.order.payload.customer_id")
    customers: list = []  # the ids of the customers so far, oldest first
    products: list = []  # the products so far, oldest first
    counts = {"customer": 0, "product": 0, "order": 0, "invoice": 0}

    def publish(key, task, payload):
        """Publishes a finished payload as the next instance of its task."""
        env.publish_event(key, payload, task=task, instance=counts[task])
        counts[task] += 1

    def customer():
        row = {
            "customer_id": len(customers) + 1,
            "name": faker.name(),
            "country": weighted(country, COUNTRIES),
        }
        customers.append(row["customer_id"])
        return row

    def product():
        row = {
            "product_id": len(products) + 1,
            "title": TITLES[int(title.integers(len(TITLES)))],
            "price": round(2 + (200 - 2) * price.random(), 2),
        }
        products.append(row)
        return row

    def order(number):
        """An order, and the invoice for it at once."""
        key = f"orders-{number}"
        chosen = products[int(product_row.integers(len(products)))]
        count = int(quantity.integers(1, 6))
        taken = {
            "order_id": number,
            "customer_id": customers[int(customer_row.integers(len(customers)))],
            "product_id": chosen["product_id"],
            "quantity": count,
            "total": round(chosen["price"] * count, 2),
        }
        publish(key, "order", taken)
        publish(key, "invoice", {"order_id": number, "amount": taken["total"]})

    gap = DistributionConfig(dist="exponential", rate=0.5)
    gap.stream = "arrival.orders"

    def source():
        for number in itertools.count(1):
            yield env.timeout(sampler.sample(gap))
            order(number)

    def start(name, task, make, number):
        yield env.timeout(0)
        publish(f"{name}-{number}", task, make())

    env.process(source())
    for number in range(1, 21):
        env.process(start("customers", "customer", customer, number))
    for number in range(1, 11):
        env.process(start("products", "product", product, number))


def run() -> None:
    """Runs the model unpaced, with a sink for each kind of record."""
    env = DynamicRealtimeEnvironment(factor=0, **ENVIRONMENT)
    bootstrap = os.environ.get("KAFKA_BOOTSTRAP_SERVERS", "localhost:9092")
    sinks = [
        (
            PostgresEgress(
                DSN,
                tables={
                    "customers": {
                        "columns": {
                            "customer_id": "INT",
                            "name": "TEXT",
                            "country": "TEXT",
                        },
                        "primary_key": "customer_id",
                    }
                },
            ),
            "customer",
        ),
        (
            PostgresEgress(
                DSN,
                tables={
                    "products": {
                        "columns": {
                            "product_id": "INT",
                            "title": "TEXT",
                            "price": "REAL",
                        },
                        "primary_key": "product_id",
                    }
                },
            ),
            "product",
        ),
        (
            KafkaEgress(bootstrap_servers=bootstrap, event_topic="retail-orders"),
            "order",
        ),
        (
            ParquetStorageEgress(
                default_path=f"{os.environ.get('DEST_PATH', 'data')}/invoices.parquet"
            ),
            "invoice",
        ),
    ]
    env.setup_egress(
        [sink for sink, _ in sinks],
        predicates=[
            lambda record, task=task: record.get("task") == task for _, task in sinks
        ],
    )
    build(env)
    try:
        env.run(until=UNTIL)
    finally:
        env.teardown()


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
    run()

A film rental store: films, customers, rentals, payments and returns

The tables have the shape of the Sakila sample database. A store has 20 films and 15 customers at the start. About one rental starts every 8 minutes, for a week. A rental names a film and a customer that exist, is paid for at once at the film's rental rate, and is returned 1 to 7 days later.

It needs the fake extra: pip install "desflow[fake]".

YAMLBuilderLow-level

Run it with the sample option, which stops after the first events and prints them:

desflow run examples/gallery/film_rental_store.yaml --sample 5

# A film rental store: films, customers, rentals, payments and returns.
#
# The tables have the shape of the Sakila sample database. A store has 20 films and 15
# customers at the start. About one rental starts every 8 minutes, for a week. A rental
# names a film and a customer that exist, is paid for at once at the film's rental rate,
# and is returned 1 to 7 days later.
#
# Needs: the fake extra.
# Run it with: desflow run examples/gallery/film_rental_store.yaml

simulation:
  sim_id: Store
  factor: 0
  random_seed: 42
  logical_start_time: 2026-01-01T00:00:00

egress:
  - {type: Console}

arrivals:
  films: {initial: 20, spawn: film}
  customers: {initial: 15, spawn: customer}
  rentals: {dist: exponential, rate: 0.002, spawn: rental}

tasks:
  film:
    payload:
      film_id: {gen: sequence, start: 1}
      title:
        gen: format
        template: "{adjective} {noun}"
        names:
          adjective: {gen: one_of, choices: [Silent, Broken, Golden, Hidden, Last]}
          noun: {gen: one_of, choices: [River, Garden, Highway, Letter, Harbour]}
      rating:
        gen: weighted
        options: {G: 2, PG: 3, PG-13: 3, R: 3, NC-17: 1}
      rental_rate: {gen: one_of, choices: [0.99, 2.99, 4.99]}
      length: {gen: integer, bounds: [60, 180]}

  customer:
    payload:
      customer_id: {gen: sequence, start: 1}
      first_name: {gen: fake, method: first_name}
      last_name: {gen: fake, method: last_name}
      active: {gen: boolean, p: 0.9}

  rental:
    vars:
      film: {gen: lookup, task: film}
    payload:
      rental_id: {gen: sequence, start: 1}
      rental_date: {gen: now}
      film_id: {gen: var, name: film, path: film_id}
      rental_rate: {gen: var, name: film, path: rental_rate}
      customer_id: {gen: lookup, task: customer, path: customer_id}
    next: payment

  payment:
    payload:
      rental_id: {gen: upstream, path: rental_id}
      customer_id: {gen: upstream, path: customer_id}
      amount: {gen: upstream, path: rental_rate}
      payment_date: {gen: now}
    next: return

  return:
    # 1 to 7 days, in seconds
    wait:
      timeout: {gen: uniform, low: 86400, high: 604800}
    payload:
      rental_id: {gen: upstream, path: rental_id}
      return_date: {gen: now}

run: {until: 7 days}

{"stream_type":"event","sim_ts":0.0,"timestamp":"...","key":"films-1","task":"film","instance":0,"value":{"film_id":1,"title":"Last Letter","rating":"NC-17","rental_rate":2.99,"length":141}}
{"stream_type":"event","sim_ts":0.0,"timestamp":"...","key":"films-2","task":"film","instance":1,"value":{"film_id":2,"title":"Golden Letter","rating":"NC-17","rental_rate":4.99,"length":93}}
{"stream_type":"event","sim_ts":0.0,"timestamp":"...","key":"films-3","task":"film","instance":2,"value":{"film_id":3,"title":"Last Highway","rating":"PG","rental_rate":0.99,"length":96}}
{"stream_type":"event","sim_ts":0.0,"timestamp":"...","key":"films-4","task":"film","instance":3,"value":{"film_id":4,"title":"Broken River","rating":"PG","rental_rate":2.99,"length":105}}
{"stream_type":"event","sim_ts":0.0,"timestamp":"...","key":"films-5","task":"film","instance":4,"value":{"film_id":5,"title":"Hidden Harbour","rating":"PG","rental_rate":4.99,"length":115}}

The Builder file builds the same model with the builder API and prints every record of its run:

python examples/gallery/film_rental_store_builder.py

"""A film rental store: films, customers, rentals, payments and returns, builder API.

The builder version of `film_rental_store.yaml`: 20 films and 15 customers at the start,
a rental about every 8 minutes for a week, paid at once and returned 1 to 7 days later.

Needs: the fake extra.
Run it with: python examples/gallery/film_rental_store_builder.py
"""

import logging
from datetime import datetime, timezone

from desflow import ConsoleEgress, SimulationContext
from desflow.values import Template, generator

UNTIL = 7 * 86400


def build() -> SimulationContext:
    """Builds the store."""
    app = SimulationContext(
        sim_id="Store",
        factor=0,
        random_seed=42,
        logical_start_time=datetime(2026, 1, 1, tzinfo=timezone.utc),
    )
    app.add_egress(ConsoleEgress())
    app.add_initial_arrival("films", 20, spawn="film")
    app.add_initial_arrival("customers", 15, spawn="customer")
    app.add_arrival("rentals", "exponential", rate=0.002, spawn="rental")
    film_payload = Template(
        {
            "film_id": {"gen": "sequence", "start": 1},
            "title": {
                "gen": "format",
                "template": "{adjective} {noun}",
                "names": {
                    "adjective": {
                        "gen": "one_of",
                        "choices": ["Silent", "Broken", "Golden", "Hidden", "Last"],
                    },
                    "noun": {
                        "gen": "one_of",
                        "choices": ["River", "Garden", "Highway", "Letter", "Harbour"],
                    },
                },
            },
            "rating": {
                "gen": "weighted",
                "options": {"G": 2, "PG": 3, "PG-13": 3, "R": 3, "NC-17": 1},
            },
            "rental_rate": {"gen": "one_of", "choices": [0.99, 2.99, 4.99]},
            "length": {"gen": "integer", "bounds": [60, 180]},
        },
        name="tasks.film",
    )
    customer_payload = Template(
        {
            "customer_id": {"gen": "sequence", "start": 1},
            "first_name": {"gen": "fake", "method": "first_name"},
            "last_name": {"gen": "fake", "method": "last_name"},
            "active": {"gen": "boolean", "p": 0.9},
        },
        name="tasks.customer",
    )
    rental_payload = Template(
        {
            "rental_id": {"gen": "sequence", "start": 1},
            "rental_date": {"gen": "now"},
            "film_id": {"gen": "var", "name": "film", "path": "film_id"},
            "rental_rate": {"gen": "var", "name": "film", "path": "rental_rate"},
            "customer_id": {"gen": "lookup", "task": "customer", "path": "customer_id"},
        },
        name="tasks.rental",
        vars={"film": {"gen": "lookup", "task": "film"}},
    )
    payment_payload = Template(
        {
            "rental_id": {"gen": "upstream", "path": "rental_id"},
            "customer_id": {"gen": "upstream", "path": "customer_id"},
            "amount": {"gen": "upstream", "path": "rental_rate"},
            "payment_date": {"gen": "now"},
        },
        name="tasks.payment",
    )
    return_payload = Template(
        {
            "rental_id": {"gen": "upstream", "path": "rental_id"},
            "return_date": {"gen": "now"},
        },
        name="tasks.return",
    )

    @app.task(name="film", history=1_000_000)
    def film(task_id):
        """Gives one film."""
        return film_payload

    @app.task(name="customer", history=1_000_000)
    def customer(task_id):
        """Gives one customer."""
        return customer_payload

    @app.task(name="rental", next="payment")
    def rental(task_id):
        """Starts a rental of a film by a customer."""
        return rental_payload

    @app.task(name="payment", next="return")
    def payment(task_id):
        """Pays for the rental at its rate."""
        return payment_payload

    # 1 to 7 days, in seconds
    @app.task(
        name="return",
        wait={"timeout": generator({"gen": "uniform", "low": 86400, "high": 604800})},
    )
    def return_(task_id):
        """Returns the film."""
        return return_payload

    return app


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
    build().run(until=UNTIL)

The Low-level file builds the model on DynamicRealtimeEnvironment, with plain Python processes, and prints every record of its run:

python examples/gallery/film_rental_store_low_level.py

The Low-level file draws each value with Sampler.generator(<place>), where the place is the key path of the blueprint's generator that makes the same value, such as tasks.film.payload.title.names.adjective, tasks.film.payload.title.names.noun and tasks.film.payload.rating and 6 more. Each draw therefore comes from the random sequence the blueprint uses at the same place, so the draws are the blueprint's. The gaps between arrivals come from Sampler.sample on a DistributionConfig whose stream is arrival.rentals, the sequence the blueprint uses for its arrivals.

"""A film rental store: films, customers, rentals, payments and returns, low-level API.

The low-level version of `film_rental_store.yaml`: plain SimPy on a `DynamicRealtimeEnvironment`,
which publishes each record with `publish_event`. Each draw is made with
`Sampler.generator(<name>)`, from the random sequence the blueprint's generator at that
place uses, so the draws are the blueprint's. A name is drawn with a Faker that is
seeded from that sequence.

Needs: the fake extra.
Run it with: python examples/gallery/film_rental_store_low_level.py
"""

import itertools
import logging
from datetime import datetime, timedelta, timezone

from faker import Faker

from desflow import (
    ConsoleEgress,
    DistributionConfig,
    DynamicRealtimeEnvironment,
    Sampler,
)

UNTIL = 7 * 86400
START = datetime(2026, 1, 1, tzinfo=timezone.utc)
ENVIRONMENT = {"logical_start_time": START, "timezone": "UTC"}
ADJECTIVES = ["Silent", "Broken", "Golden", "Hidden", "Last"]
NOUNS = ["River", "Garden", "Highway", "Letter", "Harbour"]
RATINGS = {"G": 2, "PG": 3, "PG-13": 3, "R": 3, "NC-17": 1}
RATES = [0.99, 2.99, 4.99]


def weighted(rng, options):
    """One key of `options`, chosen by its weight, as the `weighted` generator does."""
    point = rng.random() * sum(options.values())
    running = 0.0
    for value, weight in options.items():
        running += weight
        if point < running:
            return value
    return value


def one_of(rng, choices):
    """One of `choices`, each equally likely, as the `one_of` generator does."""
    return choices[int(rng.integers(len(choices)))]


def build(env: DynamicRealtimeEnvironment) -> None:
    """Starts the arrivals on `env`."""
    sampler = Sampler(seed=42)

    def faker_of(place):
        faker = Faker("en_US")
        faker.seed_instance(int(sampler.generator(f"{place}#fake").integers(2**32)))
        return faker

    first_names = faker_of("tasks.customer.payload.first_name")
    last_names = faker_of("tasks.customer.payload.last_name")
    adjective = sampler.generator("tasks.film.payload.title.names.adjective")
    noun = sampler.generator("tasks.film.payload.title.names.noun")
    rating = sampler.generator("tasks.film.payload.rating")
    rate = sampler.generator("tasks.film.payload.rental_rate")
    length = sampler.generator("tasks.film.payload.length")
    active = sampler.generator("tasks.customer.payload.active")
    film_row = sampler.generator("tasks.rental.vars.film")
    customer_row = sampler.generator("tasks.rental.payload.customer_id")
    stay = sampler.generator("tasks.return.wait.timeout")
    films: list = []  # the films so far, oldest first
    customers: list = []  # the ids of the customers so far, oldest first
    counts = {"film": 0, "customer": 0, "rental": 0, "payment": 0, "return": 0}

    def iso():
        """The logical time now, as a timestamp text."""
        moment = START + timedelta(seconds=env.now)
        return moment.isoformat(timespec="milliseconds")

    def publish(key, task, payload, instance=None):
        """Publishes a finished payload, as the next instance of its task unless the
        task has queued one."""
        if instance is None:
            instance = counts[task]
            counts[task] += 1
        env.publish_event(key, payload, task=task, instance=instance)

    def film():
        row = {
            "film_id": len(films) + 1,
            "title": f"{one_of(adjective, ADJECTIVES)} {one_of(noun, NOUNS)}",
            "rating": weighted(rating, RATINGS),
            "rental_rate": one_of(rate, RATES),
            "length": int(length.integers(60, 181)),
        }
        films.append(row)
        return row

    def customer():
        row = {
            "customer_id": len(customers) + 1,
            "first_name": first_names.first_name(),
            "last_name": last_names.last_name(),
            "active": bool(active.random() < 0.9),
        }
        customers.append(row["customer_id"])
        return row

    def rental(number):
        """A rental, its payment at once, and its return after the stay."""
        key = f"rentals-{number}"
        chosen = films[int(film_row.integers(len(films)))]
        rented = {
            "rental_id": number,
            "rental_date": iso(),
            "film_id": chosen["film_id"],
            "rental_rate": chosen["rental_rate"],
            "customer_id": customers[int(customer_row.integers(len(customers)))],
        }
        publish(key, "rental", rented)
        paid = {
            "rental_id": rented["rental_id"],
            "customer_id": rented["customer_id"],
            "amount": rented["rental_rate"],
            "payment_date": iso(),
        }
        publish(key, "payment", paid)
        instance = counts["return"]  # the return waits, so it is queued first
        counts["return"] += 1
        env.publish_event(key, {"status": "queued"}, task="return", instance=instance)
        yield env.timeout(86400 + (604800 - 86400) * stay.random())  # 1 to 7 days
        env.publish_event(key, {"status": "started"}, task="return", instance=instance)
        returned = {"rental_id": rented["rental_id"], "return_date": iso()}
        publish(key, "return", returned, instance)

    gap = DistributionConfig(dist="exponential", rate=0.002)
    gap.stream = "arrival.rentals"

    def source():
        for number in itertools.count(1):
            yield env.timeout(sampler.sample(gap))
            env.process(rental(number))

    def start(name, task, make, number):
        yield env.timeout(0)
        publish(f"{name}-{number}", task, make())

    env.process(source())
    for number in range(1, 21):
        env.process(start("films", "film", film, number))
    for number in range(1, 16):
        env.process(start("customers", "customer", customer, number))


def run() -> None:
    """Runs the model unpaced into the console."""
    env = DynamicRealtimeEnvironment(factor=0, **ENVIRONMENT)
    env.setup_egress([ConsoleEgress()])
    build(env)
    try:
        env.run(until=UNTIL)
    finally:
        env.teardown()


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
    run()

Auction streams: people, auctions and bids

The three streams have the shape of the Nexmark benchmark, with about 1 new person for 3 new auctions and 46 bids. Each auction names a seller who exists and each bid names an auction and a bidder who exist. Ten people and five auctions exist at the start, so a bid always has something to name. The run lasts 10 minutes.

It needs the fake extra: pip install "desflow[fake]".

YAMLBuilderLow-level

Run it with the sample option, which stops after the first events and prints them:

desflow run examples/gallery/auction_streams.yaml --sample 5

# Auction streams: people, auctions and bids.
#
# The three streams have the shape of the Nexmark benchmark, with about 1 new person
# for 3 new auctions and 46 bids. Each auction names a seller who exists and each bid
# names an auction and a bidder who exist. Ten people and five auctions exist at the
# start, so a bid always has something to name. The run lasts 10 minutes.
#
# Needs: the fake extra.
# Run it with: desflow run examples/gallery/auction_streams.yaml

simulation:
  sim_id: Auction
  factor: 0
  random_seed: 42
  logical_start_time: 2026-01-01T00:00:00

egress:
  - {type: Console}

arrivals:
  people: {initial: 10, dist: exponential, rate: 0.02, spawn: person}
  auctions: {initial: 5, dist: exponential, rate: 0.06, spawn: auction}
  bids: {dist: exponential, rate: 0.92, spawn: bid}

tasks:
  person:
    payload:
      person_id: {gen: sequence, start: 1000}
      name: {gen: fake, method: name}
      city: {gen: one_of, choices: [Phoenix, Boise, Cheyenne, Portland, Kent]}
      state:
        gen: weighted
        options: {AZ: 3, ID: 2, WY: 1, OR: 3, WA: 3}

  auction:
    payload:
      auction_id: {gen: sequence, start: 1}
      seller: {gen: lookup, task: person, path: person_id}
      category: {gen: integer, bounds: [1, 10]}
      reserve: {gen: integer, bounds: [100, 1000]}
      expires: {gen: now, offset: 1 hour}

  bid:
    payload:
      auction: {gen: lookup, task: auction, path: auction_id}
      bidder: {gen: lookup, task: person, path: person_id}
      price: {gen: integer, bounds: [100, 2000]}
      bid_time: {gen: now}

run: {until: 10 min}

{"stream_type":"event","sim_ts":0.0,"timestamp":"...","key":"people-1","task":"person","instance":0,"value":{"person_id":1000,"name":"Lisa Allen","city":"Phoenix","state":"AZ"}}
{"stream_type":"event","sim_ts":0.0,"timestamp":"...","key":"people-2","task":"person","instance":1,"value":{"person_id":1001,"name":"David Cunningham","city":"Boise","state":"AZ"}}
{"stream_type":"event","sim_ts":0.0,"timestamp":"...","key":"people-3","task":"person","instance":2,"value":{"person_id":1002,"name":"Sara Spence","city":"Kent","state":"ID"}}
{"stream_type":"event","sim_ts":0.0,"timestamp":"...","key":"people-4","task":"person","instance":3,"value":{"person_id":1003,"name":"Amber Bean","city":"Boise","state":"OR"}}
{"stream_type":"event","sim_ts":0.0,"timestamp":"...","key":"people-5","task":"person","instance":4,"value":{"person_id":1004,"name":"Michele Savage","city":"Kent","state":"WA"}}

The Builder file builds the same model with the builder API and prints every record of its run:

python examples/gallery/auction_streams_builder.py

"""Auction streams: people, auctions and bids, builder API.

The builder version of `auction_streams.yaml`: about 1 new person for 3 new auctions and
46 bids, with ten people and five auctions at the start so that every lookup finds a
row.

Needs: the fake extra.
Run it with: python examples/gallery/auction_streams_builder.py
"""

import logging
from datetime import datetime, timezone

from desflow import ConsoleEgress, SimulationContext
from desflow.values import Template

UNTIL = 600


def build() -> SimulationContext:
    """Builds the auction house."""
    app = SimulationContext(
        sim_id="Auction",
        factor=0,
        random_seed=42,
        logical_start_time=datetime(2026, 1, 1, tzinfo=timezone.utc),
    )
    app.add_egress(ConsoleEgress())
    app.add_arrival("people", "exponential", rate=0.02, spawn="person", initial=10)
    app.add_arrival("auctions", "exponential", rate=0.06, spawn="auction", initial=5)
    app.add_arrival("bids", "exponential", rate=0.92, spawn="bid")
    person_payload = Template(
        {
            "person_id": {"gen": "sequence", "start": 1000},
            "name": {"gen": "fake", "method": "name"},
            "city": {
                "gen": "one_of",
                "choices": ["Phoenix", "Boise", "Cheyenne", "Portland", "Kent"],
            },
            "state": {
                "gen": "weighted",
                "options": {"AZ": 3, "ID": 2, "WY": 1, "OR": 3, "WA": 3},
            },
        },
        name="tasks.person",
    )
    auction_payload = Template(
        {
            "auction_id": {"gen": "sequence", "start": 1},
            "seller": {"gen": "lookup", "task": "person", "path": "person_id"},
            "category": {"gen": "integer", "bounds": [1, 10]},
            "reserve": {"gen": "integer", "bounds": [100, 1000]},
            "expires": {"gen": "now", "offset": "1 hour"},
        },
        name="tasks.auction",
    )
    bid_payload = Template(
        {
            "auction": {"gen": "lookup", "task": "auction", "path": "auction_id"},
            "bidder": {"gen": "lookup", "task": "person", "path": "person_id"},
            "price": {"gen": "integer", "bounds": [100, 2000]},
            "bid_time": {"gen": "now"},
        },
        name="tasks.bid",
    )

    # A task that something looks up keeps its finished rows
    @app.task(name="person", history=1_000_000)
    def person(task_id):
        """Gives one person."""
        return person_payload

    @app.task(name="auction", history=1_000_000)
    def auction(task_id):
        """Gives one auction."""
        return auction_payload

    @app.task(name="bid")
    def bid(task_id):
        """Gives one bid."""
        return bid_payload

    return app


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
    build().run(until=UNTIL)

The Low-level file builds the model on DynamicRealtimeEnvironment, with plain Python processes, and prints every record of its run:

python examples/gallery/auction_streams_low_level.py

The Low-level file draws each value with Sampler.generator(<place>), where the place is the key path of the blueprint's generator that makes the same value, such as tasks.person.payload.name, tasks.person.payload.city and tasks.person.payload.state and 6 more. Each draw therefore comes from the random sequence the blueprint uses at the same place, so the draws are the blueprint's. The gaps between arrivals come from Sampler.sample on a DistributionConfig whose stream is arrival.<name>, the sequence the blueprint uses for its arrivals.

"""Auction streams: people, auctions and bids, low-level API.

The low-level version of `auction_streams.yaml`: plain SimPy on a `DynamicRealtimeEnvironment`,
which publishes each record with `publish_event`. Each draw is made with
`Sampler.generator(<name>)`, from the random sequence the blueprint's generator at that
place uses, so the draws are the blueprint's. A name is drawn with a Faker that is
seeded from that sequence.

Needs: the fake extra.
Run it with: python examples/gallery/auction_streams_low_level.py
"""

import itertools
import logging
from datetime import datetime, timedelta, timezone

from faker import Faker

from desflow import (
    ConsoleEgress,
    DistributionConfig,
    DynamicRealtimeEnvironment,
    Sampler,
)

UNTIL = 600
START = datetime(2026, 1, 1, tzinfo=timezone.utc)
ENVIRONMENT = {"logical_start_time": START, "timezone": "UTC"}
CITIES = ["Phoenix", "Boise", "Cheyenne", "Portland", "Kent"]
STATES = {"AZ": 3, "ID": 2, "WY": 1, "OR": 3, "WA": 3}


def weighted(rng, options):
    """One key of `options`, chosen by its weight, as the `weighted` generator does."""
    point = rng.random() * sum(options.values())
    running = 0.0
    for value, weight in options.items():
        running += weight
        if point < running:
            return value
    return value


def integer(rng, low, high):
    """A whole number from `low` to `high`, both included, as `integer` draws it."""
    return int(rng.integers(low, high + 1))


def build(env: DynamicRealtimeEnvironment) -> None:
    """Starts the arrivals on `env`."""
    sampler = Sampler(seed=42)
    faker = Faker("en_US")
    faker.seed_instance(
        int(sampler.generator("tasks.person.payload.name#fake").integers(2**32))
    )
    city = sampler.generator("tasks.person.payload.city")
    state = sampler.generator("tasks.person.payload.state")
    seller = sampler.generator("tasks.auction.payload.seller")
    category = sampler.generator("tasks.auction.payload.category")
    reserve = sampler.generator("tasks.auction.payload.reserve")
    auction = sampler.generator("tasks.bid.payload.auction")
    bidder = sampler.generator("tasks.bid.payload.bidder")
    price = sampler.generator("tasks.bid.payload.price")
    people: list = []  # the ids of the people so far, oldest first
    auctions: list = []  # the ids of the auctions so far, oldest first
    counts = {"person": 0, "auction": 0, "bid": 0}

    def iso(offset=0):
        """The logical time now plus `offset` seconds, as a timestamp text."""
        moment = START + timedelta(seconds=env.now + offset)
        return moment.isoformat(timespec="milliseconds")

    def person():
        counts["person"] += 1
        row = {
            "person_id": 999 + counts["person"],
            "name": faker.name(),
            "city": CITIES[int(city.integers(len(CITIES)))],
            "state": weighted(state, STATES),
        }
        people.append(row["person_id"])
        return row

    def new_auction():
        counts["auction"] += 1
        row = {
            "auction_id": counts["auction"],
            "seller": people[int(seller.integers(len(people)))],
            "category": integer(category, 1, 10),
            "reserve": integer(reserve, 100, 1000),
            "expires": iso(3600),
        }
        auctions.append(row["auction_id"])
        return row

    def bid():
        counts["bid"] += 1
        return {
            "auction": auctions[int(auction.integers(len(auctions)))],
            "bidder": people[int(bidder.integers(len(people)))],
            "price": integer(price, 100, 2000),
            "bid_time": iso(),
        }

    def arrival(name, task, make, rate, initial=0):
        """Starts `initial` entities at time 0, then one after each gap."""
        gap = DistributionConfig(dist="exponential", rate=rate)
        gap.stream = f"arrival.{name}"

        def start(number):
            key = f"{name}-{number}"
            payload = make()
            env.publish_event(key, payload, task=task, instance=counts[task] - 1)

        def source():
            for number in itertools.count(initial + 1):
                yield env.timeout(sampler.sample(gap))
                start(number)

        def first(number):
            yield env.timeout(0)
            start(number)

        env.process(source())
        for number in range(1, initial + 1):
            env.process(first(number))

    arrival("people", "person", person, 0.02, initial=10)
    arrival("auctions", "auction", new_auction, 0.06, initial=5)
    arrival("bids", "bid", bid, 0.92)


def run() -> None:
    """Runs the model unpaced into the console."""
    env = DynamicRealtimeEnvironment(factor=0, **ENVIRONMENT)
    env.setup_egress([ConsoleEgress()])
    build(env)
    try:
        env.run(until=UNTIL)
    finally:
        env.teardown()


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
    run()

A table for group-by benchmarks

Each row has three keys that are text (id1 to id3, with 100, 100 and 100,000 values), three that are whole numbers (id4 to id6) and three measures (v1, v2 and v3). A group-by on a key and a sum of a measure is the usual test of a data frame library. One row a second, 1,000 rows.

YAMLBuilderLow-level

Run it with the sample option, which stops after the first events and prints them:

desflow run examples/gallery/group_by_table.yaml --sample 5

# A table for group-by benchmarks.
#
# Each row has three keys that are text (id1 to id3, with 100, 100 and 100,000 values),
# three that are whole numbers (id4 to id6) and three measures (v1, v2 and v3). A
# group-by on a key and a sum of a measure is the usual test of a data frame library.
# One row a second, 1,000 rows.
#
# Needs: nothing
# Run it with: desflow run examples/gallery/group_by_table.yaml

simulation:
  sim_id: Table
  factor: 0
  random_seed: 42
  logical_start_time: 2026-01-01T00:00:00

egress:
  - {type: Console}

arrivals:
  rows: {dist: constant, value: 1, spawn: row, max_events: 1000}

tasks:
  row:
    payload:
      id1:
        gen: format
        template: "id{n:03d}"
        names:
          n: {gen: integer, bounds: [1, 100]}
      id2:
        gen: format
        template: "id{n:03d}"
        names:
          n: {gen: integer, bounds: [1, 100]}
      id3:
        gen: format
        template: "id{n:010d}"
        names:
          n: {gen: integer, bounds: [1, 100000]}
      id4: {gen: integer, bounds: [1, 100]}
      id5: {gen: integer, bounds: [1, 100]}
      id6: {gen: integer, bounds: [1, 100000]}
      v1: {gen: integer, bounds: [1, 5]}
      v2: {gen: integer, bounds: [1, 15]}
      v3: {gen: uniform, low: 0, high: 100, decimals: 6}

run:
  # The 1,000th row is at 1,000 s, and a run stops before the events due at its end
  # time, so it ends 1 s later.
  until: 1001

{"stream_type":"event","sim_ts":1.0,"timestamp":"...","key":"rows-1","task":"row","instance":0,"value":{"id1":"id077","id2":"id045","id3":"id0000064280","id4":50,"id5":66,"id6":60719,"v1":2,"v2":14,"v3":21.170454}}
{"stream_type":"event","sim_ts":2.0,"timestamp":"...","key":"rows-2","task":"row","instance":1,"value":{"id1":"id089","id2":"id065","id3":"id0000003738","id4":7,"id5":29,"id6":23702,"v1":1,"v2":10,"v3":59.492801}}
{"stream_type":"event","sim_ts":3.0,"timestamp":"...","key":"rows-3","task":"row","instance":2,"value":{"id1":"id076","id2":"id066","id3":"id0000095153","id4":75,"id5":86,"id6":49573,"v1":1,"v2":6,"v3":95.16441}}
{"stream_type":"event","sim_ts":4.0,"timestamp":"...","key":"rows-4","task":"row","instance":3,"value":{"id1":"id021","id2":"id045","id3":"id0000023013","id4":10,"id5":39,"id6":45451,"v1":5,"v2":14,"v3":47.143878}}
{"stream_type":"event","sim_ts":5.0,"timestamp":"...","key":"rows-5","task":"row","instance":4,"value":{"id1":"id046","id2":"id097","id3":"id0000095822","id4":82,"id5":15,"id6":23992,"v1":1,"v2":1,"v3":4.205366}}

The Builder file builds the same model with the builder API and prints every record of its run:

python examples/gallery/group_by_table_builder.py

"""A table for group-by benchmarks, builder API.

The builder version of `group_by_table.yaml`: 1,000 rows, one a second, each with the
keys id1 to id6 and the measures v1 to v3.

Needs: nothing
Run it with: python examples/gallery/group_by_table_builder.py
"""

import logging
from datetime import datetime, timezone

from desflow import ConsoleEgress, SimulationContext
from desflow.values import Template

# The 1,000th row is at 1,000 s, and a run stops before the events due at its end
# time, so it ends 1 s later.
UNTIL = 1001


def build() -> SimulationContext:
    """Builds the table."""
    app = SimulationContext(
        sim_id="Table",
        factor=0,
        random_seed=42,
        logical_start_time=datetime(2026, 1, 1, tzinfo=timezone.utc),
    )
    app.add_egress(ConsoleEgress())
    app.add_arrival("rows", "constant", value=1, spawn="row", max_events=1000)
    row_payload = Template(
        {
            "id1": {
                "gen": "format",
                "template": "id{n:03d}",
                "names": {"n": {"gen": "integer", "bounds": [1, 100]}},
            },
            "id2": {
                "gen": "format",
                "template": "id{n:03d}",
                "names": {"n": {"gen": "integer", "bounds": [1, 100]}},
            },
            "id3": {
                "gen": "format",
                "template": "id{n:010d}",
                "names": {"n": {"gen": "integer", "bounds": [1, 100000]}},
            },
            "id4": {"gen": "integer", "bounds": [1, 100]},
            "id5": {"gen": "integer", "bounds": [1, 100]},
            "id6": {"gen": "integer", "bounds": [1, 100000]},
            "v1": {"gen": "integer", "bounds": [1, 5]},
            "v2": {"gen": "integer", "bounds": [1, 15]},
            "v3": {"gen": "uniform", "low": 0, "high": 100, "decimals": 6},
        },
        name="tasks.row",
    )

    @app.task(name="row")
    def row(task_id):
        """Gives one row."""
        return row_payload

    return app


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
    build().run(until=UNTIL)

The Low-level file builds the model on DynamicRealtimeEnvironment, with plain Python processes, and prints every record of its run:

python examples/gallery/group_by_table_low_level.py

The Low-level file draws each value with Sampler.generator(<place>), where the place is the key path of the blueprint's generator that makes the same value, such as {place}.id1.names.n with the place of the task filled in. Each draw therefore comes from the random sequence the blueprint uses at the same place, so the draws are the blueprint's. The file waits 1 simulation second between rows, as the blueprint's constant arrival does.

"""A table for group-by benchmarks, low-level API.

The low-level version of `group_by_table.yaml`: plain SimPy on a `DynamicRealtimeEnvironment`,
which publishes each record with `publish_event`. Each draw is made with
`Sampler.generator(<name>)`, from the random sequence the blueprint's generator at that
place uses, so the draws are the blueprint's.

Needs: nothing
Run it with: python examples/gallery/group_by_table_low_level.py
"""

import logging
from datetime import datetime, timezone

from desflow import ConsoleEgress, DynamicRealtimeEnvironment, Sampler

# The 1,000th row is at 1,000 s, and a run stops before the events due at its end
# time, so it ends 1 s later.
UNTIL = 1001
START = datetime(2026, 1, 1, tzinfo=timezone.utc)
ENVIRONMENT = {"logical_start_time": START, "timezone": "UTC"}


def integer(rng, low, high):
    """A whole number from `low` to `high`, both included, as `integer` draws it."""
    return int(rng.integers(low, high + 1))


def build(env: DynamicRealtimeEnvironment) -> None:
    """Starts the arrivals on `env`."""
    sampler = Sampler(seed=42)
    place = "tasks.row.payload"
    id1 = sampler.generator(f"{place}.id1.names.n")
    id2 = sampler.generator(f"{place}.id2.names.n")
    id3 = sampler.generator(f"{place}.id3.names.n")
    id4 = sampler.generator(f"{place}.id4")
    id5 = sampler.generator(f"{place}.id5")
    id6 = sampler.generator(f"{place}.id6")
    v1 = sampler.generator(f"{place}.v1")
    v2 = sampler.generator(f"{place}.v2")
    v3 = sampler.generator(f"{place}.v3")

    def source():
        for number in range(1, 1001):
            yield env.timeout(1)
            row = {
                "id1": f"id{integer(id1, 1, 100):03d}",
                "id2": f"id{integer(id2, 1, 100):03d}",
                "id3": f"id{integer(id3, 1, 100000):010d}",
                "id4": integer(id4, 1, 100),
                "id5": integer(id5, 1, 100),
                "id6": integer(id6, 1, 100000),
                "v1": integer(v1, 1, 5),
                "v2": integer(v2, 1, 15),
                "v3": round(100 * v3.random(), 6),
            }
            env.publish_event(f"rows-{number}", row, task="row", instance=number - 1)

    env.process(source())


def run() -> None:
    """Runs the model unpaced into the console."""
    env = DynamicRealtimeEnvironment(factor=0, **ENVIRONMENT)
    env.setup_egress([ConsoleEgress()])
    build(env)
    try:
        env.run(until=UNTIL)
    finally:
        env.teardown()


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
    run()

Taxi rides with a pickup, a drop-off, a distance and a fare

A ride starts about every 10 seconds. Its distance is 0.5 to 20 km, its duration is 3 to 40 minutes, and the fare is 3 plus 2.2 for each km and 0.4 for each minute. The pickup is in a box about 10 km across. Each ride is one record, written at the pickup with the drop-off time that follows from the duration.

YAMLBuilderLow-level

Run it with the sample option, which stops after the first events and prints them:

desflow run examples/gallery/taxi_rides.yaml --sample 5

# Taxi rides with a pickup, a drop-off, a distance and a fare.
#
# A ride starts about every 10 seconds. Its distance is 0.5 to 20 km, its duration is
# 3 to 40 minutes, and the fare is 3 plus 2.2 for each km and 0.4 for each minute. The
# pickup is in a box about 10 km across. Each ride is one record, written at the pickup
# with the drop-off time that follows from the duration.
#
# Needs: nothing
# Run it with: desflow run examples/gallery/taxi_rides.yaml

simulation:
  sim_id: Taxi
  factor: 0
  random_seed: 42
  logical_start_time: 2026-01-01T00:00:00

egress:
  - {type: Console}

arrivals:
  rides: {dist: exponential, rate: 0.1, spawn: ride}

tasks:
  ride:
    vars:
      distance: {gen: uniform, low: 0.5, high: 20, decimals: 2}
      minutes: {gen: uniform, low: 3, high: 40, decimals: 1}
    payload:
      ride_id: {gen: sequence, start: 1}
      pickup_time: {gen: now}
      dropoff_time:
        gen: add
        args:
          - {gen: now}
          - {gen: expr, expr: "minutes * 60"}
      passengers:
        gen: weighted
        options: {1: 70, 2: 15, 3: 5, 4: 5, 5: 5}
      pickup_latitude: {gen: uniform, low: 40.7, high: 40.8, decimals: 5}
      pickup_longitude: {gen: uniform, low: -74.02, high: -73.93, decimals: 5}
      distance_km: {gen: var, name: distance}
      fare: {gen: expr, expr: "round(3 + 2.2 * distance + 0.4 * minutes, 2)"}
      payment:
        gen: weighted
        options: {card: 65, cash: 33, dispute: 2}

run: {until: 10 min}

{"stream_type":"event","sim_ts":5.350962446908744,"timestamp":"...","key":"rides-1","task":"ride","instance":0,"value":{"ride_id":1,"pickup_time":"2026-01-01T00:00:05.350+00:00","dropoff_time":"2026-01-01T00:29:53.350+00:00","passengers":5,"pickup_latitude":40.76959,"pickup_longitude":-73.97817,"distance_km":15.93,"fare":49.97,"payment":"dispute"}}
{"stream_type":"event","sim_ts":12.782361457212403,"timestamp":"...","key":"rides-2","task":"ride","instance":1,"value":{"ride_id":2,"pickup_time":"2026-01-01T00:00:12.782+00:00","dropoff_time":"2026-01-01T00:24:24.782+00:00","passengers":1,"pickup_latitude":40.79611,"pickup_longitude":-73.95995,"distance_km":16.95,"fare":49.97,"payment":"cash"}}
{"stream_type":"event","sim_ts":21.60201181850169,"timestamp":"...","key":"rides-3","task":"ride","instance":2,"value":{"ride_id":3,"pickup_time":"2026-01-01T00:00:21.602+00:00","dropoff_time":"2026-01-01T00:38:57.602+00:00","passengers":2,"pickup_latitude":40.78903,"pickup_longitude":-74.0196,"distance_km":10.45,"fare":41.43,"payment":"cash"}}
{"stream_type":"event","sim_ts":37.29261778662655,"timestamp":"...","key":"rides-4","task":"ride","instance":3,"value":{"ride_id":4,"pickup_time":"2026-01-01T00:00:37.292+00:00","dropoff_time":"2026-01-01T00:23:31.292+00:00","passengers":1,"pickup_latitude":40.76259,"pickup_longitude":-73.99615,"distance_km":8.99,"fare":31.94,"payment":"card"}}
{"stream_type":"event","sim_ts":39.600079308854646,"timestamp":"...","key":"rides-5","task":"ride","instance":4,"value":{"ride_id":5,"pickup_time":"2026-01-01T00:00:39.600+00:00","dropoff_time":"2026-01-01T00:28:27.600+00:00","passengers":1,"pickup_latitude":40.70961,"pickup_longitude":-73.97145,"distance_km":16.28,"fare":49.94,"payment":"card"}}

The Builder file builds the same model with the builder API and prints every record of its run:

python examples/gallery/taxi_rides_builder.py

"""Taxi rides with a pickup, a drop-off, a distance and a fare, builder API.

The builder version of `taxi_rides.yaml`: a ride about every 10 seconds, for 10 minutes.

Needs: nothing
Run it with: python examples/gallery/taxi_rides_builder.py
"""

import logging
from datetime import datetime, timezone

from desflow import ConsoleEgress, SimulationContext
from desflow.values import Template

UNTIL = 600


def build() -> SimulationContext:
    """Builds the taxi company."""
    app = SimulationContext(
        sim_id="Taxi",
        factor=0,
        random_seed=42,
        logical_start_time=datetime(2026, 1, 1, tzinfo=timezone.utc),
    )
    app.add_egress(ConsoleEgress())
    app.add_arrival("rides", "exponential", rate=0.1, spawn="ride")
    ride_payload = Template(
        {
            "ride_id": {"gen": "sequence", "start": 1},
            "pickup_time": {"gen": "now"},
            "dropoff_time": {
                "gen": "add",
                "args": [{"gen": "now"}, {"gen": "expr", "expr": "minutes * 60"}],
            },
            "passengers": {
                "gen": "weighted",
                "options": {1: 70, 2: 15, 3: 5, 4: 5, 5: 5},
            },
            "pickup_latitude": {
                "gen": "uniform",
                "low": 40.7,
                "high": 40.8,
                "decimals": 5,
            },
            "pickup_longitude": {
                "gen": "uniform",
                "low": -74.02,
                "high": -73.93,
                "decimals": 5,
            },
            "distance_km": {"gen": "var", "name": "distance"},
            "fare": {
                "gen": "expr",
                "expr": "round(3 + 2.2 * distance + 0.4 * minutes, 2)",
            },
            "payment": {
                "gen": "weighted",
                "options": {"card": 65, "cash": 33, "dispute": 2},
            },
        },
        name="tasks.ride",
        vars={
            "distance": {"gen": "uniform", "low": 0.5, "high": 20, "decimals": 2},
            "minutes": {"gen": "uniform", "low": 3, "high": 40, "decimals": 1},
        },
    )

    @app.task(name="ride")
    def ride(task_id):
        """Gives one ride."""
        return ride_payload

    return app


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
    build().run(until=UNTIL)

The Low-level file builds the model on DynamicRealtimeEnvironment, with plain Python processes, and prints every record of its run:

python examples/gallery/taxi_rides_low_level.py

The Low-level file draws each value with Sampler.generator(<place>), where the place is the key path of the blueprint's generator that makes the same value, such as {place}.payload.passengers with the place of the task filled in. Each draw therefore comes from the random sequence the blueprint uses at the same place, so the draws are the blueprint's. The gaps between arrivals come from Sampler.sample on a DistributionConfig whose stream is arrival.rides, the sequence the blueprint uses for its arrivals.

"""Taxi rides with a pickup, a drop-off, a distance and a fare, low-level API.

The low-level version of `taxi_rides.yaml`: plain SimPy on a `DynamicRealtimeEnvironment`,
which publishes each record with `publish_event`. Each draw is made with
`Sampler.generator(<name>)`, from the random sequence the blueprint's generator at that
place uses, so the draws are the blueprint's.

Needs: nothing
Run it with: python examples/gallery/taxi_rides_low_level.py
"""

import itertools
import logging
from datetime import datetime, timedelta, timezone

from desflow import (
    ConsoleEgress,
    DistributionConfig,
    DynamicRealtimeEnvironment,
    Sampler,
)

UNTIL = 600
START = datetime(2026, 1, 1, tzinfo=timezone.utc)
ENVIRONMENT = {"logical_start_time": START, "timezone": "UTC"}
PASSENGERS = {1: 70, 2: 15, 3: 5, 4: 5, 5: 5}
PAYMENT = {"card": 65, "cash": 33, "dispute": 2}


def weighted(rng, options):
    """One key of `options`, chosen by its weight, as the `weighted` generator does."""
    point = rng.random() * sum(options.values())
    running = 0.0
    for value, weight in options.items():
        running += weight
        if point < running:
            return value
    return value


def uniform(rng, low, high, decimals):
    """A number in [low, high), as the `uniform` generator draws it."""
    return round(low + (high - low) * rng.random(), decimals)


def build(env: DynamicRealtimeEnvironment) -> None:
    """Starts the arrivals on `env`."""
    sampler = Sampler(seed=42)
    place = "tasks.ride"
    passengers = sampler.generator(f"{place}.payload.passengers")
    latitude = sampler.generator(f"{place}.payload.pickup_latitude")
    longitude = sampler.generator(f"{place}.payload.pickup_longitude")
    payment = sampler.generator(f"{place}.payload.payment")
    distance = sampler.generator(f"{place}.vars.distance")
    minutes = sampler.generator(f"{place}.vars.minutes")
    gap = DistributionConfig(dist="exponential", rate=0.1)
    gap.stream = "arrival.rides"

    def source():
        for number in itertools.count(1):
            yield env.timeout(sampler.sample(gap))
            km = uniform(distance, 0.5, 20, 2)
            length = uniform(minutes, 3, 40, 1)
            pickup = START + timedelta(seconds=env.now)
            dropoff = datetime.fromisoformat(
                pickup.isoformat(timespec="milliseconds")
            ) + timedelta(seconds=length * 60)
            ride = {
                "ride_id": number,
                "pickup_time": pickup.isoformat(timespec="milliseconds"),
                "dropoff_time": dropoff.isoformat(timespec="milliseconds"),
                "passengers": weighted(passengers, PASSENGERS),
                "pickup_latitude": uniform(latitude, 40.7, 40.8, 5),
                "pickup_longitude": uniform(longitude, -74.02, -73.93, 5),
                "distance_km": km,
                "fare": round(3 + 2.2 * km + 0.4 * length, 2),
                "payment": weighted(payment, PAYMENT),
            }
            env.publish_event(f"rides-{number}", ride, task="ride", instance=number - 1)

    env.process(source())


def run() -> None:
    """Runs the model unpaced into the console."""
    env = DynamicRealtimeEnvironment(factor=0, **ENVIRONMENT)
    env.setup_egress([ConsoleEgress()])
    build(env)
    try:
        env.run(until=UNTIL)
    finally:
        env.teardown()


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
    run()

Messages and reviews with a known intent or sentiment

Each message has an intent and each review a sentiment, drawn first by weight. The text is then written from one of the templates for that label, so the label is known for every record, and can be used to check a classifier, a retrieval system or a language model against what it should find.

YAMLBuilderLow-level

Run it with the sample option, which stops after the first events and prints them:

desflow run examples/gallery/labelled_text.yaml --sample 5

# Messages and reviews with a known intent or sentiment.
#
# Each message has an intent and each review a sentiment, drawn first by weight. The
# text is then written from one of the templates for that label, so the label is known
# for every record, and can be used to check a classifier, a retrieval system or a
# language model against what it should find.
#
# Needs: nothing
# Run it with: desflow run examples/gallery/labelled_text.yaml

simulation:
  sim_id: Text
  factor: 0
  random_seed: 42
  logical_start_time: 2026-01-01T00:00:00

egress:
  - {type: Console}

arrivals:
  messages: {dist: exponential, rate: 0.5, spawn: message}
  reviews: {dist: exponential, rate: 0.2, spawn: review}

tasks:
  message:
    vars:
      intent:
        gen: weighted
        options: {order_status: 4, refund: 2, cancel: 1, hello: 3}
    payload:
      message_id: {gen: sequence, start: 1}
      intent: {gen: var, name: intent}
      text:
        gen: switch
        on: {gen: var, name: intent}
        cases:
          order_status:
            gen: weighted
            choices:
              - value:
                  gen: format
                  template: "Where is my order {n}?"
                  names:
                    n: {gen: integer, bounds: [1000, 9999]}
                weight: 2
              - {value: "Has my parcel been sent yet?", weight: 1}
          refund:
            gen: weighted
            choices:
              - {value: "I want my money back for this order.", weight: 2}
              - value:
                  gen: format
                  template: "Please refund order {n}."
                  names:
                    n: {gen: integer, bounds: [1000, 9999]}
                weight: 1
          cancel:
            gen: weighted
            choices:
              - {value: "Cancel my order, I changed my mind.", weight: 1}
              - {value: "Please stop my subscription.", weight: 1}
          hello:
            gen: weighted
            choices:
              - {value: "Hi, I have a question.", weight: 1}
              - {value: "Hello, is anyone there?", weight: 1}

  review:
    vars:
      sentiment:
        gen: weighted
        options: {positive: 6, neutral: 2, negative: 2}
      product: {gen: one_of, choices: [lamp, notebook, backpack, headphones]}
    payload:
      review_id: {gen: sequence, start: 1}
      sentiment: {gen: var, name: sentiment}
      text:
        gen: switch
        on: {gen: var, name: sentiment}
        cases:
          positive:
            gen: weighted
            choices:
              - value:
                  gen: format
                  template: "The {product} is great and arrived early."
                weight: 1
              - value:
                  gen: format
                  template: "I love this {product}, I would buy it again."
                weight: 1
          neutral: {gen: format, template: "The {product} is fine, nothing special."}
          negative:
            gen: weighted
            choices:
              - value: {gen: format, template: "The {product} broke after a week."}
                weight: 1
              - value: {gen: format, template: "I would not recommend this {product}."}
                weight: 1

run: {until: 10 min}

{"stream_type":"event","sim_ts":5.954583282580391,"timestamp":"...","key":"messages-1","task":"message","instance":0,"value":{"message_id":1,"intent":"cancel","text":"Cancel my order, I changed my mind."}}
{"stream_type":"event","sim_ts":10.226505406484272,"timestamp":"...","key":"messages-2","task":"message","instance":1,"value":{"message_id":2,"intent":"hello","text":"Hi, I have a question."}}
{"stream_type":"event","sim_ts":10.668409519913007,"timestamp":"...","key":"messages-3","task":"message","instance":2,"value":{"message_id":3,"intent":"order_status","text":"Where is my order 5723?"}}
{"stream_type":"event","sim_ts":11.60968254261269,"timestamp":"...","key":"messages-4","task":"message","instance":3,"value":{"message_id":4,"intent":"order_status","text":"Where is my order 8265?"}}
{"stream_type":"event","sim_ts":12.326845051488377,"timestamp":"...","key":"reviews-1","task":"review","instance":0,"value":{"review_id":1,"sentiment":"positive","text":"The notebook is great and arrived early."}}

The Builder file builds the same model with the builder API and prints every record of its run:

python examples/gallery/labelled_text_builder.py

"""Messages and reviews with a known intent or sentiment, builder API.

The builder version of `labelled_text.yaml`: a label is drawn first by weight, and the
text is written from a template for that label.

Needs: nothing
Run it with: python examples/gallery/labelled_text_builder.py
"""

import logging
from datetime import datetime, timezone

from desflow import ConsoleEgress, SimulationContext
from desflow.values import Template

UNTIL = 600


def _number(low: int, high: int) -> dict:
    return {"gen": "integer", "bounds": [low, high]}


def build() -> SimulationContext:
    """Builds the support desk and the shop."""
    app = SimulationContext(
        sim_id="Text",
        factor=0,
        random_seed=42,
        logical_start_time=datetime(2026, 1, 1, tzinfo=timezone.utc),
    )
    app.add_egress(ConsoleEgress())
    app.add_arrival("messages", "exponential", rate=0.5, spawn="message")
    app.add_arrival("reviews", "exponential", rate=0.2, spawn="review")

    def says(template: str, **names: dict) -> dict:
        return {"gen": "format", "template": template, "names": names}

    message_payload = Template(
        {
            "message_id": {"gen": "sequence", "start": 1},
            "intent": {"gen": "var", "name": "intent"},
            "text": {
                "gen": "switch",
                "on": {"gen": "var", "name": "intent"},
                "cases": {
                    "order_status": {
                        "gen": "weighted",
                        "choices": [
                            {
                                "value": says(
                                    "Where is my order {n}?", n=_number(1000, 9999)
                                ),
                                "weight": 2,
                            },
                            {"value": "Has my parcel been sent yet?", "weight": 1},
                        ],
                    },
                    "refund": {
                        "gen": "weighted",
                        "choices": [
                            {
                                "value": "I want my money back for this order.",
                                "weight": 2,
                            },
                            {
                                "value": says(
                                    "Please refund order {n}.", n=_number(1000, 9999)
                                ),
                                "weight": 1,
                            },
                        ],
                    },
                    "cancel": {
                        "gen": "weighted",
                        "choices": [
                            {
                                "value": "Cancel my order, I changed my mind.",
                                "weight": 1,
                            },
                            {"value": "Please stop my subscription.", "weight": 1},
                        ],
                    },
                    "hello": {
                        "gen": "weighted",
                        "choices": [
                            {"value": "Hi, I have a question.", "weight": 1},
                            {"value": "Hello, is anyone there?", "weight": 1},
                        ],
                    },
                },
            },
        },
        name="tasks.message",
        vars={
            "intent": {
                "gen": "weighted",
                "options": {"order_status": 4, "refund": 2, "cancel": 1, "hello": 3},
            }
        },
    )
    review_payload = Template(
        {
            "review_id": {"gen": "sequence", "start": 1},
            "sentiment": {"gen": "var", "name": "sentiment"},
            "text": {
                "gen": "switch",
                "on": {"gen": "var", "name": "sentiment"},
                "cases": {
                    "positive": {
                        "gen": "weighted",
                        "choices": [
                            {
                                "value": says(
                                    "The {product} is great and arrived early."
                                ),
                                "weight": 1,
                            },
                            {
                                "value": says(
                                    "I love this {product}, I would buy it again."
                                ),
                                "weight": 1,
                            },
                        ],
                    },
                    "neutral": says("The {product} is fine, nothing special."),
                    "negative": {
                        "gen": "weighted",
                        "choices": [
                            {
                                "value": says("The {product} broke after a week."),
                                "weight": 1,
                            },
                            {
                                "value": says("I would not recommend this {product}."),
                                "weight": 1,
                            },
                        ],
                    },
                },
            },
        },
        name="tasks.review",
        vars={
            "sentiment": {
                "gen": "weighted",
                "options": {"positive": 6, "neutral": 2, "negative": 2},
            },
            "product": {
                "gen": "one_of",
                "choices": ["lamp", "notebook", "backpack", "headphones"],
            },
        },
    )

    @app.task(name="message")
    def message(task_id):
        """Gives one message with its intent."""
        return message_payload

    @app.task(name="review")
    def review(task_id):
        """Gives one review with its sentiment."""
        return review_payload

    return app


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
    build().run(until=UNTIL)

The Low-level file builds the model on DynamicRealtimeEnvironment, with plain Python processes, and prints every record of its run:

python examples/gallery/labelled_text_low_level.py

The Low-level file draws each value with Sampler.generator(<place>), where the place is the key path of the blueprint's generator that makes the same value, such as tasks.message.vars.intent, tasks.review.vars.sentiment and tasks.review.vars.product. Each draw therefore comes from the random sequence the blueprint uses at the same place, so the draws are the blueprint's. The gaps between arrivals come from Sampler.sample on a DistributionConfig whose stream is arrival.<name>, the sequence the blueprint uses for its arrivals.

"""Messages and reviews with a known intent or sentiment, low-level API.

The low-level version of `labelled_text.yaml`: plain SimPy on a `DynamicRealtimeEnvironment`,
which publishes each record with `publish_event`. Each draw is made with
`Sampler.generator(<name>)`, from the random sequence the blueprint's generator at that
place uses, so the draws are the blueprint's.

Needs: nothing
Run it with: python examples/gallery/labelled_text_low_level.py
"""

import itertools
import logging
from datetime import datetime, timezone

from desflow import (
    ConsoleEgress,
    DistributionConfig,
    DynamicRealtimeEnvironment,
    Sampler,
)

UNTIL = 600
START = datetime(2026, 1, 1, tzinfo=timezone.utc)
ENVIRONMENT = {"logical_start_time": START, "timezone": "UTC"}
INTENTS = {"order_status": 4, "refund": 2, "cancel": 1, "hello": 3}
SENTIMENTS = {"positive": 6, "neutral": 2, "negative": 2}
PRODUCTS = ["lamp", "notebook", "backpack", "headphones"]
# The texts of a message, by intent. A text with {n} takes an order number
MESSAGES = {
    "order_status": [
        ("Where is my order {n}?", 2),
        ("Has my parcel been sent yet?", 1),
    ],
    "refund": [
        ("I want my money back for this order.", 2),
        ("Please refund order {n}.", 1),
    ],
    "cancel": [
        ("Cancel my order, I changed my mind.", 1),
        ("Please stop my subscription.", 1),
    ],
    "hello": [("Hi, I have a question.", 1), ("Hello, is anyone there?", 1)],
}
# The texts of a review, by sentiment. A text takes the product
REVIEWS = {
    "positive": [
        ("The {product} is great and arrived early.", 1),
        ("I love this {product}, I would buy it again.", 1),
    ],
    "neutral": [("The {product} is fine, nothing special.", 1)],
    "negative": [
        ("The {product} broke after a week.", 1),
        ("I would not recommend this {product}.", 1),
    ],
}


def position(rng, weights):
    """The position of one weight, chosen by its size, as the `weighted` generator does."""
    point = rng.random() * sum(weights)
    running = 0.0
    for index, weight in enumerate(weights):
        running += weight
        if point < running:
            return index
    return len(weights) - 1


def build(env: DynamicRealtimeEnvironment) -> None:
    """Starts the arrivals on `env`."""
    sampler = Sampler(seed=42)
    intent = sampler.generator("tasks.message.vars.intent")
    sentiment = sampler.generator("tasks.review.vars.sentiment")
    product = sampler.generator("tasks.review.vars.product")
    counts = {"message": 0, "review": 0}

    def message_text(label):
        """The text of a message: a template chosen by weight, with an order number."""
        place = f"tasks.message.payload.text.cases.{label}"
        templates = MESSAGES[label]
        chosen = position(sampler.generator(place), [w for _, w in templates])
        template = templates[chosen][0]
        number = sampler.generator(f"{place}.choices.{chosen}.value.names.n")
        return template.format(n=int(number.integers(1000, 10000)))

    def review_text(label, name):
        """The text of a review: a template chosen by weight, for the product."""
        templates = REVIEWS[label]
        weights = [w for _, w in templates]
        place = f"tasks.review.payload.text.cases.{label}"
        chosen = 0 if len(weights) == 1 else position(sampler.generator(place), weights)
        return templates[chosen][0].format(product=name)

    def message():
        counts["message"] += 1
        label = list(INTENTS)[position(intent, list(INTENTS.values()))]
        return {
            "message_id": counts["message"],
            "intent": label,
            "text": message_text(label),
        }

    def review():
        counts["review"] += 1
        label = list(SENTIMENTS)[position(sentiment, list(SENTIMENTS.values()))]
        name = PRODUCTS[int(product.integers(len(PRODUCTS)))]
        return {
            "review_id": counts["review"],
            "sentiment": label,
            "text": review_text(label, name),
        }

    def arrival(name, task, make, rate):
        gap = DistributionConfig(dist="exponential", rate=rate)
        gap.stream = f"arrival.{name}"

        def source():
            for number in itertools.count(1):
                yield env.timeout(sampler.sample(gap))
                env.publish_event(
                    f"{name}-{number}", make(), task=task, instance=number - 1
                )

        env.process(source())

    arrival("messages", "message", message, 0.5)
    arrival("reviews", "review", review, 0.2)


def run() -> None:
    """Runs the model unpaced into the console."""
    env = DynamicRealtimeEnvironment(factor=0, **ENVIRONMENT)
    env.setup_egress([ConsoleEgress()])
    build(env)
    try:
        env.run(until=UNTIL)
    finally:
        env.teardown()


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
    run()

A point-in-time training set: feature records and the label that follows each

Five customers each get a feature record every day (sessions and spend over the last 30 days). Seven days after each one a label record says whether the customer left, with a high chance when the sessions were few. The label carries as_of, the time of the feature record it follows. To build the training set, join each label to the newest feature record of the same customer whose sim_ts is at or before the label's sim_ts less 7 days, so no feature that came after the start of the label's window reaches it.

YAMLBuilderLow-level

Run it with the sample option, which stops after the first events and prints them:

desflow run examples/gallery/point_in_time_training_set.yaml --sample 5

# A point-in-time training set: feature records and the label that follows each.
#
# Five customers each get a feature record every day (sessions and spend over the last
# 30 days). Seven days after each one a label record says whether the customer left,
# with a high chance when the sessions were few. The label carries `as_of`, the time of
# the feature record it follows. To build the training set, join each label to the
# newest feature record of the same customer whose `sim_ts` is at or before the label's
# `sim_ts` less 7 days, so no feature that came after the start of the label's window
# reaches it.
#
# Needs: nothing
# Run it with: desflow run examples/gallery/point_in_time_training_set.yaml

simulation:
  sim_id: Churn
  factor: 0
  random_seed: 42
  logical_start_time: 2026-01-01T00:00:00

egress:
  - {type: Console}

arrivals:
  customers:
    count: 5
    key: {gen: sequence, format: "customer-{}", start: 1}
    dist: constant
    value: 1 day
    spawn: features

tasks:
  features:
    payload:
      customer_id: {gen: entity}
      as_of: {gen: now}
      sessions_30d: {gen: integer, bounds: [0, 30]}
      spend_30d: {gen: uniform, low: 0, high: 500, decimals: 2}
    next: label

  label:
    wait: {timeout: 7 days}
    payload:
      customer_id: {gen: entity}
      as_of: {gen: upstream, path: as_of}
      churned:
        gen: when
        condition: upstream.sessions_30d < 5
        then: {gen: boolean, p: 0.6}
        else: {gen: boolean, p: 0.05}

run: {until: 40 days}

{"stream_type":"event","sim_ts":86400.0,"timestamp":"...","key":"customer-1","task":"features","instance":0,"value":{"customer_id":"customer-1","as_of":"2026-01-02T00:00:00.000+00:00","sessions_30d":19,"spend_30d":261.84}}
{"stream_type":"event","sim_ts":86400.0,"timestamp":"...","key":"customer-1","task":"label","instance":0,"value":{"status":"queued"}}
{"stream_type":"event","sim_ts":86400.0,"timestamp":"...","key":"customer-2","task":"features","instance":1,"value":{"customer_id":"customer-2","as_of":"2026-01-02T00:00:00.000+00:00","sessions_30d":21,"spend_30d":88.83}}
{"stream_type":"event","sim_ts":86400.0,"timestamp":"...","key":"customer-2","task":"label","instance":1,"value":{"status":"queued"}}
{"stream_type":"event","sim_ts":86400.0,"timestamp":"...","key":"customer-3","task":"features","instance":2,"value":{"customer_id":"customer-3","as_of":"2026-01-02T00:00:00.000+00:00","sessions_30d":11,"spend_30d":443.75}}

The Builder file builds the same model with the builder API and prints every record of its run:

python examples/gallery/point_in_time_training_set_builder.py

"""A point-in-time training set: feature records and the label that follows each,
builder API.

The builder version of `point_in_time_training_set.yaml`: five customers, each with a
feature record every day and, 7 days after each, a label record.

Needs: nothing
Run it with: python examples/gallery/point_in_time_training_set_builder.py
"""

import logging
from datetime import datetime, timezone

from desflow import ConsoleEgress, SimulationContext
from desflow.values import Template

UNTIL = 40 * 86400
CUSTOMERS = 5


def build() -> SimulationContext:
    """Builds the customers. A blueprint's `count: 5` is five arrivals here, each with
    the key of its customer."""
    app = SimulationContext(
        sim_id="Churn",
        factor=0,
        random_seed=42,
        logical_start_time=datetime(2026, 1, 1, tzinfo=timezone.utc),
    )
    app.add_egress(ConsoleEgress())
    for number in range(1, CUSTOMERS + 1):
        app.add_arrival(
            f"customers{number}",
            "constant",
            value=86400,  # 1 day
            spawn="features",
            key=lambda n, number=number: f"customer-{number}",
        )
    features_payload = Template(
        {
            "customer_id": {"gen": "entity"},
            "as_of": {"gen": "now"},
            "sessions_30d": {"gen": "integer", "bounds": [0, 30]},
            "spend_30d": {"gen": "uniform", "low": 0, "high": 500, "decimals": 2},
        },
        name="tasks.features",
    )
    label_payload = Template(
        {
            "customer_id": {"gen": "entity"},
            "as_of": {"gen": "upstream", "path": "as_of"},
            "churned": {
                "gen": "when",
                "condition": "upstream.sessions_30d < 5",
                "then": {"gen": "boolean", "p": 0.6},
                "else": {"gen": "boolean", "p": 0.05},
            },
        },
        name="tasks.label",
    )

    @app.task(name="features", next="label")
    def features(task_id):
        """Gives the features of a customer."""
        return features_payload

    @app.task(name="label", wait={"timeout": "7 days"})
    def label(task_id):
        """Says, 7 days later, whether the customer left."""
        return label_payload

    return app


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
    build().run(until=UNTIL)

The Low-level file builds the model on DynamicRealtimeEnvironment, with plain Python processes, and prints every record of its run:

python examples/gallery/point_in_time_training_set_low_level.py

The Low-level file draws each value with Sampler.generator(<place>), where the place is the key path of the blueprint's generator that makes the same value, such as tasks.features.payload.sessions_30d, tasks.features.payload.spend_30d and tasks.label.payload.churned.then and 1 more. Each draw therefore comes from the random sequence the blueprint uses at the same place, so the draws are the blueprint's. The file waits one day between the feature records and seven days before each label, as the blueprint's constant arrival and wait do.

"""A point-in-time training set: feature records and the label that follows each, low-level API.

The low-level version of `point_in_time_training_set.yaml`: plain SimPy on a `DynamicRealtimeEnvironment`,
which publishes each record with `publish_event`. Each draw is made with
`Sampler.generator(<name>)`, from the random sequence the blueprint's generator at that
place uses, so the draws are the blueprint's.

Needs: nothing
Run it with: python examples/gallery/point_in_time_training_set_low_level.py
"""

import itertools
import logging
from datetime import datetime, timedelta, timezone

from desflow import ConsoleEgress, DynamicRealtimeEnvironment, Sampler

UNTIL = 40 * 86400
START = datetime(2026, 1, 1, tzinfo=timezone.utc)
ENVIRONMENT = {"logical_start_time": START, "timezone": "UTC"}


def build(env: DynamicRealtimeEnvironment) -> None:
    """Starts the customers on `env`."""
    sampler = Sampler(seed=42)
    sessions = sampler.generator("tasks.features.payload.sessions_30d")
    spend = sampler.generator("tasks.features.payload.spend_30d")
    few = sampler.generator("tasks.label.payload.churned.then")
    many = sampler.generator("tasks.label.payload.churned.else")
    counts = {"features": 0, "label": 0}

    def customer(key):
        for _ in itertools.count():
            yield env.timeout(86400)
            env.process(flow(key))

    def flow(key):
        """The features of a customer now, then the label 7 days later."""
        moment = START + timedelta(seconds=env.now)
        features = {
            "customer_id": key,
            "as_of": moment.isoformat(timespec="milliseconds"),
            "sessions_30d": int(sessions.integers(0, 31)),
            "spend_30d": round(500 * spend.random(), 2),
        }
        env.publish_event(key, features, task="features", instance=counts["features"])
        counts["features"] += 1
        instance = counts["label"]  # the label waits, so it is queued first
        counts["label"] += 1
        env.publish_event(key, {"status": "queued"}, task="label", instance=instance)
        yield env.timeout(7 * 86400)
        env.publish_event(key, {"status": "started"}, task="label", instance=instance)
        # A customer with few sessions leaves with a high chance
        chance, rng = (0.6, few) if features["sessions_30d"] < 5 else (0.05, many)
        label = {
            "customer_id": key,
            "as_of": features["as_of"],
            "churned": bool(rng.random() < chance),
        }
        env.publish_event(key, label, task="label", instance=instance)

    for number in range(1, 6):
        env.process(customer(f"customer-{number}"))


def run() -> None:
    """Runs the model unpaced into the console."""
    env = DynamicRealtimeEnvironment(factor=0, **ENVIRONMENT)
    env.setup_egress([ConsoleEgress()])
    build(env)
    try:
        env.run(until=UNTIL)
    finally:
        env.teardown()


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
    run()