Implementation:Datajuicer Data juicer Job Monitor
| Knowledge Sources | |
|---|---|
| Domains | Data_Processing, Job_Management |
| Last Updated | 2026-02-14 16:00 GMT |
Overview
Concrete tool for monitoring and displaying real-time progress of Data-Juicer processing jobs provided by Data-Juicer.
Description
JobProgressMonitor is a class that tracks and displays detailed progress information for Data-Juicer jobs. It uses JobUtils to load job summaries, dataset mappings, and partition statuses, then formats this data into a comprehensive console report showing job overview (status, dataset, sample counts, timing), overall progress (percentage, partition counts, estimated remaining time), per-partition status (sample counts, current and completed operations, checkpoints), and optionally detailed operation metrics (duration, throughput, reduction ratio). The companion function show_job_progress provides a convenient one-call interface. The module also supports a CLI watch mode that continuously refreshes progress at a configurable interval.
Usage
Use when you need to monitor the progress of a long-running Data-Juicer data processing job, either interactively from the command line or programmatically to retrieve progress data as a dictionary.
Code Reference
Source Location
- Repository: Datajuicer_Data_juicer
- File: data_juicer/utils/job/monitor.py
Signature
class JobProgressMonitor:
def __init__(self, job_id: str,
base_dir: str = "outputs/partition-checkpoint-eventlog"):
"""
Initialize the job progress monitor.
Args:
job_id: The job ID to monitor.
base_dir: Base directory containing job outputs.
"""
def display_progress(self, detailed: bool = False):
"""Display job progress information to the console."""
def get_progress_data(self) -> Dict[str, Any]:
"""Get progress data as a dictionary for programmatic use."""
def show_job_progress(
job_id: str,
base_dir: str = "outputs/partition-checkpoint-eventlog",
detailed: bool = False
) -> Dict[str, Any]:
"""
Utility function to show job progress.
Args:
job_id: The job ID to monitor.
base_dir: Base directory containing job outputs.
detailed: Whether to show detailed operation information.
Returns:
Dictionary containing all progress data.
"""
Import
from data_juicer.utils.job.monitor import JobProgressMonitor, show_job_progress
I/O Contract
Inputs
| Name | Type | Required | Description |
|---|---|---|---|
| job_id | str | Yes | The unique identifier of the job to monitor |
| base_dir | str | No | Base directory containing job outputs. Default: "outputs/partition-checkpoint-eventlog" |
| detailed | bool | No | Whether to show detailed operation metrics. Default: False |
Outputs
| Name | Type | Description |
|---|---|---|
| console output | str | Formatted progress report printed to stdout |
| progress_data | Dict[str, Any] | Dictionary containing job_id, job_summary, dataset_mapping, partition_status, and overall_progress |
Usage Examples
Quick Progress Check
from data_juicer.utils.job.monitor import show_job_progress
# Display progress for a specific job
progress = show_job_progress("20250728_233517_510abf")
# Display detailed progress with per-operation metrics
progress = show_job_progress("20250728_233517_510abf", detailed=True)
Programmatic Monitoring
from data_juicer.utils.job.monitor import JobProgressMonitor
monitor = JobProgressMonitor("20250728_233517_510abf")
data = monitor.get_progress_data()
print(f"Progress: {data['overall_progress']['progress_percentage']:.1f}%")
print(f"Completed: {data['overall_progress']['completed_partitions']} partitions")
CLI Watch Mode
# From the command line with continuous refresh
python -m data_juicer.utils.job.monitor 20250728_233517_510abf --watch --interval 10 --detailed