Implementation:Datajuicer Data juicer Job Stopper
| Knowledge Sources | |
|---|---|
| Domains | Data_Processing, Job_Management |
| Last Updated | 2026-02-14 16:00 GMT |
Overview
Concrete tool for gracefully stopping running Data-Juicer jobs by discovering and terminating their processes provided by Data-Juicer.
Description
JobStopper is a class that stops Data-Juicer jobs by reading event logs to find associated process and thread IDs, then terminating those processes. It first attempts graceful termination (SIGTERM) with a configurable timeout, falling back to force-kill (SIGKILL) if the process does not exit in time. After termination, it cleans up job resources and updates the job summary status to "stopped" with a timestamp and stop reason. The companion function stop_job provides a convenient one-call interface. The module also supports a CLI with options for listing all jobs, force termination, and verbose logging.
Usage
Use when you need to stop a running Data-Juicer processing job safely, ensuring proper cleanup of resources and consistent job state tracking.
Code Reference
Source Location
- Repository: Datajuicer_Data_juicer
- File: data_juicer/utils/job/stopper.py
Signature
class JobStopper:
def __init__(self, job_id: str,
base_dir: str = "outputs/partition-checkpoint-eventlog"):
"""
Initialize the job stopper.
Args:
job_id: The job ID to stop.
base_dir: Base directory containing job outputs.
"""
def terminate_process_gracefully(self, proc, timeout: int = 10) -> bool:
"""Terminate a process gracefully with timeout, falling back to force-kill."""
def cleanup_job_resources(self) -> None:
"""Clean up job resources and update job summary to 'stopped' status."""
def stop_job(self, force: bool = False, timeout: int = 30) -> Dict[str, Any]:
"""
Stop the DataJuicer job using event log-based process discovery.
Args:
force: Force termination without graceful shutdown.
timeout: Termination timeout in seconds.
Returns:
Dictionary with job_id, success, processes/threads found and
terminated counts, and any errors.
"""
def stop_job(
job_id: str,
base_dir: str = "outputs/partition-checkpoint-eventlog",
force: bool = False,
timeout: int = 30
) -> Dict[str, Any]:
"""Stop a DataJuicer job using event log-based process discovery."""
Import
from data_juicer.utils.job.stopper import JobStopper, stop_job
I/O Contract
Inputs
| Name | Type | Required | Description |
|---|---|---|---|
| job_id | str | Yes | The unique identifier of the job to stop |
| base_dir | str | No | Base directory containing job outputs. Default: "outputs/partition-checkpoint-eventlog" |
| force | bool | No | Force termination without graceful shutdown. Default: False |
| timeout | int | No | Termination timeout in seconds. Default: 30 |
Outputs
| Name | Type | Description |
|---|---|---|
| result | Dict[str, Any] | Dictionary containing job_id, success (bool), processes_found, processes_terminated, threads_found, threads_terminated, and errors list |
| job_summary.json | File on disk | Updated job summary with status="stopped", stop_time, and stop_reason |
Usage Examples
Quick Stop
from data_juicer.utils.job.stopper import stop_job
# Stop a running job
result = stop_job("20250728_233517_510abf")
if result["success"]:
print(f"Terminated {result['processes_terminated']} processes")
Using the Class Directly
from data_juicer.utils.job.stopper import JobStopper
stopper = JobStopper("20250728_233517_510abf")
result = stopper.stop_job(force=True, timeout=60)
print(f"Success: {result['success']}")
print(f"Errors: {result['errors']}")
CLI Usage
# Stop a job from the command line
python -m data_juicer.utils.job.stopper 20250728_233517_510abf --force --timeout 60
# List all known jobs
python -m data_juicer.utils.job.stopper --list