AgitatedDove14 worked like a charm, thanks a lot!
I want to collect the dataframes from teh red tasks, and display them in the pipeline task
Well this will have to wait a bit... my clearml-server is causing problems
Hi WackyRabbit7
I have a pipeline controller task, which launches 30 tasks. Semantically there are 10 applications, and I run 3 tasks for each (those 3 are sequential, so in the UI it looks like 10 lines of 3 tasks).
👍
In one of those 3 tasks that run for every app, I save a dataframe under the name "my_dataframe".
I'm assuming as an artifact:
What I want to achieve is once all tasks are over, to collect all those "my_dataframe" artifacts (10 in number), extract a single line from each, and concatenate them to a new dataframe. I want to register this new dataframe to the pipeline task. It's kind of a summary of the most important detail in the process.
I see that makes sense to me.
Notice that "monitor_arrtifact" does not change the artifact itself, so this is not a perfect match to your use case.
So the question is, how do I get a "callback" so that I collect those artifacts.
From the top of my head (an initial design, might have a bit of typos 🙂 ):
` class Collector(object):
_tasks_to_collect = []
@classmethod
def collect_me(cls, a_pipeline, a_node):
cls._tasks_to_collect.append(a_node.executed)
@classmethod
def process_results(cls, a_pipeline, a_node:
result = []
for task_id in cls._tasks_to_collect:
df = Task.get_task(task_id).artifacts['my_artifact'].get()
# do something
result.append('something here?!')
# this will return the pipeline Task
Task.current_task().upload_artifact(name='processed results', result)
Add the collector steps (i.e. the 10 Tasks
pipe.add_task(...
post_execute_callback=Collector.collect_me
)
Add to the final?! step of the pipeline (I'm assuming there is one)
pipe.add_task(...
post_execute_callback=Collector.process_results
) `
wdyt?
and then how would I register the final artifact to the pipelien? AgitatedDove14 ⬆
Okay, looks interesting but actually there is no final task, this is the pipeline layout
is this already available or only on github?
So could you re-explain assuming my piepline object is created by pipeline = PipelineController(...) ?
I'm using pipe.start_locally so I imagine I don't have to .wait() right?
is it possible to access the children tasks of the pipeline from the pipeline object?
So if I'm collecting from the middle ones, shouldn't the callback be attached to them?
If this is the case I would do:
` # Add the collector steps (i.e. the 10 Tasks
pipe.add_task(...
post_execute_callback=Collector.collect_me
)
pipe.start()
pipe.wait()
Collector.process_results(pipe) `wdyt?
WackyRabbit7 , isn't this what you need?
https://clear.ml/docs/latest/docs/references/sdk/automation_controller_pipelinecontroller#get_running_nodes
okay but still I want to take only a row of each artifact
What do you mean?
How do I get from the node to the task object?
pipeline_task = Task.get_task(task_id=Task.current_task().parent)
So could you re-explain assuming my piepline object is created by
pipeline = PipelineController(...)
?
pipe.add_step(name='stage_train', parents=['stage_process', ], monitor_artifact=['my_created_artifact'], base_task_project='examples', base_task_name='pipeline step 3 train model', parameter_override={'General/dataset_task_id': '${stage_process.id}'})This will put the artifact names "my_created_artifact" from the step Task, on the main pipeline, automatically
BTW: the new pipeline decorator interface example is here:
https://github.com/allegroai/clearml/blob/master/examples/pipeline/pipeline_from_decorator.py
okay but still I want to take only a row of each artifact
I think I got it, I'll ping her again if it won't succeed
How do I get from the node to the task object?
