Job Management¶

DataJuicer provides utilities for monitoring and managing processing jobs.

Processing Snapshot¶

Analyze job status from event logs and DAG structure.

# JSON output
python -m data_juicer.utils.job.snapshot /path/to/job_dir

# Human-readable output
python -m data_juicer.utils.job.snapshot /path/to/job_dir --human-readable

Output includes:

Job status and progress percentage
Partition completion counts
Operation metrics
Checkpoint coverage
Timing information

Execution Plan and DAG Monitoring¶

use_dag controls execution-plan generation and DAG monitoring, which produce the DAG structure the snapshot analyzer reads.

use_dag: null   # null keeps the executor default

With null, DAG monitoring is enabled for ray and ray_partitioned and disabled for default. Set true to enable it or false to disable it.

Resource-Aware Partitioning¶

The system automatically optimizes partition sizes based on cluster resources and data characteristics.

partition:
  mode: "auto"
  target_size_mb: 256  # Target partition size (configurable)

The optimizer:

Detects CPU, memory, and GPU resources
Samples data to determine modality and memory usage
Calculates partition size targeting the configured size (default 256MB)
Determines optimal worker count

Logging¶

Logs are organized per job. The layout below uses log.txt as an example; CLI runs generate filenames from the export path and timestamp.

{job_dir}/
├── events_{timestamp}.jsonl   # Machine-readable events
├── logs/
│   ├── log.txt                # Main log
│   ├── log_DEBUG.txt          # Debug logs
│   ├── log_ERROR.txt          # Error logs
│   └── log_WARNING.txt        # Warning logs
└── job_summary.json           # Summary (on completion)

Set event_log_dir in your recipe to choose the application log directory; it defaults to <work_dir>/logs. Machine-readable events_*.jsonl files are saved directly in <work_dir> for job tracking.

To configure application logging in Python:

from data_juicer.utils.logger_utils import setup_logger

setup_logger(
    save_dir="./outputs",
    filename="log.txt",
    level="INFO",
    redirect=False
)

setup_logger() uses save_dir and filename to configure log output and writes separate files for each log level. Configure rotation and retention through your application’s logging integration.

API Reference¶

ProcessingSnapshotAnalyzer¶

from data_juicer.utils.job.snapshot import ProcessingSnapshotAnalyzer

analyzer = ProcessingSnapshotAnalyzer(job_dir)
snapshot = analyzer.generate_snapshot()
json_data = analyzer.to_json_dict(snapshot)

ResourceDetector¶

from data_juicer.core.executor.partition_size_optimizer import ResourceDetector

local = ResourceDetector.detect_local_resources()
cluster = ResourceDetector.detect_ray_cluster()
workers = ResourceDetector.calculate_optimal_worker_count()

PartitionSizeOptimizer¶

from data_juicer.core.executor.partition_size_optimizer import PartitionSizeOptimizer

optimizer = PartitionSizeOptimizer(cfg)
recommendations = optimizer.get_partition_recommendations(dataset, pipeline)

Troubleshooting¶

Check job status:

python -m data_juicer.utils.job.snapshot /path/to/job

Analyze events:

cat /path/to/job/events_*.jsonl | head -20

Check resources:

from data_juicer.core.executor.partition_size_optimizer import ResourceDetector
print(ResourceDetector.detect_local_resources())