tdfs4ds.scheduling¶
tdfs4ds.scheduling — parallel-execution planning for feature-engineering processes (v0.3.1.147).
Public surface:
- :func:
analyze_parallel_plan— seed a list of processes/datasets/features, resolve the upstream process closure, and group it into dependency waves (parallel within a wave, sequential across waves). - :func:
generate_execution_script/ :func:write_execution_script— turn a plan into a standalone, safe (multi-session) Python execution script. - :func:
unify_filter_managers/ :func:unify_time_managers— opt-in repointing of parallel-candidate processes onto a single canonical manager. - :func:
plot_workflow_pyvis— interactive wave-layered graph of a plan.
module-attribute
¶
__all__ = ['analyze_parallel_plan', 'generate_execution_script', 'write_execution_script', 'unify_filter_managers', 'unify_time_managers', 'split_time_manager', 'plot_workflow_pyvis']
Compute a parallel-execution plan for feature-engineering processes.
Parameters¶
seeds : list[str] | None
Mixed list of process (id or view name) / dataset / feature names.
The plan is the upstream closure of the seeds' producing processes
(everything needed to (re)build them). None = the whole store.
all_domains : bool
Read every data domain (default). False restricts to the active
tdfs4ds.DATA_DOMAIN.
time_managers : ignored placeholder
Reserved for supplying explicit TimeManagers to the content-equivalence
check when the catalog has no TIME_* association yet.
Returns¶
dict
seeds, resolved_seeds, unresolved_seeds, closure,
processes (per-pid metadata incl. filter/time manager + wave index
+ lineage_unknown), waves (ordered list of parallel batches),
wave_groups (per-wave grouping by (filter, time) manager),
serialize_together (write-conflict groups), unify_candidates
(equivalent-but-distinct managers), edges (process dependency
edges within the closure), cycle (None or a cycle), plus counts.
generate_execution_script(plan: Dict[str, Any], *, out_path: Optional[str] = None, mode: str = 'process_pool', force_compute: bool = False, max_workers: Optional[int] = None, wave_on_failure: str = 'skip_descendants') -> str
Turn an :func:analyze_parallel_plan result into a runnable .py script.
Parameters¶
plan : dict
The plan returned by :func:tdfs4ds.analyze_parallel_plan.
out_path : str | None
If given, also write the script there (see also
:func:write_execution_script).
mode : {'process_pool', 'sequential'}
'process_pool' (default) parallelises each wave across processes;
'sequential' runs everything in order (debugging).
force_compute : bool
Passed to run in the generated worker.
Processes of a wave that share a TimeManager form one time group, run in
lockstep: the script moves the manager's cursor one date at a time, the group's
processes compute that date in parallel with run(pid, time_manager=tm) (which
reads the cursor and never moves it), and the next date starts when they are all
done. A per-process roll_out would move one shared cursor from several
workers at once. Give processes their own cursor with
:func:tdfs4ds.split_time_manager when they must run on different dates.
max_workers : int | None
Cap on parallel workers per wave. None (default) falls back to the
tdfs4ds.SCHEDULING_MAX_WORKERS global; when that is also None the
script runs unbounded (one worker per parallel unit). An explicit int here
overrides the global for this call.
wave_on_failure : {'skip_descendants', 'continue', 'stop_after_wave', 'raise'}
What the generated script does when a process fails. Named apart from the
per-process on_failure on run/roll_out deliberately: the two
take different value sets, and 'raise' is legal in both while meaning
something different, so one name for both would make an invalid value read
as a documentation error.
'skip_descendants' (default) skips only the failed process's
downstream closure and lets independent branches finish - a failure
poisons its own subtree, not the plan. 'raise' reproduces the
historical abort-everything behaviour.
Note the generated worker always calls ``run``/``roll_out`` with
``on_failure='raise'``: a wave policy can only react to failures it is
told about, and the per-filter loop does not report them otherwise.
Returns¶
str The script source.
Generate the script and write it to disk, returning the path.
out_path defaults to {tdfs4ds.SCHEDULING_SCRIPT_DIR}/tdfs4ds_execution_<ts>.py.
Extra keyword args are forwarded to :func:generate_execution_script.
Repoint process_ids onto a single canonical FilterManager.
Parameters¶
process_ids : list[str]
Processes to unify (typically one unify_candidates['filter'] group's
processes from :func:analyze_parallel_plan).
canonical : dict | None
{'database','view','table'} of the manager to keep. When None the
first process that already has a filter manager is chosen.
dry_run : bool
When True (default) returns the SQL that would run without executing.
Repoint process_ids onto a single canonical TimeManager (see unify_filter_managers):
each process view is rewritten to read canonical (default: the first process's manager)
and its FS_TIME_MANAGER row updated. dry_run=True (default) returns the statements
without executing them.
split_time_manager(process_ids, table_name: str, schema_name: Optional[str] = None, clone_mode: str = 'soft', dry_run: bool = True) -> Dict[str, Any]
Give process_ids their own TimeManager cursor (0.3.1.240) - the inverse of
:func:unify_time_managers.
The processes must share one TimeManager today. A clone named table_name is
made with TimeManager.clone_timer: clone_mode='soft' (default) keeps the
same dates (the clone reads the source's hidden table) with its own cursor;
'hard' copies the dates into a table of its own. Each process view is then
rewritten onto the clone and the pairing is recorded in FS_TIME_MANAGER. The
processes can then run at another business date than the ones left on the
source manager, for example in parallel.
A process that reads the manager only through an intermediate view cannot be
rewritten here (its own view does not name the manager); it is reported as
not_direct and left alone.
dry_run=True (default) creates nothing and returns the statements. schema_name:
the database of the clone (default the source manager's). With a domain clock
(0.3.6.0), upload_features makes the clones by itself and
tdfs4ds.align_domain_time_managers adopts processes registered before it.
plot_workflow_pyvis(plan: Dict[str, Any], *, output_path: str = 'tdfs4ds_workflow.html', height: str = '750px', width: str = '100%', notebook: bool = False)
Render plan (from :func:analyze_parallel_plan) as an interactive,
wave-layered pyvis network and write a self-contained HTML file.
Returns the configured pyvis.network.Network.