Skip to content

Fast-Forward to Iceberg

This example generates a day of a shop's orders at factor=0.0 and appends them to an Apache Iceberg table through an Iceberg REST catalog. Write to Iceberg Tables shows the other catalogs. Each tab shows the example written one way: as a YAML blueprint, with the builder API, or with the low-level API. Ways to write a simulation compares the three.

ParquetStorageEgress writes files and leaves registering them as a table to you. IcebergStorageEgress appends into an Iceberg table instead, so one run ends with a table another engine can query. Look for the day's events as rows of simulation.events, and for the few snapshots that hold them.

Each flush of the buffer is one Iceberg commit. Every commit writes a manifest, a manifest list and a new metadata.json, so query planning slows down as snapshots build up. This example therefore gives the Iceberg egress a large batch_size of its own. At factor=0.0 the buffer is flushed each time it holds 200,000 records, and once more at the end of the run. Each flush adds one snapshot to the table.

Before You Start

The example needs Docker and the odctl profile catalog. Each Quick Start below starts the profile, and the Check the Result section stops it. Services for the Examples shows how to install odctl.

Steps

Run It

This example builds the simulation of the builder example in the Builder tab from a YAML blueprint, with no Python. A day of shop orders is generated at factor: 0.0, and each flush of the buffer becomes one Iceberg commit.

Quick Start

desflow examples get downloads the example at the release you install, into a folder of its own. Pick uv or pip:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
# 1. Install
uv tool install "desflow[iceberg]" --with-executables-from "odctl>=1.0,<2"

# 2. Download the example
desflow examples get iceberg --api yaml
cd iceberg/yaml

# 3. Find the bottleneck, with no run
desflow explain iceberg.yaml

# 4. Start its service
odctl up catalog

# 5. Run it
desflow run iceberg.yaml

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
# 1. Install
pip install "desflow[iceberg]" "odctl>=1.0,<2"

# 2. Download the example
desflow examples get iceberg --api yaml
cd iceberg/yaml

# 3. Find the bottleneck, with no run
desflow explain iceberg.yaml

# 4. Start its service
odctl up catalog

# 5. Run it
desflow run iceberg.yaml

What It Does

The run appends a day of lifecycle events to the simulation.events table and ends on its own. With batch_size: 200000 on the egress, the buffer is flushed once per 200,000 records, and each flush is one commit. The day therefore lands in a few snapshots. Without a router, each event is written to default_table as one flat row, and telemetry is left out.

The catalog is contacted on the first write, not when the blueprint is loaded, so a blueprint with a wrong catalog address still loads and fails when the first batch is written.

ICEBERG_URI, ICEBERG_WAREHOUSE, S3_ENDPOINT, S3_ACCESS_KEY, S3_SECRET_KEY and S3_REGION override the catalog and credentials.

This version builds the simulation with the builder API (SimulationContext) and writes with IcebergStorageEgress.

Quick Start

desflow examples get downloads the example at the release you install, into a folder of its own. Pick uv or pip:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
# 1. Install
uv tool install "desflow[iceberg]" --with-executables-from "odctl>=1.0,<2"

# 2. Download the example
desflow examples get iceberg --api builder
cd iceberg/builder

# 3. Start its service
odctl up catalog

# 4. Run it
uv run --no-project --with "desflow[iceberg]" iceberg_example.py

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
# 1. Install
pip install "desflow[iceberg]" "odctl>=1.0,<2"

# 2. Download the example
desflow examples get iceberg --api builder
cd iceberg/builder

# 3. Start its service
odctl up catalog

# 4. Run it
python iceberg_example.py

ICEBERG_URI, ICEBERG_WAREHOUSE, ICEBERG_NAMESPACE, S3_ENDPOINT, S3_ACCESS_KEY, S3_SECRET_KEY and S3_REGION override the catalog and credentials.

A second run appends to the same table rather than replacing it, because IcebergStorageEgress appends every record by default. To start again with an empty catalog, stop the service as Check the Result shows and start it again.

This version wires the simulation by hand on DynamicRealtimeEnvironment, the low-level API, and writes with IcebergStorageEgress.

Quick Start

desflow examples get downloads the example at the release you install, into a folder of its own. Pick uv or pip:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
# 1. Install
uv tool install "desflow[iceberg]" --with-executables-from "odctl>=1.0,<2"

# 2. Download the example
desflow examples get iceberg --api low-level
cd iceberg/low-level

# 3. Start its service
odctl up catalog

# 4. Run it
uv run --no-project --with "desflow[iceberg]" iceberg_example.py

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
# 1. Install
pip install "desflow[iceberg]" "odctl>=1.0,<2"

