The Pipeline Runner
Relevant source files
The following files were used as context for generating this wiki page:The Pipeline Runner is the core execution engine of the DocETL framework. Its primary responsibility is to interpret a pipeline configuration, process datasets through a series of defined operations, and manage the overall workflow from data loading to result saving. The main implementation is the DSLRunner class, which can be invoked directly or, more commonly, through the high-level Pipeline API class.
The runner orchestrates the entire Extract, Transform, Load (ETL) process. It handles loading initial datasets, executing each step defined in the pipeline configuration, applying operations like map, reduce, and filter, managing intermediate data, and saving the final output. It also provides an entry point for pipeline optimization.
Core Components
The pipeline execution is managed by two main classes: Pipeline and DSLRunner.
The Pipeline API Class
The Pipeline class, defined in docetl/api.py, serves as the primary user-facing interface for programmatically defining and executing pipelines. It acts as a high-level wrapper around the DSLRunner.
The constructor for the Pipeline class accepts all the necessary components to define a workflow:
| Parameter | Type | Description |
|---|---|---|
name | str | The name of the pipeline. |
datasets | dict[str, Dataset] | A dictionary of input datasets. |
operations | list[OpType] | A list of all operations used in the pipeline. |
steps | list[PipelineStep] | The sequence of steps that define the pipeline's flow. |
output | PipelineOutput | The output configuration. |
parsing_tools | list[ParsingTool | Callable] | Custom functions for data parsing. |
default_model | str | None | The default LLM to use for operations. |
optimizer_config | dict[str, Any] | Configuration for the pipeline optimizer. |
Sources: docetl/api.py:116-141
The Pipeline class delegates the core execution and optimization tasks to the DSLRunner.
This diagram shows that the Pipeline class is the main API, which in turn uses the DSLRunner to perform the actual work of running and optimizing the pipeline.
The DSLRunner Class
The DSLRunner is the engine that executes the pipeline defined in a configuration dictionary. It can be instantiated directly or via its from_yaml class method.
Sources: docetl/runner.py
Key Methods
| Method | Description | Source |
|---|---|---|
from_yaml(cls, yaml_file: str, ...) | A class method to create a DSLRunner instance from a YAML configuration file. | [docetl/runner.py:73-83](https://github.com/odra28/docetl/blob/HEAD/docetl/runner.py#L73-L83) |
load_run_save() | The main execution method. It loads datasets, runs all pipeline steps, saves the output, and returns the total cost. | [docetl/runner.py:100-116](https://github.com/odra28/docetl/blob/HEAD/docetl/runner.py#L100-L116) |
optimize(resume: bool, ...) | Invokes the pipeline optimizer to rewrite the configuration for better performance or lower cost. | [docetl/runner.py:85-98](https://github.com/odra28/docetl/blob/HEAD/docetl/runner.py#L85-L98) |
Execution Flow
The primary execution flow is handled by the load_run_save method within the DSLRunner. This method orchestrates the entire process from start to finish.
The following diagram illustrates the sequence of actions taken by the load_run_save method.
Sources: docetl/runner.py:100-116
Data Parsing
The runner supports data parsing during the initial data loading phase. This is configured within the datasets section of the pipeline configuration. The parsing key specifies a list of parsing operations to be applied.
A built-in parsing function, txt_to_string, is available to read content from a text file path specified in the input data.
Sources: tests/basic/test_pipeline_with_parsing.py:27-30
The framework also supports custom parsing functions. A Python callable can be provided in the parsing_tools list of the Pipeline constructor. The function name is then referenced in the parsing configuration.
An example of a custom parsing function custom_exploder:
def custom_exploder(doc: dict) -> list[dict]:
text = doc["text"]
return [{"text": t} for t in text]
pipeline = Pipeline(
# ...
datasets={
"input_data": Dataset(
type="file",
path=tmp_input.name,
parsing=[
{
"function": "custom_exploder",
}
],
)
},
parsing_tools=[custom_exploder],
# ...
)
Sources: tests/basic/test_pipeline_with_parsing.py:43-45, tests/basic/test_pipeline_with_parsing.py:118-132
Pipeline Optimization
The DSLRunner provides an optimize() method, which is exposed through the Pipeline.optimize() API. This method uses an Optimizer to rewrite the pipeline configuration. The goal is to improve efficiency, reduce cost, or increase accuracy.
The Pipeline.optimize() method returns a new, optimized Pipeline instance.
# docetl/api.py
def optimize(
self,
max_threads: int | None = None,
resume: bool = False,
save_path: str | None = None,
) -> "Pipeline":
# ...
runner = DSLRunner(
config,
# ...
)
optimized_config, _ = runner.optimize(
resume=resume,
return_pipeline=False,
save_path=save_path,
)
updated_pipeline = Pipeline(...)
updated_pipeline._update_from_dict(optimized_config)
return updated_pipeline
Sources: docetl/api.py:165-192
The optimization process can involve various strategies, such as injecting cheaper pre-filters (cascade_filtering) or consolidating redundant operations (arbitrary_rewrite). These strategies are implemented as Directives.
Sources: docetl/reasoning_optimizer/directives/cascade_filtering.py, docetl/reasoning_optimizer/directives/arbitrary_rewrite.py
Usage Example
The Pipeline class provides a fluent API for defining and running a pipeline. The run() method demonstrates the complete process of instantiating the DSLRunner and executing the pipeline.
This sequence diagram illustrates how a call to pipeline.run() is delegated to the DSLRunner to execute the load_run_save workflow.
Sources: docetl/api.py:194-210
A test case from the repository shows a practical example of using the DSLRunner with a YAML file:
# tests/basic/test_pipeline_with_parsing.py
def test_pipeline_with_parsing(config_file):
# ... (setup temporary files)
# Create and run the DSLRunner
runner = DSLRunner.from_yaml(config_file)
total_cost = runner.load_run_save()
# Check if the output file was created
assert os.path.exists(output_file.name), "Output file was not created"
# ... (verify output)
Sources: tests/basic/test_pipeline_with_parsing.py:59-79
Frontend Integration
The frontend DocWrangler UI interacts with the pipeline execution logic, primarily through YAML configurations. Components like NaturalLanguagePipelineDialog and TutorialsDialog use a restoreFromYAML hook. This function takes a pipeline configuration (often generated by an AI assistant or from a template), creates a Blob, and applies it to the workspace. This YAML is the same format that the DSLRunner consumes on the backend.
// website/src/components/NaturalLanguagePipelineDialog.tsx
// Create pipeline file object with the generated name
const pipelineBlob = new Blob([updatedYaml], {
type: "application/x-yaml",
});
const pipelineFile: FileType = {
name: `${pipelineName}.yaml`,
path: `${pipelineName}.yaml`,
type: "pipeline-yaml",
blob: pipelineBlob,
};
// Restore pipeline from the YAML
await restoreFromYAML(pipelineFile);
Sources: website/src/components/NaturalLanguagePipelineDialog.tsx:668-683, website/src/components/TutorialsDialog.tsx:162-177
This demonstrates that the DSLRunner and its YAML-based configuration serve as the bridge between the interactive UI and the backend processing engine.
Summary
The Pipeline Runner, centered around the DSLRunner class, is the operational heart of DocETL. It provides a robust mechanism for executing complex, multi-step document processing workflows defined in a declarative YAML format. It is abstracted by the user-friendly Pipeline Python API, which simplifies programmatic pipeline definition and execution. The runner's architecture supports data parsing, step-by-step execution, intermediate data management, and an extensible optimization framework, making it a versatile and powerful engine for LLM-based data pipelines.