Artifacts and Logging

Prefect artifacts (markdown, table, link, progress) and run-scoped logging with get_run_logger, log_prints, levels, and third-party capture.

Airflow gives you two observability surfaces: the task log, and whatever you bolt on around it. Prefect gives you three: run-scoped logs, states (covered earlier), and artifacts — a first-class, published, versioned output attached to a run. Artifacts have no Airflow equivalent, and once you have them you stop misusing logs for things logs were never good at.

This chapter is about the two you reach for constantly: artifacts and logging. Both flow to the API and show up in the UI, both are keyed to the run that produced them, and both are cheap to add. The difference is what they are for. A log line is a timestamped event in a stream you scroll. An artifact is a rendered, durable, addressable piece of output you link to.

Why artifacts exist

In Airflow, when a task wants to surface a result — a row-count summary, a data-quality report, a link to the file it wrote — you have bad options. You log it (and it scrolls away). You push it to XCom (and it is invisible unless someone opens the XCom tab and reads raw JSON). You write a custom Slack notifier. None of these produce something a teammate can look at three days later without replaying the run in their head.

Prefect artifacts are the fix. An artifact is a piece of output you explicitly publish from inside a flow or task run. It renders in the UI, it is attached to the run, and — crucially — it can be keyed, giving you a versioned history of that output across every run that emitted it. There are four types:

  • create_markdown_artifact — a rendered Markdown document (reports, summaries, tables you hand-format).
  • create_table_artifact — a structured table from a list of dicts or a dict of columns.
  • create_link_artifact — a hyperlink to something the run produced (an S3 object, a dashboard, a report URL).
  • create_progress_artifact / update_progress_artifact — a live progress bar for a long-running loop.

All of them are imported from prefect.artifacts and called from within run context.

A Markdown run-report

The most common artifact is a formatted summary at the end of a flow. Instead of logging twelve lines that a human has to reassemble, you publish one rendered document.

from prefect import flow, task
from prefect.artifacts import create_markdown_artifact


@task
def load_orders() -> dict:
    # ... real work ...
    return {"ingested": 1_284, "rejected": 17, "source": "orders_api"}


@flow
def daily_ingest():
    stats = load_orders()

    report = f"""# Daily Ingest Report

**Source:** `{stats['source']}`

| Metric        | Value |
|---------------|------:|
| Rows ingested | {stats['ingested']:,} |
| Rows rejected | {stats['rejected']:,} |
| Reject rate   | {stats['rejected'] / stats['ingested']:.2%} |

Rejected rows were written to the quarantine table for review.
"""

    create_markdown_artifact(
        key="daily-ingest-report",
        markdown=report,
        description="Row counts and reject rate for the daily ingest.",
    )

The markdown argument is rendered as-is in the UI. The description is metadata shown alongside it. And key is the interesting part, which we come back to below.

Reach for a Markdown artifact when the output is meant to be read by a person — a summary, a data-quality writeup, a “here is what this run did” report. Do not reach for it to trace execution step by step; that is what logs are for.

A table of row counts

When your output is genuinely tabular, create_table_artifact renders it as a sortable table rather than making you hand-build Markdown pipes. It accepts either a list of row dicts or a dict of columns.

from prefect import flow
from prefect.artifacts import create_table_artifact


@flow
def validate_tables():
    row_counts = [
        {"table": "stg_orders",    "rows": 1_284, "expected": ">1000", "ok": True},
        {"table": "stg_customers", "rows": 402,   "expected": ">300",  "ok": True},
        {"table": "stg_returns",   "rows": 0,     "expected": ">0",    "ok": False},
    ]

    create_table_artifact(
        key="table-row-counts",
        table=row_counts,
        description="Post-load row counts per staging table.",
    )

Each dict is a row; the union of keys becomes the columns. This is the artifact to publish when you have per-entity metrics — one row per table, per partition, per source file — and you want them scannable rather than buried in prose.

A task that writes a file has produced something outside Prefect. The run should point at it. create_link_artifact turns that into a clickable link in the UI, so nobody has to reconstruct the S3 path from a log line.

from prefect import flow, task
from prefect.artifacts import create_link_artifact


@task
def export_report(rows: list[dict]) -> str:
    key = "reports/2026-06-29/orders.parquet"
    # ... write to s3://analytics-exports/<key> ...
    return f"s3://analytics-exports/{key}"


@flow
def nightly_export():
    uri = export_report(rows=[...])

    create_link_artifact(
        key="nightly-export-file",
        link=uri,
        link_text="orders.parquet (S3)",
        description="The Parquet file produced by tonight's export.",
    )

link is the URL; link_text is the label shown. This is the artifact that closes the loop between “the pipeline ran” and “here is the thing it made.” An Airflow user would have logged the URI and hoped someone copied it correctly.

A progress bar for a long loop