# 2. Download the example
desflow examples get iceberg --api low-level
cd iceberg/low-level

# 3. Start its service
odctl up catalog

# 4. Run it
python iceberg_example.py

ICEBERG_URI, ICEBERG_WAREHOUSE, ICEBERG_NAMESPACE, S3_ENDPOINT, S3_ACCESS_KEY, S3_SECRET_KEY and S3_REGION override the catalog and credentials.

A second run appends to the same table rather than replacing it, because IcebergStorageEgress appends every record by default. To start again with an empty catalog, stop the service as Check the Result shows and start it again.

Check the Result

The three tabs write the same records, which a test checks by running each version unpaced. A real run appends them as rows of the simulation.events table. When the run has ended, leave the service up and read the table back from the catalog, with the same endpoints and credentials as the blueprint:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
uv run --no-project --with "desflow[iceberg]" \
python -c "
from pyiceberg.catalog import load_catalog
catalog = load_catalog(
    'odctl', type='rest', uri='http://127.0.0.1:8181',
    **{'s3.endpoint': 'http://127.0.0.1:8333', 's3.access-key-id': 'user',
       's3.secret-access-key': 'password', 's3.region': 'us-east-1'},
)
table = catalog.load_table('simulation.events')
print(table.scan().to_arrow().num_rows, len(table.snapshots()))
"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
python -c "
from pyiceberg.catalog import load_catalog
catalog = load_catalog(
    'odctl', type='rest', uri='http://127.0.0.1:8181',
    **{'s3.endpoint': 'http://127.0.0.1:8333', 's3.access-key-id': 'user',
       's3.secret-access-key': 'password', 's3.region': 'us-east-1'},
)
table = catalog.load_table('simulation.events')
print(table.scan().to_arrow().num_rows, len(table.snapshots()))
"

It prints the number of rows and the number of snapshots. After one run with the seed in the file these are 515786 3: one row for each lifecycle event of the day, in three snapshots (two flushes of 200,000 records and the final flush). After a second run both numbers grow, because each run appends. When you have checked the result, stop the service:

1
odctl down catalog --volumes

In iceberg/yaml, which desflow examples get iceberg --api yaml creates, desflow run iceberg.yaml --sample 4 --stdout needs no catalog. It prints the first four events as JSON lines instead:

1
2
3
4
{"stream_type":"event","sim_ts":0.6380949691741167,"timestamp":"...","key":"orders-1","task":"pack_order","instance":0,"value":{"path_id":"Shop.service.packing","sim_id":"Shop","service":"packing","status":"queued"}}
{"stream_type":"event","sim_ts":0.6380949691741167,"timestamp":"...","key":"orders-1","task":"pack_order","instance":0,"value":{"path_id":"Shop.service.packing","sim_id":"Shop","service":"packing","status":"started"}}
{"stream_type":"event","sim_ts":0.6739613317861898,"timestamp":"...","key":"orders-2","task":"pack_order","instance":1,"value":{"path_id":"Shop.service.packing","sim_id":"Shop","service":"packing","status":"queued"}}
{"stream_type":"event","sim_ts":0.6739613317861898,"timestamp":"...","key":"orders-2","task":"pack_order","instance":1,"value":{"path_id":"Shop.service.packing","sim_id":"Shop","service":"packing","status":"started"}}

Each event becomes one row of the table, and the two fields of value, path_id and status, become columns of their own.

Full Files

Full files

catalog is a mapping of PyIceberg catalog properties, passed to PyIceberg's load_catalog. schemas names a type for each column, so timestamp is created as a timestamp rather than inferred as a string. The ISO time each record carries is converted to a timestamp before the write.

Files live in the examples/yaml/ folder of the repository, and the label on each block below is its path there.

examples/yaml/iceberg.yaml
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
# Lakehouse data generation in YAML.
#
# The YAML version of iceberg_example.py, which --api builder downloads. A day of shop
# orders is generated at factor 0, and each flush of the buffer becomes one Iceberg
# commit. Without a router, each event is written to default_table as one flat row
# and telemetry is left out. The catalog is contacted on the first write.
#
# Needs the odctl catalog profile: odctl up catalog. ICEBERG_URI, ICEBERG_WAREHOUSE,
# S3_ENDPOINT, S3_ACCESS_KEY and S3_SECRET_KEY override the catalog and credentials.
# Run it from its folder with: desflow run iceberg.yaml

simulation:
  sim_id: Shop
  factor: 0.0
  random_seed: 42
  # A day before now, so the records are stamped as history.
  logical_start_time: -1d

