ClearML/clearml
1
0
Fork 0
mirror of https://github.com/clearml/clearml synced 2025-02-12 07:35:08 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Add pipeline parameter to task pipeline example (#664)

Browse Source
This commit is contained in:
pollfly 2022-05-05 12:13:48 +03:00 committed by GitHub
parent bd6c38f8d1
commit 4ccb357de4
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 41 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

58
examples/pipeline/pipeline_from_tasks.py
View File

@ -4,7 +4,11 @@ from clearml.automation import PipelineController
def pre_execute_callback_example(a_pipeline, a_node, current_param_override): def pre_execute_callback_example(a_pipeline, a_node, current_param_override):
# type (PipelineController, PipelineController.Node, dict) -> bool # type (PipelineController, PipelineController.Node, dict) -> bool
print('Cloning Task id={} with parameters: {}'.format(a_node.base_task_id, current_param_override)) print(
"Cloning Task id={} with parameters: {}".format(
a_node.base_task_id, current_param_override
)
)
# if we want to skip this node (and subtree of this node) we return False # if we want to skip this node (and subtree of this node) we return False
# return True to continue DAG execution # return True to continue DAG execution
return True return True
@ -12,7 +16,7 @@ def pre_execute_callback_example(a_pipeline, a_node, current_param_override):
def post_execute_callback_example(a_pipeline, a_node): def post_execute_callback_example(a_pipeline, a_node):
# type (PipelineController, PipelineController.Node) -> None # type (PipelineController, PipelineController.Node) -> None
print('Completed Task id={}'.format(a_node.executed)) print("Completed Task id={}".format(a_node.executed))
# if we need the actual executed Task: Task.get_task(task_id=a_node.executed) # if we need the actual executed Task: Task.get_task(task_id=a_node.executed)
return return
@ -20,25 +24,43 @@ def post_execute_callback_example(a_pipeline, a_node):
# Connecting ClearML with the current pipeline, # Connecting ClearML with the current pipeline,
# from here on everything is logged automatically # from here on everything is logged automatically
pipe = PipelineController( pipe = PipelineController(
name='Pipeline demo', name="Pipeline demo", project="examples", version="0.0.1", add_pipeline_tags=False
project='examples',
version='0.0.1',
add_pipeline_tags=False,
) )
pipe.set_default_execution_queue('default') pipe.add_parameter(
"url",
"https://files.community.clear.ml/examples%252F.pipelines%252FPipeline%20demo/stage_data.8f17b6316ce442ce8904f6fccb1763de/artifacts/dataset/f6d08388e9bc44c86cab497ad31403c4.iris_dataset.pkl",
"dataset_url",
)
pipe.add_step(name='stage_data', base_task_project='examples', base_task_name='Pipeline step 1 dataset artifact') pipe.set_default_execution_queue("default")
pipe.add_step(name='stage_process', parents=['stage_data', ],
base_task_project='examples', base_task_name='Pipeline step 2 process dataset', pipe.add_step(
parameter_override={'General/dataset_url': '${stage_data.artifacts.dataset.url}', name="stage_data",
'General/test_size': 0.25}, base_task_project="examples",
base_task_name="Pipeline step 1 dataset artifact",
parameter_override={"General/dataset_url": "${pipeline.url}"},
)
pipe.add_step(
name="stage_process",
parents=["stage_data"],
base_task_project="examples",
base_task_name="Pipeline step 2 process dataset",
parameter_override={
"General/dataset_url": "${stage_data.artifacts.dataset.url}",
"General/test_size": 0.25,
},
pre_execute_callback=pre_execute_callback_example, pre_execute_callback=pre_execute_callback_example,
post_execute_callback=post_execute_callback_example post_execute_callback=post_execute_callback_example,
) )
pipe.add_step(name='stage_train', parents=['stage_process', ], pipe.add_step(
base_task_project='examples', base_task_name='Pipeline step 3 train model', name="stage_train",
parameter_override={'General/dataset_task_id': '${stage_process.id}'}) parents=["stage_process"],
base_task_project="examples",
base_task_name="Pipeline step 3 train model",
parameter_override={"General/dataset_task_id": "${stage_process.id}"},
)
# for debugging purposes use local jobs # for debugging purposes use local jobs
# pipe.start_locally() # pipe.start_locally()
@ -46,4 +68,4 @@ pipe.add_step(name='stage_train', parents=['stage_process', ],
# Starting the pipeline (in the background) # Starting the pipeline (in the background)
pipe.start() pipe.start()
print('done') print("done")