The one artifact with genuinely different behavior is progress. create_progress_artifact returns an id; update_progress_artifact moves the bar. Unlike the others, it is meant to be updated during the run, not published once at the end.

from prefect import flow
from prefect.artifacts import (
    create_progress_artifact,
    update_progress_artifact,
)


@flow
def backfill(partitions: list[str]):
    progress_id = create_progress_artifact(
        progress=0.0,
        key="backfill-progress",
        description="Reprocessing historical partitions.",
    )

    total = len(partitions)
    for i, partition in enumerate(partitions, start=1):
        process_partition(partition)  # your work
        update_progress_artifact(
            artifact_id=progress_id,
            progress=(i / total) * 100,
        )

progress is a percentage from 0 to 100. While the flow runs, the UI shows a live bar instead of a wall of “processed partition 47/312” log lines. For a long backfill or a big fan-out, this is the difference between a run you can glance at and a run you have to read. Use it whenever “how far along is it?” is a question someone will actually ask.

Keyed artifacts and the versioned timeline

Here is the feature that makes artifacts more than pretty logs. When you pass a key, Prefect treats every artifact with that key as a version of the same logical output. The UI gives you a dedicated page for the key with a timeline of every version across every run that emitted it.

That means key="daily-ingest-report" is not just this run’s report — it is the daily ingest report, with today’s version on top and every prior day beneath it. You can watch the reject rate drift over a week without opening seven separate runs. Keys must be lowercase, hyphen-or-numeric slugs (no spaces, no uppercase), which is the small tax for that history.

Unkeyed artifacts still render, but only on the run that produced them — no cross-run timeline. So the rule is simple: if the output is a recurring thing you will want to compare over time (a report, a metrics table, a validation summary), key it. If it is a genuinely one-off note tied to a single run, you can skip the key.

A data-quality report that builds its own history

The keyed-timeline behavior turns an ordinary validation task into something you cannot easily build in Airflow: a report that accumulates its own history with zero extra infrastructure. Emit a keyed table artifact from a check that runs every day, and after a week you have a page — one URL, key="dq-orders" — where each version sits above the last. No metrics database, no Grafana panel, no separate dq_history table to maintain. The same call that publishes today’s numbers is the thing that archives yesterday’s.

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


@task
def run_checks(rows: list[dict]) -> list[dict]:
    logger = get_run_logger()
    total = len(rows)
    nulls = sum(1 for r in rows if r.get("customer_id") is None)
    dupes = total - len({r["order_id"] for r in rows})
    negatives = sum(1 for r in rows if r.get("amount", 0) < 0)

    results = [
        {"check": "null customer_id", "failed": nulls,     "of": total, "ok": nulls == 0},
        {"check": "duplicate order_id", "failed": dupes,    "of": total, "ok": dupes == 0},
        {"check": "negative amount",    "failed": negatives, "of": total, "ok": negatives == 0},
    ]
    for r in results:
        if not r["ok"]:
            logger.warning("DQ fail: %s — %d/%d rows", r["check"], r["failed"], r["of"])
    return results


@flow
def data_quality(source: str = "orders"):
    rows = load_rows(source)          # your extract
    results = run_checks(rows)

    create_table_artifact(
        key=f"dq-{source}",
        table=results,
        description=f"Data-quality checks for {source}, run {source} nightly.",
    )

Because the key is derived from the source, dq-orders and dq-customers each get their own independent timeline. Open dq-orders a month later and you can watch the duplicate count creep up the day a producer’s dedupe logic broke, without opening thirty runs to find the one where it started. The artifact is the trend line — the run that emitted the newest version is one click away, but the history reads on its own.

Note what the logs are still doing here: the logger.warning per failing check is the event trace (“this specific run saw 12 duplicates”), while the keyed table is the durable, comparable result. Same run, two surfaces, each doing the job it is good at.

Artifact or log? The dividing line

Both go to the API, so the question is what the reader will do with it.

Emit an artifact when the output is a result — something a person is meant to look at, compare, or click: a summary report, a table of metrics, a link to a file, a progress indicator on long work. Artifacts are for the handful of things per run that matter after the run is over.

Write a log when the output is an event in the execution trace — “starting extract,” “retrying connection,” “loaded 1,284 rows.” Logs are for the running narrative you consult when something went wrong. There can be hundreds of them and that is fine; you scroll, filter, and search.

The failure mode from the Airflow world is using logs for both, so the one number anyone cares about is buried in line 340 of a log nobody reads until an incident. Artifacts pull that number to the surface.

Logging: get_run_logger over plain logging

Prefect ships a run-aware logger. Inside a flow or task, call get_run_logger() and use it exactly like a standard library logger — .info(), .warning(), .error(), .debug(), .critical(). What you get for free is that every line is tagged with the flow run, task run, and their names, sent to the Prefect API, and rendered in the UI against the correct run.