egress:
  - type: Iceberg
    config:
      # PyIceberg catalog properties, passed to load_catalog.
      catalog:
        name: odctl
        type: rest
        uri: ${ICEBERG_URI:-http://127.0.0.1:8181}
        warehouse: ${ICEBERG_WAREHOUSE:-s3://warehouse/}
        s3.endpoint: ${S3_ENDPOINT:-http://127.0.0.1:8333}
        s3.access-key-id: ${S3_ACCESS_KEY:-user}
        s3.secret-access-key: ${S3_SECRET_KEY:-password}
        s3.region: ${S3_REGION:-us-east-1}
      default_table: simulation.events
      # Pinned rather than inferred, so timestamp is a timestamp, not a string.
      schemas:
        simulation.events:
          sim_ts: double
          timestamp: timestamp
          key: string
          path_id: string
          status: string
    # One commit per flush, sized so a day of events lands in a few snapshots.
    batch_size: 200000

resources:
  packers: {current_cap: 4, max_cap: 10}

services:
  packing: {dist: normal, mean: 2.0, std: 0.2}

arrivals:
  orders: {dist: exponential, rate: 2.0, spawn: pack_order}

tasks:
  pack_order:
    service: packing
    resource: packers
    payload: {path_id: Shop.service.packing, status: finished}

run: {until: 1 day}

This script simulates the shop over a 1-day period. It routes lifecycle events into one Iceberg table. It pins the schema, so the event time is stored as a timestamp. Without the schema, the column would be inferred as a string.

Scripts live in the examples/ folder of the repository, and the label on the block below is this one's path there.

examples/builder/iceberg_example.py
  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
"""
Lakehouse Data Generation Example.

The Iceberg version of `builder/parquet_example.py`. The same fast-forward engine,
but each flush of the buffer becomes one Iceberg commit rather than one Parquet file,
so the run ends with a table another engine can query instead of a folder someone
still has to register.

Every commit writes a manifest, a manifest list and a new `metadata.json`.
`batch_size` on `add_egress` sizes the buffer, so the run produces a handful of
snapshots rather than hundreds.

Requires the odctl `catalog` profile: `odctl up catalog`.
"""

import logging
import os
from datetime import datetime, timedelta, timezone

import pyarrow as pa

from desflow import IcebergStorageEgress, SimulationContext

# Logging is configured here rather than in a wrapper, because this script is run
# directly. Without it the run prints no INFO log lines, such as its progress
# messages.
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    datefmt="%H:%M:%S",
)

logger = logging.getLogger(__name__)

NAMESPACE = os.getenv("ICEBERG_NAMESPACE", "simulation")
EVENTS_TABLE = f"{NAMESPACE}.events"

# Pinned rather than inferred. Inference reads the ISO timestamp the environment
# writes as a string, which is not what a consumer of an event table expects.
EVENTS_SCHEMA = pa.schema(
    [
        ("sim_ts", pa.float64()),
        ("timestamp", pa.timestamp("us")),
        ("key", pa.string()),
        ("path_id", pa.string()),
        ("status", pa.string()),
    ]
)


def create_table_router(events_table: str):
    """
    Router Factory: Generates a router returning `namespace.table`, and reshapes
    each event into the pinned schema.
    """

    def table_router(data: dict) -> str | None:
        if data.get("path_id") == "system.simulation.lag_seconds":
            return None

        if data.get("stream_type") != "event":
            return None

        # FLATTEN EVENT INTO THE PINNED COLUMNS
        nested_value = data.pop("value", None)
        if isinstance(nested_value, dict):
            data.update(nested_value)

        # A pinned timestamp column takes a datetime. PyArrow rejects the ISO string
        # the environment writes, so the conversion belongs here.
        data["timestamp"] = datetime.fromisoformat(data["timestamp"])

        return events_table

    return table_router


# ==========================================
# 1. CATALOG CONFIGURATION
# ==========================================
# RestCatalog fetches the catalog configuration as it is constructed, so building
# the client below contacts the catalog when this module is imported. Creating the
# namespace waits longer: the connector does that on its first batch, inside the run.
def build_catalog():
    """Connects to the Iceberg REST catalog from the odctl `catalog` profile."""
    from pyiceberg.catalog.rest import RestCatalog

    return RestCatalog(
        "odctl",
        **{
            "uri": os.getenv("ICEBERG_URI", "http://127.0.0.1:8181"),
            "warehouse": os.getenv("ICEBERG_WAREHOUSE", "s3://warehouse/"),
            "s3.endpoint": os.getenv("S3_ENDPOINT", "http://127.0.0.1:8333"),
            "s3.access-key-id": os.getenv("S3_ACCESS_KEY", "user"),
            "s3.secret-access-key": os.getenv("S3_SECRET_KEY", "password"),
            "s3.region": os.getenv("S3_REGION", "us-east-1"),
        },
    )


router = create_table_router(EVENTS_TABLE)

# The day of backdating this example exists to demonstrate. It sits at module scope
# because the builder below needs it, and the builder has to stay at module scope for
# the decorators further down to attach to it.
LOGICAL_START_TIME = datetime.now(timezone.utc) - timedelta(days=1)

# ==========================================
# 2. Infrastructure Builder
# ==========================================
app = (
    SimulationContext(
        sim_id="Shop",
        factor=0.0,
        random_seed=42,
        logical_start_time=LOGICAL_START_TIME,
    )
    .add_egress(
        IcebergStorageEgress(
            catalog=build_catalog(),
            table_router=router,
            schemas={EVENTS_TABLE: EVENTS_SCHEMA},
        ),
        # One commit per flush. Sized so a day of events lands in a few snapshots.
        batch_size=200_000,
    )
    .add_resource("packers", current_cap=4, max_cap=10)
    .add_service("packing", dist="normal", mean=2.0, std=0.2)
    .add_arrival("orders", dist="exponential", rate=2.0)
)


# ==========================================
# 3. Simulation Logic
# ==========================================
@app.task(service_id="packing", resource_id="packers")
def pack_order(task_id: int, context):
    """
    Returns the exact flat dictionary expected by the table router
    to represent the 'finished' state of the lifecycle.
    """
    return {"path_id": "Shop.service.packing", "status": "finished"}


@app.arrival_loop("orders")
def arrival_generator(context):
    task_id = 0
    while True:
        entity = yield context.wait_for_arrival("orders")
        context.spawn(pack_order(task_id, context, entity=entity))
        task_id += 1


# ==========================================
# 4. Execution
# ==========================================
def run():
    """Generates 1 day of shop orders instantly, into an Iceberg table."""
    logger.info(
        "Generating historical data mimicking start from "
        f"{LOGICAL_START_TIME.strftime('%Y-%m-%d %H:%M:%S')}"
    )
    logger.info("Fast-forwarding (factor=0.0)...")

    # The clock detaches and executes 1 day of operations instantaneously
    app.run(until="1 day")

    table = build_catalog().load_table(EVENTS_TABLE)
    logger.info(
        f"Wrote {table.scan().to_arrow().num_rows} rows to '{EVENTS_TABLE}' "
        f"in {len(table.metadata.snapshots)} snapshots at {table.location()}"
    )


if __name__ == "__main__":
    run()

This script simulates the shop over a 1-day period. It routes lifecycle events into one Iceberg table. It pins the schema, so the event time is stored as a timestamp. Without the schema, the column would be inferred as a string.

Scripts live in the examples/ folder of the repository, and the label on the block below is this one's path there.

examples/low-level/iceberg_example.py
  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
"""Lakehouse data generation, low-level API.

The low-level version of `builder/iceberg_example.py`. It wires
`DynamicRealtimeEnvironment`, the registry and the connectors by hand rather than
through the builder, which shows what `SimulationContext` does for you.

`factor=0.0` detaches the clock from real time, so a day of shop orders is generated
as fast as the machine allows and committed into an Iceberg table. Each flush of the
buffer is one commit, so `batch_sizes` is what keeps the snapshot count down.

Requires the odctl `catalog` profile: `odctl up catalog`.
"""

import logging
import os
from datetime import datetime, timedelta, timezone

import numpy as np
import pyarrow as pa

from desflow import (
    CapacityConfig,
    DistributionConfig,
    DynamicRealtimeEnvironment,
    DynamicResource,
    IcebergStorageEgress,
    Sampler,
    SimParameter,
    time_to_seconds,
)

logging.basicConfig(
    level=logging.INFO, format="%(levelname)s [%(asctime)s] %(message)s"
)
logger = logging.getLogger("iceberg_example")

NAMESPACE = os.getenv("ICEBERG_NAMESPACE", "simulation")
EVENTS_TABLE = f"{NAMESPACE}.events"

# Pinned rather than inferred. Inference reads the ISO timestamp the environment
# writes as a string, which is not what a consumer of an event table expects.
EVENTS_SCHEMA = pa.schema(
    [
        ("sim_ts", pa.float64()),
        ("timestamp", pa.timestamp("us")),
        ("key", pa.string()),
        ("path_id", pa.string()),
        ("status", pa.string()),
    ]
)


def build_catalog():
    """Connects to the Iceberg REST catalog from the odctl `catalog` profile."""
    from pyiceberg.catalog.rest import RestCatalog

    return RestCatalog(
        "odctl",
        **{
            "uri": os.getenv("ICEBERG_URI", "http://127.0.0.1:8181"),
            "warehouse": os.getenv("ICEBERG_WAREHOUSE", "s3://warehouse/"),
            "s3.endpoint": os.getenv("S3_ENDPOINT", "http://127.0.0.1:8333"),
            "s3.access-key-id": os.getenv("S3_ACCESS_KEY", "user"),
            "s3.secret-access-key": os.getenv("S3_SECRET_KEY", "password"),
            "s3.region": os.getenv("S3_REGION", "us-east-1"),
        },
    )


def create_table_router(events_table: str):
    """
    Router Factory: Generates a router returning `namespace.table`, and reshapes
    each event into the pinned schema.
    """

    def table_router(data: dict) -> str | None:
        if data.get("path_id") == "system.simulation.lag_seconds":
            return None

        if data.get("stream_type") != "event":
            return None

        # FLATTEN EVENT INTO THE PINNED COLUMNS
        nested_value = data.pop("value", None)
        if isinstance(nested_value, dict):
            data.update(nested_value)

        # A pinned timestamp column takes a datetime. PyArrow rejects the ISO string
        # the environment writes, so the conversion belongs here.
        data["timestamp"] = datetime.fromisoformat(data["timestamp"])

        return events_table

    return table_router


# The shop starts with 4 packers and has room for 10.
PARAMS = SimParameter(
    sim_id="Shop",
    arrival={"orders": DistributionConfig(dist="exponential", rate=2.0)},
    service={"packing": DistributionConfig(dist="normal", mean=2.0, std=0.2)},
    resources={"packers": CapacityConfig(current_cap=4, max_cap=10)},
)


def pack_order(env, sampler, packers, key, task_id):
    """What @app.task does: queued, started, the service time, then finished."""
    path_id = "Shop.service.packing"
    env.publish_event(
        key,
        {
            "path_id": path_id,
            "sim_id": "Shop",
            "service": "packing",
            "status": "queued",
        },
        task="pack_order",
        instance=task_id,
    )
    with packers.request() as req:
        yield req
        env.publish_event(
            key,
            {
                "path_id": path_id,
                "sim_id": "Shop",
                "service": "packing",
                "status": "started",
            },
            task="pack_order",
            instance=task_id,
        )
        yield env.timeout(sampler.sample(env.registry.get_config(path_id)))
        env.publish_event(
            key,
            {
                "path_id": path_id,
                "status": "finished",
            },
            task="pack_order",
            instance=task_id,
        )


def arrival_generator(env, sampler, packers):
    gap = env.registry.get_config("Shop.arrival.orders")
    task_id = 0
    while True:
        yield env.timeout(sampler.sample(gap))
        env.process(pack_order(env, sampler, packers, f"orders-{task_id + 1}", task_id))
        task_id += 1


def run():
    # ---------------------------------------------------------
    # 1. SIMULATION SETUP
    # ---------------------------------------------------------
    start_time = datetime.now(timezone.utc) - timedelta(days=1)
    env = DynamicRealtimeEnvironment(factor=0.0, logical_start_time=start_time)
    env.registry.register_sim_parameter(PARAMS)

    egress = IcebergStorageEgress(
        catalog=build_catalog(),
        table_router=create_table_router(EVENTS_TABLE),
        schemas={EVENTS_TABLE: EVENTS_SCHEMA},
    )

    # One commit per flush, so a day of events lands in a handful of snapshots.
    env.setup_egress([egress], batch_sizes=[200_000], flush_interval=86400)

    packers = DynamicResource(env, "Shop", "packers")
    sampler = Sampler(rng=np.random.default_rng(42))

    env.process(arrival_generator(env, sampler, packers))

    run_duration_str = "1 day"
    run_duration_sec = time_to_seconds(run_duration_str)

    logger.info(
        f"Generating historical data from {start_time.strftime('%Y-%m-%d %H:%M:%S')}"
    )
    logger.info("Fast-forwarding (factor=0.0)...")

    try:
        env.run(until=run_duration_sec)
    finally:
        env.teardown()

    table = build_catalog().load_table(EVENTS_TABLE)
    logger.info(
        f"Wrote {table.scan().to_arrow().num_rows} rows to '{EVENTS_TABLE}' "
        f"in {len(table.metadata.snapshots)} snapshots at {table.location()}"
    )


if __name__ == "__main__":
    run()