Prefect and Dagster

sqlload load --events jsonl writes one JSON object per load event to stderr, next to the logs. The first recipe for each orchestrator reads stderr line by line, sends each event to the orchestrator’s logger at its level, and records a row per table. A table that fails logs an error, and the run fails when sqlload load exits with a non-zero code. Run every recipe in the project folder, with sqlload on PATH.

Prefect

The flow logs through get_run_logger() and saves a table artifact, sqlload-load, with one row per table.

import json
import subprocess

from prefect import flow, get_run_logger
from prefect.artifacts import create_table_artifact


@flow
def sqlload_load(project: str = ".") -> None:
    logger = get_run_logger()
    load = subprocess.Popen(
        ["sqlload", "load", "--events", "jsonl", "--progress", "never"],
        cwd=project,
        stderr=subprocess.PIPE,
        text=True,
    )
    tables = []
    for line in load.stderr:
        try:
            event = json.loads(line)
        except ValueError:
            event = None
        if not isinstance(event, dict) or "schema_version" not in event:
            logger.info(line.rstrip())
            continue
        kind, table = event["event"], event.get("table", "")
        if kind == "table_done":
            logger.info("loaded %s: %s rows in %s ms", table, event["rows"], event["elapsed_ms"])
            tables.append({"table": table, "status": "loaded", "rows": event["rows"], "ms": event["elapsed_ms"]})
        elif kind == "table_failed":
            logger.error("failed %s: %s", table, event["reason"])
            tables.append({"table": table, "status": "failed", "rows": 0, "ms": event["elapsed_ms"]})
        elif kind == "table_not_started":
            logger.warning("not started %s: %s", table, event["reason"])
            tables.append({"table": table, "status": "not started", "rows": 0, "ms": 0})
        elif kind == "rate_limited":
            logger.warning("%s waits %s ms under rate_limit %s", event["connection"], event["wait_ms"], event["rate"])
        elif kind == "run_finished":
            logger.info("%s loaded, %s failed, %s rows", event["loaded"], event["failed"], event["rows"])
        else:
            logger.debug(line.rstrip())
    create_table_artifact(key="sqlload-load", table=tables, description="sqlload load")
    if load.wait() != 0:
        raise RuntimeError(f"sqlload load exited with {load.returncode}")


if __name__ == "__main__":
    sqlload_load()

One asset per table

One sqlload load loads every table of the job. To give each table its own Prefect asset, with its own health and metadata, record the assets from the job’s --json report once the load and its retries have finished:

import json
import subprocess

from prefect import flow, task
from prefect.assets import add_asset_metadata, materialize


class LoadFailed(Exception):  # args: the exit code and the --json report
    def __str__(self) -> str:
        error = self.args[1].get("error", {}).get("message")
        return f"sqlload load exited with {self.args[0]}" + (f": {error}" if error else "")


def retry_on_75(task, task_run, state) -> bool:
    error = state.result(raise_on_failure=False)
    return isinstance(error, LoadFailed) and error.args[0] == 75


@task(retries=2, retry_delay_seconds=30, retry_condition_fn=retry_on_75)
def sqlload_load(project: str) -> dict:
    load = subprocess.run(["sqlload", "load", "--json"], cwd=project, stdout=subprocess.PIPE, text=True)
    report = json.loads(load.stdout)
    if load.returncode != 0:
        raise LoadFailed(load.returncode, report)
    return report


@materialize("sqlload://table")  # each call names its table with .with_options(assets=[...])
def record_table(key: str, table: dict) -> None:
    add_asset_metadata(key, {k: table[k] for k in ("rows", "duration_seconds", "write_strategy", "source", "note")})
    if table["status"] == "failed":
        raise RuntimeError(f"{table['table']}: {table['error']}")


@flow
def load_warehouse(project: str = ".") -> None:
    outcome = sqlload_load(project, return_state=True).result(raise_on_failure=False)
    if isinstance(outcome, Exception) and not isinstance(outcome, LoadFailed):
        raise outcome
    report = outcome.args[1] if isinstance(outcome, LoadFailed) else outcome
    for table in report.get("tables", []):
        if table["status"] in ("loaded", "failed"):
            key = f"sqlload://{table['table']}"
            record_table.with_options(assets=[key])(key, table, return_state=True)
    if isinstance(outcome, LoadFailed):
        raise outcome


if __name__ == "__main__":
    load_warehouse()

