data-juicer

Processing Data

This guide covers running data processing pipelines with Data-Juicer—both CLI and Python API.

If you haven’t run your first pipeline yet, start with the Quick Start. For the full parameter list, see Global Configuration Reference.

CLI

Basic Usage

dj-process --config my-recipe.yaml

Data-Juicer reads the recipe, executes operators in the listed order by default, and writes results to export_path.

Command-Line Overrides

Override any recipe parameter without modifying the YAML:

dj-process --config recipe.yaml --np 8 --export_path ./out/result.parquet
dj-process --config recipe.yaml --language_id_score_filter.lang=en

Auto-Install Operator Dependencies

dj-install --config my-recipe.yaml

The tool scans source files for the operators in the recipe and preinstalls the dependencies it identifies. See Installation for the scan scope and environment preparation.

Python API

The Python API provides finer control than YAML recipes—ideal for training scripts, notebooks, or automated pipelines.

Option 1: Load a Recipe

from data_juicer.config import init_configs
from data_juicer.core import DefaultExecutor

cfg = init_configs(args=['--config', 'my-recipe.yaml'])
executor = DefaultExecutor(cfg)
dataset = executor.run()

If you already hold a dataset object in memory (e.g. assembled or sampled upstream), you can skip the recipe’s data source and reuse only its operator pipeline:

dataset = executor.run(dataset=my_dataset, skip_export=True)

Option 2: Instantiate Operators from Config

No YAML needed—assemble an operator chain in Python:

from data_juicer.ops import load_ops
from data_juicer.core import NestedDataset

# Load ops from dict config (same format as YAML process list)
ops = load_ops([
    {'language_id_score_filter': {'lang': 'en', 'min_score': 0.8}},
    {'text_length_filter': {'min_len': 10, 'max_len': 50000}},
    {'document_minhash_deduplicator': {'tokenization': 'space', 'window_size': 5}},
])

dataset = NestedDataset(NestedDataset.from_json('my-data.jsonl'))
dataset = dataset.process(ops)

Option 3: Fine-Grained Single-Operator Control

When you need conditional logic, loops, or intermediate inspection:

from data_juicer.ops.filter import LanguageIDScoreFilter, TextLengthFilter
from data_juicer.ops.deduplicator import DocumentMinhashDeduplicator
from data_juicer.core import NestedDataset

dataset = NestedDataset(NestedDataset.from_json('my-data.jsonl'))

# Step 1: Language filter
lang_filter = LanguageIDScoreFilter(lang='en', min_score=0.8)
dataset = lang_filter.run(dataset=dataset)
print(f"After language filter: {len(dataset)} samples")

# Step 2: Conditional dedup — only when dataset is large
if len(dataset) > 10000:
    dedup = DocumentMinhashDeduplicator(tokenization='space', window_size=5)
    dataset = dedup.run(dataset=dataset)
    print(f"After dedup: {len(dataset)} samples")

# Step 3: Length filter
length_filter = TextLengthFilter(min_len=10, max_len=50000)
dataset = length_filter.run(dataset=dataset)

Option 4: Dynamic Operator Composition

Choose operators programmatically based on data characteristics—useful for automated pipelines:

from data_juicer.ops import load_ops
from data_juicer.core import NestedDataset

dataset = NestedDataset(NestedDataset.from_json('input.jsonl'))

# Inspect data to decide processing strategy
sample = dataset[0]
ops_config = []

# Add language filter if text field exists
if 'text' in sample:
    ops_config.append({'language_id_score_filter': {'lang': 'en', 'min_score': 0.5}})

# Add image filter if images present
if 'images' in sample and sample['images']:
    ops_config.append({'image_shape_filter': {'min_width': 256, 'min_height': 256}})

# Common cleaning
ops_config.append({'clean_html_mapper': {}})
ops_config.append({'text_length_filter': {'min_len': 10}})

ops = load_ops(ops_config)
dataset = dataset.process(ops)

API Models

Use an API operator such as extract_keyword_mapper to process text with a hosted model. Set api_model to the model name supplied by your service and configure the connection in model_params.

OpenAI-Compatible Services

Install the API dependencies in your activated environment:

uv pip install "py-data-juicer[ai_services]"

Set the service URL and API key in your shell:

export OPENAI_BASE_URL="https://api.openai.com/v1"
export OPENAI_API_KEY="your-api-key"

