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', 'clone_shared_view_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.
clone_shared_view_time_manager(view: str, *, clone_name: Optional[str] = None, dry_run: bool = True) -> Dict[str, Any]
Give the processes that read a TimeManager through a shared view one clone of their own (0.3.6.8).
view ('db.view', not a process view) filters its source with a TimeManager, for
example the domain clock, and several processes read it. None of them can get its own
soft clone, because rewriting view for one rewrites it for all. This makes one soft
clone of that manager next to view (<view>_TM or clone_name: same dates, own
cursor, at the manager's current step), rewrites only the manager reference inside view
(nothing else of its text changes, and it is never copied), and records the clone as the
TimeManager of every registered process reading view (FS_TIME_MANAGER + lineage).
The readers then move together, apart from the manager and from other processes;
roll_out(data_domain=...) keeps the clone in step with the clock.
Copying view per process, or moving its date filter into each process view, gives each process a cursor of its own; that changes your SQL and is left to you.
An existing object with the clone's name is reused only when it is already a soft clone over the same dates; anything else is refused (nothing is created or rewritten).
dry_run=True (default) returns the statements and the readers without changing
anything. Returns {view, manager, clone, readers, skipped, statements, dry_run} plus
error and hint when refused; never raises for a refusal.
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.