I have also tried with type hints and it still broadcasts to string. Very weird...

BTW I would really appreciate it if you let me know when you get it fixed 🙏

Oh, I see. I guess somehow I can retrieve that information via Task.logger , since it is stored in JSON format? Thanks!

I think it could be a convenient approach. The new parameter abort_on_failed_steps could be a list containing the name of the steps for which the pipeline will stop its execution if any of them fail (so that we can ignore other steps that are not crucial to continue the pipeline execution)

But maybe another solution would be to pass the configuration files paths as function arguments, then read and parse them inside the pipeline

AgitatedDove14 I ended up with two pipelines being executed until they completed the workflow but duplicating each of their steps. You can check it here:
https://clearml.slack.com/files/U02A5DGPMPU/F02SR3G9RDK/image.png

Hey AgitatedDove14 ! Any news on this? 🙂

AgitatedDove14 I have the strong feeling it must be an agent issue, because when I place PipelineDecorator.run_locally() before calling the pipeline, everything works perfectly. See:

Well, this is just a mock example 🙂 . In the real application I'm working on there will be more than one configuration file (in principle one for the data and one for the DL model). Regarding the fix, I am not in a hurry at the moment. I'll happily wait for tomorrow (or the day after) when the commit is pushed!

Great! Thank you 👍

Hi CostlyOstrich36 AgitatedDove14
Oh no, I am not trying to say that I am using each agent to run a single task. I have several agents listening to a number of queues so that they are busy most of the time. As we talked about, it is not possible to run multiple pipelines ( PipelineDecorator.pipeline ) simultaneously in a single process. That's why I had been testing locally launching pipelines in different subprocesses and this way I have managed to run several pipelines concurrently....

AgitatedDove14 After checking, I discovered that apparently it doesn't matter if each pipeline is executed by a different worker, the error persists. Honestly this has me puzzled. I'm really looking forward to getting this functionality right because it's an aspect that would make ClearML shine even more.

AgitatedDove14 So did you get the same results without step duplication?

AgitatedDove14 Exactly, I've run into the same problem

Great AgitatedDove14 , I tested it on the mock example and it worked just as I expected 🎉

That's right! run_locally() does just what I was expecting

What exactly do you mean by that? From VS Code I execute the following script, and then the agents take care of executing the code remotely:
` import pandas as pd

from clearml import Task, TaskTypes
from clearml.automation.controller import PipelineDecorator

CACHE = False

@PipelineDecorator.component(
name="Wind data creator",
return_values=["wind_series"],
cache=CACHE,
execution_queue="data_cpu",
task_type=TaskTypes.data_processing,
)
def generate_wind(start_date: st...

Yes, although I use both terms interchangeably. The information will actually be contained in JSON files.

Hi AgitatedDove14 , it's nice to know you've already pinpointed the problem! I think the solution you propose is a good one, but does that mean I have to unpack all the dictionary values as parameters of the pipeline function? Wouldn't that make the function too "dirty"? Or do you mean you will soon push a commit that will allow me to keep passing a dictionary and ClearML automatically flatten it?

AgitatedDove14 The pipelines are executed by the agents that are listening to the queue given by pipeline_execution_queue="controllers"

But how can I reference that exact daemon execution? I tried with the ID but it fails:

clearml-agent daemon AGENT_ID --stop

Sure, it would be very intuitive if the command to stop an agent would be as easy as:
clearml-agent daemon --stop AGENT_PID

Yep, I've already unmarked the venv caching setting, but still the agent reinstalls all the requirements again.
Maybe it has to do with the fact that I am not working on a Git repository and clearML is not able to locate the requirements.txt file?

Hi AgitatedDove14 , great, glad it was fixed quickly!

By the way, before releasing version 1.1.3 you might want to take a look at this mock example. I'm trying to run the same pipeline (with different configurations) in a single for loop, as you can see below:
` from clearml import Task
from clearml.automation.controller import PipelineDecorator

@PipelineDecorator.component(return_values=["msg"], execution_queue="myqueue1")
def step_1(msg: str):
msg += "\nI've survived step 1!"
re...

Hi AgitatedDove14
Using task.get_parameters I get the parameters in a dictionary, but the values are still of type 'string'. The quickest solution I can think of is parsing with eval built-in. wdyt?

For sure! Excluding some parts related to preprocessing, this is the code I would like to parallelize with dask.distributed.Client .

` from typing import Any, Dict, List, Tuple, Union
from pathlib import Path

import xarray as xr
from clearml import Task
from dask.distributed import Client, LocalCluster

def start_dask_client(
n_workers: int = None, threads_per_worker: int = None, memory_limit: str = "2Gb"
) -> Client:
cluster = LocalCluster(
n_workers=n_workers,
...

But when I call Task.current_task().task_id within the code of one of the pipeline components, I get the task ID of the component itself. I want the pipeline task ID

Oh, I see. This explains the surprising behavior. But what if Task.init code is created automatically by PipelineDecorator.component ? How can I pass arguments to the init method in that case?

Yes, I like it! I was already get used to the ' execute_steps_as_functions' argument of PipelineDecorator.debug_pipeline() but I find your proposal to be more intuitive.

Well, I can see the difference here. Using the new pipelines generation the user has the flexibility to play with the returned values of each step. We can process those values before passing them to the next step, so maybe makes little sense to include those callbacks in this case