Implementation:Datajuicer Data juicer JobSnapshot

Knowledge Sources	Datajuicer_Data_juicer
Domains	Job Management, Monitoring, Diagnostics
Last Updated	2026-02-14 16:00 GMT

Overview

Processing snapshot analyzer that reconstructs the complete state of a Data-Juicer job by parsing event logs and DAG structures to produce comprehensive snapshots of partition and operation statuses.

Description

The snapshot module provides detailed diagnostic analysis for Data-Juicer processing jobs. Key components include:

Data Classes:

• ProcessingStatus -- Enum with states: NOT_STARTED, IN_PROGRESS, COMPLETED, FAILED, CHECKPOINTED.
• OperationStatus -- Tracks individual operation state including timing, row counts, checkpoint time, and error messages.
• PartitionStatus -- Tracks per-partition state including creation/processing times, current/completed/failed/checkpointed operations.
• JobSnapshot -- Complete job state aggregation with partition/operation statuses, DAG structure, checkpoint information, and resumability flag.

Analyzer:

• ProcessingSnapshotAnalyzer -- The main analyzer class that:
  • Loads events from events_*.jsonl files with backward compatibility.
  • Loads DAG execution plans from dag_execution_plan.json.
  • Loads job summaries from job_summary.json.
  • Reconstructs processing state by iterating through events and building partition/operation status dictionaries.
  • Determines overall job status from partition status counts.
  • Calculates statistics (completed/failed/in-progress counts for both partitions and operations).
  • Generates JobSnapshot objects with timing, checkpointing, and resumability information.
  • Converts snapshots to JSON-serializable dictionaries with comprehensive progress tracking, including per-partition/operation progress percentages and formatted durations.

Convenience Functions:

• create_snapshot -- One-liner to create a snapshot from a work directory.
• main -- CLI entry point supporting both human-readable and JSON output formats.

Usage

Use this module for job diagnostics, progress monitoring, failure analysis, and determining whether a job can be resumed from checkpoints. It can be invoked as a command-line tool or used programmatically for integration with monitoring systems.

Code Reference

Source Location

• Repository: Datajuicer_Data_juicer
• File: data_juicer/utils/job/snapshot.py

Signature

class ProcessingStatus(Enum):
    NOT_STARTED = "not_started"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    FAILED = "failed"
    CHECKPOINTED = "checkpointed"

@dataclass
class OperationStatus:
    operation_name: str
    operation_idx: int
    status: ProcessingStatus
    # ... timing, row counts, checkpoint, error fields

@dataclass
class PartitionStatus:
    partition_id: int
    status: ProcessingStatus
    # ... timing, operations lists

@dataclass
class JobSnapshot:
    job_id: str
    # ... timing, counts, statuses, DAG, checkpoint info

class ProcessingSnapshotAnalyzer:
    def __init__(self, work_dir: str): ...
    def generate_snapshot(self) -> JobSnapshot: ...
    def to_json_dict(self, snapshot: JobSnapshot) -> Dict: ...

def create_snapshot(work_dir: str, detailed: bool = False) -> JobSnapshot: ...

Import

from data_juicer.utils.job.snapshot import (
    ProcessingSnapshotAnalyzer, JobSnapshot, create_snapshot
)

I/O Contract

Inputs

Name	Type	Required	Description
work_dir	str	Yes	Path to the Data-Juicer work directory containing event logs and job summary.
detailed	bool	No	Whether to include detailed output (reserved for future use).

Outputs

Name	Type	Description
snapshot	JobSnapshot	Complete processing snapshot with partition/operation statuses, timing, and checkpoint info.
json_dict	Dict	JSON-serializable dictionary with job_info, overall_progress, partition_progress, operation_progress, dag_structure, file_paths, and metadata.

Usage Examples

from data_juicer.utils.job.snapshot import (
    create_snapshot, ProcessingSnapshotAnalyzer
)

# Quick snapshot creation
snapshot = create_snapshot("/outputs/partition-checkpoint-eventlog/job_id")
print(f"Status: {snapshot.overall_status.value}")
print(f"Partitions: {snapshot.completed_partitions}/{snapshot.total_partitions}")
print(f"Resumable: {snapshot.resumable}")

# Detailed JSON output
analyzer = ProcessingSnapshotAnalyzer("/outputs/partition-checkpoint-eventlog/job_id")
snapshot = analyzer.generate_snapshot()
json_data = analyzer.to_json_dict(snapshot)

# Access progress details
print(f"Overall: {json_data['overall_progress']['overall_percentage']:.1f}%")
for pid, partition in json_data['partition_progress'].items():
    print(f"  Partition {pid}: {partition['status']} "
          f"({partition['progress_percentage']:.0f}%)")

# CLI usage
python -m data_juicer.utils.job.snapshot /path/to/work_dir
python -m data_juicer.utils.job.snapshot /path/to/work_dir --human-readable

Related Pages

• Datajuicer_Data_juicer_JobUtils -- Common job utilities that provide foundational state loading