from prefect import flow, task
from prefect.logging import get_run_logger


@task
def extract(source: str) -> list[dict]:
    logger = get_run_logger()
    logger.info("Extracting from %s", source)
    rows = fetch(source)
    logger.info("Fetched %d rows", len(rows))
    if not rows:
        logger.warning("Source %s returned no rows", source)
    return rows


@flow
def etl(source: str = "orders_api"):
    logger = get_run_logger()
    logger.info("Starting ETL for %s", source)
    rows = extract(source)
    logger.info("ETL complete: %d rows", len(rows))

get_run_logger() must be called inside run context — within a flow or task while it is executing. Call it at import time or in plain helper code and it raises, because there is no run to attach to.

The contrast is a plain logging.getLogger(__name__). That still writes to your console, but Prefect does not capture it, so it never reaches the API or the UI — the line exists locally and nowhere the platform can see it. For anything you want visible against the run, use the run logger.

Log levels and PREFECT_LOGGING_LEVEL

The levels are the standard DEBUG < INFO < WARNING < ERROR < CRITICAL. The threshold is a setting, not something you hard-code: PREFECT_LOGGING_LEVEL controls the root level Prefect emits, defaulting to INFO. Drop it to DEBUG when you are chasing something:

export PREFECT_LOGGING_LEVEL="DEBUG"

The broader PREFECT_LOGGING_* family tunes the rest — most usefully PREFECT_LOGGING_LEVEL for the floor and the per-logger settings loaded from Prefect’s logging config. You can point PREFECT_LOGGING_SETTINGS_PATH at your own YAML if you need to reshape formatters or handlers, but for the common case the level env var is all you touch. Because it is a setting, it obeys the same profile precedence as everything else in Prefect — set it per profile, per deployment environment, or per shell.

log_prints: capture stray print statements

Not every useful line comes from a logger. Third-party code, quick scripts, and half the internet’s example snippets just print(). Prefect can capture those and route them through the run logger as INFO logs, so they land in the UI like everything else. Turn it on with log_prints=True:

from prefect import flow, task


@task
def crunch(n: int):
    print(f"crunching {n} items")  # captured as an INFO log


@flow(log_prints=True)
def pipeline():
    crunch(100)

The scoping rule matters and mirrors how Prefect handles most configuration: log_prints set on a flow is inherited by its tasks, unless a task overrides it. So @flow(log_prints=True) captures prints in the flow and in every task it calls — but a @task(log_prints=False) inside that flow opts back out, and vice versa. Set it broadly at the flow level and override the exceptions.

log_prints is a convenience, not a substitute for real logging. It captures at INFO with no level control and no structure. Use it to catch output from code you do not own; use get_run_logger() for output you write yourself.

Capturing a noisy third-party logger

By default Prefect only ships its own loggers to the API. A library that logs through its own logger — httpx, boto3, sqlalchemy — writes to your console but not to the run. When you need that library’s output attached to the run (debugging a flaky API call, say), add its logger name to PREFECT_LOGGING_EXTRA_LOGGERS:

export PREFECT_LOGGING_EXTRA_LOGGERS="httpx,sqlalchemy.engine"

It is a comma-separated list of logger names. Once listed, those loggers are captured and their records flow to the API alongside your run logs, at whatever level they emit. This is the escape hatch for “the failure is inside a library and I need to see what it saw.” Leave it off in normal operation — a chatty client logger at DEBUG will flood your run logs — and switch it on when you are actually debugging.

How run logs reach the UI

The mechanics are worth understanding so you know what you are paying for. When you log through the run logger (or via captured prints / extra loggers), Prefect batches those records and sends them to the API, which stores them against the flow run and task run. The UI reads them back, which is why you can open a run and read its logs without SSHing anywhere — the log lives in the API, not on the worker’s disk.

You can turn API shipping off with PREFECT_LOGGING_TO_API_ENABLED=false, in which case logs stay local (console only) and the UI shows nothing for that run. That is occasionally what you want for a noisy local dev loop, but in any real deployment you leave it on — the whole point is that logs outlive the ephemeral worker that produced them.

Logging an exception

When a task fails, the state carries the traceback — but often you want the failing context logged before you re-raise or swallow, with the stack trace attached to the run log rather than only the state. logger.exception does exactly that: it logs at ERROR and appends the current traceback, and it is only valid inside an except block.

from prefect import task
from prefect.logging import get_run_logger


@task
def call_pricing_api(order_id: str) -> dict:
    logger = get_run_logger()
    try:
        return pricing_client.get(order_id)
    except TimeoutError:
        # ERROR-level log with the full traceback attached to the run
        logger.exception("Pricing API timed out for order %s", order_id)
        raise

