Implementation:Ucbepic Docetl DSLRunner Save

Knowledge Sources	DocETL DocETL Docs
Domains	Data_Engineering, Observability
Last Updated	2026-02-08 01:40 GMT

Overview

Concrete tool for saving pipeline results and tracking execution output provided by DSLRunner and ThreadSafeConsole.

Description

The DSLRunner.save() method writes final pipeline output to JSON or CSV files. It works alongside ThreadSafeConsole which captures all console output (cost tracking, operation progress, execution summaries) in a thread-safe buffer. The runner also supports intermediate checkpointing via _save_checkpoint() and _load_from_checkpoint_if_exists().

Usage

The save method is called automatically at the end of load_run_save(). Use ThreadSafeConsole.get_output() to retrieve captured console output for external consumption (e.g., streaming to WebSocket clients in the playground).

Code Reference

Source Location

• Repository: docetl
• File: docetl/runner.py (save: L541-571), docetl/console.py (ThreadSafeConsole: L74-96)

Signature

class DSLRunner:
    def save(self, data: list[dict]) -> None:
        """Save final pipeline output to configured path (JSON or CSV)."""

class ThreadSafeConsole(Console):
    def __init__(self, *args, **kwargs):
        """Thread-safe Rich console with StringIO buffer capture."""

    def get_output(self) -> str:
        """Get buffered output, processing carriage returns from Rich spinners."""

Import

from docetl.runner import DSLRunner
from docetl.console import ThreadSafeConsole

I/O Contract

Inputs

Name	Type	Required	Description
data	list[dict]	Yes	Final pipeline output records

Outputs

Name	Type	Description
output file	JSON or CSV	Written to path configured in pipeline output section
console output	str	Execution logs, cost summaries, operation progress

Usage Examples

Automatic Save (via load_run_save)

from docetl.runner import DSLRunner

runner = DSLRunner(config)
total_cost = runner.load_run_save()  # Automatically calls save()

Manual Save

runner = DSLRunner(config)
runner.load()
output, _, _ = runner.last_op_container.next()
runner.save(output)  # Explicitly save results

Related Pages

Implements Principle

• Principle:Ucbepic_Docetl_Output_Management