• The assets come from the final report, after retries, so an outage that a retry recovers from leaves no failed asset.
• On a failed load, LoadFailed carries the report, so the tables that loaded still record and the ones that failed record as failed. The flow then fails with an error naming the load’s exit code, and the error sqlload load printed when the run couldn’t start.
• return_state=True keeps one failed table from stopping the others from recording.
• A table --fail-fast didn’t start has status skipped and records nothing.
• Only exit code 75 retries: every failure may pass on a rerun. See Retrying failed runs.
• Declaring every table as an asset of the load task instead would give them all the task’s one final state, which is wrong when only one table fails.

Dagster

The asset logs through context.log and attaches a tables table to its materialization, or to the failure when the load fails.

import json
import subprocess

import dagster as dg


@dg.asset
def sqlload_tables(context: dg.AssetExecutionContext) -> dg.MaterializeResult:
    load = subprocess.Popen(
        ["sqlload", "load", "--events", "jsonl", "--progress", "never"],
        stderr=subprocess.PIPE,
        text=True,
    )
    records = []
    for line in load.stderr:
        try:
            event = json.loads(line)
        except ValueError:
            event = None
        if not isinstance(event, dict) or "schema_version" not in event:
            context.log.info(line.rstrip())
            continue
        kind, table = event["event"], event.get("table", "")
        if kind == "table_done":
            context.log.info(f"loaded {table}: {event['rows']} rows in {event['elapsed_ms']} ms")
            records.append(dg.TableRecord({"table": table, "status": "loaded", "rows": event["rows"], "ms": event["elapsed_ms"]}))
        elif kind == "table_failed":
            context.log.error(f"failed {table}: {event['reason']}")
            records.append(dg.TableRecord({"table": table, "status": "failed", "rows": 0, "ms": event["elapsed_ms"]}))
        elif kind == "table_not_started":
            context.log.warning(f"not started {table}: {event['reason']}")
            records.append(dg.TableRecord({"table": table, "status": "not started", "rows": 0, "ms": 0}))
        elif kind == "rate_limited":
            context.log.warning(f"{event['connection']} waits {event['wait_ms']} ms under rate_limit {event['rate']}")
        elif kind == "run_finished":
            context.log.info(f"{event['loaded']} loaded, {event['failed']} failed, {event['rows']} rows")
        else:
            context.log.debug(line.rstrip())
    schema = dg.TableSchema(
        columns=[
            dg.TableColumn("table", "string"),
            dg.TableColumn("status", "string"),
            dg.TableColumn("rows", "int"),
            dg.TableColumn("ms", "int"),
        ]
    )
    metadata = {"tables": dg.MetadataValue.table(records, schema=schema)}
    if load.wait() != 0:
        raise dg.Failure(f"sqlload load exited with {load.returncode}", metadata=metadata)
    return dg.MaterializeResult(metadata=metadata)


defs = dg.Definitions(assets=[sqlload_tables])

The two event recipes log lines that are not events, such as warnings, as they are, and table_started, table_progress and table_committing at debug level.

One asset per table

A multi_asset declares one asset per table of the job and yields a MaterializeResult for each table the final --json report lists as loaded:

import json
import subprocess

import dagster as dg

# The job's tables, as sqlload plan names them.
TABLES = ["raw.shop__customers", "raw.shop__orders", "raw.shop__products", "raw.shop__refunds"]


@dg.multi_asset(specs=[dg.AssetSpec(t.split("."), skippable=True) for t in TABLES])
def sqlload_tables(context: dg.AssetExecutionContext):
    load = subprocess.run(["sqlload", "load", "--json"], stdout=subprocess.PIPE, text=True)
    report = json.loads(load.stdout)
    if load.returncode == 75 and context.retry_number < 2:
        raise dg.RetryRequested(max_retries=2, seconds_to_wait=30)
    failed = []
    for table in report.get("tables", []):
        if table["status"] == "loaded":
            metadata = {k: table[k] for k in ("rows", "duration_seconds", "write_strategy", "source", "note")}
            yield dg.MaterializeResult(asset_key=table["table"].split("."), metadata=metadata)
        elif table["status"] in ("failed", "skipped"):
            failed.append(f"{table['table']}: {table['error']}")
    if load.returncode != 0:
        error = report.get("error", {}).get("message")
        raise dg.Failure("\n".join([f"sqlload load exited with {load.returncode}", *failed, *filter(None, [error])]))


defs = dg.Definitions(assets=[sqlload_tables])

• Exit code 75 asks Dagster to run the step again, up to twice, before anything is materialized, so a retried outage leaves no failed asset.
• On the last attempt, each loaded table materializes. A table that failed or wasn’t started doesn’t, and the step fails with each such table’s error, or with the error sqlload load printed when the run couldn’t start.
• skippable=True lets a table with nothing to load, such as a files table with no new file, go unmaterialized when the load succeeds.
• Keep TABLES in line with the job’s tables: yielding a table that isn’t listed fails the step.