Live Log Streaming (Pipes)

RockyResource.run_streaming() is a Pipes-style alternative to RockyResource.run(). It spawns the binary with subprocess.Popen rather than waiting for the whole process to finish.

While the run proceeds, run_streaming forwards rocky’s stderr to context.log.info one line at a time. That stderr is where the engine’s Rust tracing layer writes its info!() and warn!() output. After the subprocess exits, run_streaming parses the final stdout JSON into a RunResult.

The payoff is live progress on a long pipeline. You see each model copy, contract check, and drift action as it happens. You do not wait for one log dump after a 30-minute rocky run finishes.

Quickstart

import dagster as dg
from dagster_rocky import RockyResource

rocky = RockyResource(config_path="rocky.toml")

@dg.asset
def my_warehouse_data(context: dg.AssetExecutionContext, rocky: RockyResource):
    # Use run_streaming so the run viewer streams progress in real time
    result = rocky.run_streaming(context, filter="tenant=acme")
    return result.tables_copied

When materialized, the Dagster run viewer shows lines like:

[INFO] rocky: INFO discovering 12 sources
[INFO] rocky: INFO catalog acme_warehouse created
[INFO] rocky: INFO copying acme.orders (15000 rows)
[INFO] rocky: INFO copying acme.payments (42000 rows)
[INFO] rocky: INFO drift check passed for acme schema
[INFO] rocky: INFO run complete in 18000ms

Each line is forwarded as the engine emits it, not at the end.

API parity with run()

run_streaming accepts the same keyword arguments as run():

result = rocky.run_streaming(
    context,
    filter="tenant=acme",
    governance_override={"workspace_ids": [12345]},
    run_models=True,
    partition="2026-04-08",
    lookback=2,
    parallel=4,
)

The first positional argument is the Dagster execution context. It is an AssetExecutionContext inside a @multi_asset, or an OpExecutionContext inside a @op. Every partition selection flag from the partitions guide works the same way here.

Automatic wiring in RockyComponent

RockyComponent calls run_streaming by default. Every multi-asset materialization therefore streams its logs. There is nothing to configure. To switch to full Pipes, see RockyComponent default below.

Failure handling

run_streaming matches run()’s failure semantics:

Outcome	Behavior
Success (exit 0)	Returns the parsed RunResult
Partial success (exit ≠0, stdout starts with {)	Returns the parsed RunResult (Rocky’s partial-success contract)
Hard failure (exit ≠0, no JSON)	Raises dg.Failure with the last 20 stderr lines in the metadata
Binary missing	Raises dg.Failure with installation instructions
Subprocess timeout	Kills the process, joins the reader thread, raises dg.Failure with the configured timeout in the message and the stderr tail

The stderr_tail metadata on a failure holds the progress lines the engine printed before it crashed. That tells you more than a bare exit code.

How it works under the hood

+-------------------+         +-----------------------+
|  Dagster context  |         |  rocky subprocess     |
|                   |         |                       |
|  context.log <----+---<<<---+ stderr (line-buffered)|
|                   |         |                       |
|       buffer  <---+---<<<---+ stdout (JSON output)  |
+-------------------+         +-----------------------+
        |                              |
        |                              v
        |                         exit code
        |                              |
        v                              |
   parse RunResult <------+-----<<<----+
                          |
                    (after wait)

1. subprocess.Popen spawns rocky with stdout=PIPE, stderr=PIPE, bufsize=1 (line-buffered).
2. Two daemon threads drain the pipes concurrently: a stderr-forwarder that sends each non-empty line to context.log.info with a rocky: prefix, and a stdout-accumulator that collects the JSON payload.
3. The main thread waits on the process. A separate watchdog thread enforces the timeout. It kills the process group if the process has not exited in time.
4. After the subprocess exits, the reader threads join (with a 2-second grace period for any in-flight lines).
5. If exit is clean or partial-success, the captured stdout is parsed into a RunResult.

Three execution modes

RockyResource ships three ways to run rocky:

	run()	run_streaming()	run_pipes()
Live log streaming	❌ buffered	✅ stderr forwarding	✅ via Pipes protocol
Structured MaterializationEvent from Pipes	❌	❌	✅
Returns	RunResult	RunResult	PipesClientCompletedInvocation
Needs Dagster context	no	yes	yes
Engine Pipes support required	no	no	yes (covered by the engine floor, 1.35.0)
Watchdog timeout	yes	yes	plan step only

run(): buffered (non-Dagster callers)

result = rocky.run(filter="tenant=acme")

For scripts, tests, notebooks, or any code that just wants the typed result without a Dagster context. Engine stderr goes to the dagster_rocky.resource logger, not to context.log.

run_streaming(): Pipes-style (live progress, batch result)

@dg.asset
def my_asset(context, rocky: RockyResource):
    result = rocky.run_streaming(context, filter="tenant=acme")
    return result.tables_copied

Live progress with a batch result. It does not depend on Pipes message emission, so it works against any rocky binary.

run_pipes(): full Dagster Pipes (structured events)

@dg.asset
def my_asset(context: dg.AssetExecutionContext, rocky: RockyResource):
    yield from rocky.run_pipes(context, filter="tenant=acme").get_results()

The resource runs rocky plan first, which writes .rocky/plans/<plan-id>.json. It then runs rocky apply <plan-id> through dg.PipesSubprocessClient, which sets the DAGSTER_PIPES_CONTEXT and DAGSTER_PIPES_MESSAGES env vars. The plan id travels as extras={"plan_id": plan_id}. See Plan artifact per materialization.

The engine detects those env vars and emits structured Pipes messages. See Engine-side emission for the message types. In the run viewer they arrive as MaterializationEvents, carrying strategy, duration_ms, rows_copied, sql_hash, and partition_key, plus AssetCheckEvaluations.

timeout_seconds bounds only the plan step in this mode. PipesSubprocessClient owns the apply subprocess and has no kill hook. Set a Dagster run timeout, or use run_streaming(), when you need a hard bound. See the RockyResource reference.

If your own code starts a subprocess inside a Pipes run, that subprocess inherits DAGSTER_PIPES_* variables by default. Remove all variables with that prefix from its environment unless the child is meant to speak on the same Pipes channel. Rocky removes them from processes it starts itself, including hooks and fulfillment workers.

Returns a PipesClientCompletedInvocation. Call .get_results() to extract the materialization events Dagster built from the Pipes messages.

If rocky plan emits no plan_id, run_pipes raises dg.Failure. It never runs without one.

Engine-side: Dagster Pipes message emission

The rocky engine implements the Dagster Pipes protocol directly, with no external dependency. On a run it:

1. Detects DAGSTER_PIPES_CONTEXT and DAGSTER_PIPES_MESSAGES env vars at the start of rocky run.
2. Opens the messages channel (file path or stderr stream) per the protocol params, and writes opened immediately (params: {"extras": {}}) — the handshake Dagster’s reader needs before it will report anything at all.
3. Emits one JSON-line message per progress event:
   • log at run start and completion
   • report_asset_materialization per output.materializations entry
   • report_asset_check per output.check_results entry
   • per output.anomalies entry: a report_asset_check (check name row_count_anomaly, severity WARN, passed=false, with rocky/current_count, rocky/baseline_avg, rocky/deviation_pct and rocky/reason metadata)
   • per output.anomaly_evaluated entry not already covered by an anomaly above: a report_asset_check (check name row_count_anomaly) — passed=true for a table the detector evaluated with no anomaly, or passed=false with a rocky/reason for a table it did not evaluate. Matches the same three verdicts execution_mode: streaming gives.
   • per output.drift.actions_taken entry: a report_asset_check (check name drift, severity WARN, passed=true, with table/action/reason metadata) plus a log at WARN level. drift is never a declared check spec — dagster-rocky converts it to an AssetObservation with rocky/drift_* metadata on receipt, matching what execution_mode: streaming yields.
   • closed at run end
4. When env vars are not set, the entire path is a no-op; zero overhead for non-Dagster callers.

The current engine emission is batch at end of run. Events emit right before the JSON output payload, not as each table completes. A future engine release can move to per-event streaming without changing the wire protocol or any consumer.

If Dagster closes the messages channel after receiving opened, Rocky logs subsequent write errors but does not cancel the run. Warehouse writes can continue while Dagster receives no further events. A run that needs strict observation should monitor the Rocky process and reconcile its final result and warehouse state after a lost channel.

RockyComponent default

RockyComponent streams by default, with execution_mode: streaming. Each rocky run is buffered by run_streaming, and the component’s own result-emitter translates Rocky’s JSON output into Dagster events.

For full Pipes integration with structured engine events, set execution_mode: pipes on the component. Each run then goes through run_pipes. The engine emits materialization and check events directly over the Pipes wire. Asset-key translation and subset filtering happen at the reader layer:

type: dagster_rocky.RockyComponent
attributes:
  config_path: rocky.toml
  execution_mode: pipes