To use another OpenAI-compatible service, replace the URL and key with that service’s values. Save this recipe as extract.yaml and set dataset_path to a JSONL file with a text field:

dataset_path: ./input.jsonl
export_path: ./outputs/extracted.jsonl
keep_stats_in_res_ds: true

process:
  - extract_keyword_mapper:
      api_model: gpt-4o-mini
      model_params:
        api_backend: openai_compatible
      sampling_params:
        temperature: 0

Run dj-process --config extract.yaml. The operator saves the extracted keywords under __dj__meta__.keyword; keep_stats_in_res_ds: true retains that metadata in the output.

openai_compatible is the default backend. You can also supply base_url and api_key in model_params; these values take precedence over the corresponding environment variables.

LiteLLM

To route requests through LiteLLM, set model_params.api_backend: litellm and use a provider-qualified model name. For example, replace the operator entry in the recipe above with:

process:
  - extract_keyword_mapper:
      api_model: openai/gpt-4o-mini
      model_params:
        api_backend: litellm
      sampling_params:
        temperature: 0

Configure credentials for the selected provider. The OpenAI example uses OPENAI_API_KEY; other providers use their own LiteLLM authentication settings. You can supply api_key and base_url in model_params when the provider requires them.

Choose an operator that supports the task you need. The API backend supports chat, embedding, and Responses endpoints; operators that expose api_endpoint can select the corresponding endpoint. Use sampling_params for request options such as temperature and model_params for the backend, service URL, and credentials.

Operator Execution Order

By default, operators run top-to-bottom sequentially. Order matters:

  1. Cheap filters first: Text length, language ID—reduce sample count early

  2. Dedup in the middle: Requires global state; run after initial filtering

  3. Expensive operators last: GPU inference only sees the filtered subset

process:
  # Cheap
  - text_length_filter: { min_len: 10, max_len: 50000 }
  - language_id_score_filter: { lang: en, min_score: 0.5 }
  # Dedup
  - document_minhash_deduplicator: { tokenization: space, window_size: 5 }
  # Expensive
  - clean_html_mapper: {}
  - perplexity_filter: { lang: en, max_ppl: 1500 }

Performance Tuning

Op Fusion

Fuses compatible operators to reduce repeated processing. Both default and Ray execution paths support fusion. Throughput gains depend on the recipe and data; measure them for your workload:

op_fusion: true
fusion_strategy: probe   # probe: group and sort by measured speed; greedy: order by fusion group

Fusion groups compatible operators and may change their order in the recipe. With probe, the default executor and standard Analyzer use the first 1,000 rows of the current dataset by default, or all rows of a smaller dataset. Each operator runs the probe on copies of this batch, with one copy per runtime process. Ray executors arrange operators by fusion group.

GPU Mapper Fusion

Fuses consecutive GPU Mappers into one GPU pass:

op_fusion: true
mapper_fusion: true
adaptive_batch_size: true

The default executor uses adaptive_batch_size to probe and adjust batch sizes for batched operators.

Sampling Dry Run

For a source with at least 1,000 rows, validate a recipe on 1,000 samples with the default executor by replacing its dataset_path with a structured dataset configuration. Keep its process list:

dataset:
  max_sample_num: 1000
  configs:
    - type: local
      path: path/to/your/dataset.jsonl

Replace the path with your dataset. max_sample_num sets the sample count; for a small trial, choose a count no larger than the source. Larger budgets are filled by repeating samples.

Sampling runs after data loading. To also reduce input reads, prepare a small sample file and use it as the input. Remove max_sample_num to process the full dataset.

Checkpointing & Resumption

use_checkpoint: true

The default executor locates checkpoints by job_id and the working-directory base. Specify a fixed ID on the first run. To resume after interruption, run the same command with the same input, recipe, and working-directory configuration:

dj-process --config your-recipe.yaml --use_checkpoint true --job_id recipe-checkpoint

Checkpoints live in the resolved <cfg.work_dir>/ckpt, where work_dir includes job_id. use_checkpoint disables data caching and is mutually exclusive with op_fusion.

For ray_partitioned, finer strategies are available; resume with --resume <job_id>. See Global Config.

Tracing & Debugging

open_tracer: true
trace_num: 10

Writes before/after comparisons per operator. See Tracing.

Next Steps

↵ to select ↑↓ to navigate esc to close