The message you pass is the human summary; the traceback is appended automatically, so you do not interpolate the exception yourself. Re-raising after logging keeps Prefect’s retry and state machinery intact — you have added visibility without changing control flow. Use logger.exception in any except where the failure is worth a first-class, searchable ERROR line in the run log, which is most of them.

A custom logging config and structured fields

For the common case, PREFECT_LOGGING_LEVEL and the extra-loggers list are all you touch. When you need to reshape the output itself — a JSON formatter for a log aggregator, a quieter handler for one chatty logger, extra fields on every record — point PREFECT_LOGGING_SETTINGS_PATH at your own logging.yml. Prefect ships a default config internally; your file is merged over it, so you override only the pieces you care about.

# logging.yml — a JSON formatter for the API-bound run logs
version: 1
disable_existing_loggers: false

formatters:
  json:
    format: '{"ts": "%(asctime)s", "level": "%(levelname)s", "flow_run": "%(flow_run_id)s", "msg": "%(message)s"}'

handlers:
  api:
    level: INFO
    class: prefect.logging.handlers.APILogHandler

loggers:
  prefect.flow_runs:
    level: INFO
    handlers: [api]
  prefect.task_runs:
    level: INFO
    handlers: [api]
export PREFECT_LOGGING_SETTINGS_PATH="/etc/prefect/logging.yml"

The %(flow_run_id)s token is available because Prefect injects run context onto every record it handles — the same mechanism that lets the UI file a line against the right run. You can lean on that to attach your own structured fields at the call site with the standard-library extra= dict, then reference them in a formatter:

logger = get_run_logger()
logger.info(
    "Loaded batch",
    extra={"rows": len(clean), "source": source, "batch_id": batch_id},
)

With a formatter that names %(rows)s/%(source)s/%(batch_id)s, those values land in each record as first-class columns rather than being smeared into the message string — which is what makes the difference between logs you grep and logs you actually query in whatever aggregator sits downstream. This is the seam where Prefect’s run-aware logging plugs straight into the standard-library logging model you already know: same logging.yml schema, same extra= convention, with run identity threaded through for free.

A worked ETL: structured logs plus a summary artifact

Putting both features together in the shape you will actually use — logs narrate the run, one keyed artifact publishes the result:

from prefect import flow, task
from prefect.logging import get_run_logger
from prefect.artifacts import create_markdown_artifact


@task
def extract(source: str) -> list[dict]:
    logger = get_run_logger()
    logger.info("Extract: reading from %s", source)
    rows = fetch(source)
    logger.info("Extract: %d rows", len(rows))
    return rows


@task
def transform(rows: list[dict]) -> tuple[list[dict], int]:
    logger = get_run_logger()
    clean = [r for r in rows if is_valid(r)]
    rejected = len(rows) - len(clean)
    if rejected:
        logger.warning("Transform: dropped %d invalid rows", rejected)
    logger.info("Transform: %d clean rows", len(clean))
    return clean, rejected


@task
def load(rows: list[dict]) -> None:
    logger = get_run_logger()
    logger.info("Load: writing %d rows", len(rows))
    write_to_warehouse(rows)


@flow(log_prints=True)
def daily_etl(source: str = "orders_api"):
    logger = get_run_logger()
    logger.info("Starting daily ETL for %s", source)

    raw = extract(source)
    clean, rejected = transform(raw)
    load(clean)

    reject_rate = rejected / len(raw) if raw else 0.0
    create_markdown_artifact(
        key="daily-etl-summary",
        markdown=f"""# Daily ETL Summary — `{source}`

| Metric        | Value |
|---------------|------:|
| Rows read     | {len(raw):,} |
| Rows loaded   | {len(clean):,} |
| Rows rejected | {rejected:,} |
| Reject rate   | {reject_rate:.2%} |
""",
        description="Row counts and reject rate for the daily ETL run.",
    )
    logger.info("Daily ETL complete")

Open a run and you get the full narrative in the logs — extract counts, the transform warning when rows were dropped, the load. Open the daily-etl-summary artifact and you get today’s numbers on top of every prior day’s, one scannable timeline of how the pipeline’s health is trending. The logs answer “what happened in this run?”; the keyed artifact answers “how is this pipeline doing over time?” — two questions Airflow makes you answer with the same overloaded log stream.

Final thoughts

Logs and artifacts are not competing features; they are a division of labor. Logs are the execution trace — high volume, event-shaped, consulted when something breaks. Artifacts are the published result — low volume, meant to be read, and (when keyed) versioned into a timeline you can watch across runs. Coming from Airflow, the instinct is to cram everything into the log because it is the only surface you had. Prefect’s artifacts give the numbers-that-matter their own home, and the discipline of choosing which is which is what keeps your runs legible six months later. Log the narrative; publish the result; key anything you will want to compare.

Next: Testing Prefect Flows

Comments