Hi guys I m currently facing an issue with Parameter how can

Question

Hi guys I m currently facing an issue with Parameter how can i get the value of a Parameter

Anna Geller · Answer

could you move the code block below to this thread here

Anna Geller · Answer

or rather all blocks smile this helps us keep the main channel cleaner and makes it easier to follow for others reading this

Thomas Mignon · Answer

Here is my error ```File home1 datawork semexp workspace prefect graph semaphore template task py line 26 in run os path join workspace f command test File home datawork semaphore exp conda conda envs prefect lib python3 9 posixpath py line 76 in join a = os fspath a TypeError expected str bytes or os PathLike object not Parameter``` Because `command` was a parameter passed downstream And here are my 2 files the main flow py who is creating the graph of a flow of flow and a flow named purify flow py who use a template task to create tasks inside his flow Here is main flow py ```from dask jobqueue import PBSCluster from prefect import Flow task Parameter from prefect tasks prefect import create flow run from prefect executors import DaskExecutor from prefect graph semaphore template task import SemaphoreTemplate from prefect graph purify flow import flow as purifyFlow task def partition workspace semaphore audience version day month year SemaphoreTemplate home datawork semaphore exp conda conda envs semaphore audience 4 4 9 lib python3 7 bin generate partition year + month + day + J input purify directory basepath= home datawork semaphore exp cache data common daily purify workspace semaphore audience version run task def event session workspace semaphore audience version day month year SemaphoreTemplate home datawork semaphore exp conda conda envs semaphore audience 4 4 9 lib python3 7 bin generate event session year + month + day + J input partition directory basepath= home datawork semaphore exp cache data common daily partition workspace semaphore audience version run with Flow main flow executor=DaskExecutor cluster class=PBSCluster cluster kwargs= cores 1 memory 120GB n workers 1 queue omp resource spec select=1 ncpus=56 mem=500gb walltime 01 00 00 as flow workspace = Parameter workspace semaphore audience version = Parameter semaphore audience version semaphore metadata version = Parameter semaphore metadata version day = Parameter day default= 20200101 month = Parameter month default= 20200101 year = Parameter year default= 20200101 purifyFlow run name= purify flow parameters=dict year=year month=month day=day workspace=workspace semaphore audience version=semaphore audience version ``` and here is purify flow py ```from dask jobqueue import PBSCluster from prefect executors import DaskExecutor from prefect import Flow Parameter Task from prefect utilities tasks import defaults from attrs from semaphore template task import SemaphoreTemplate class PurifyTemplate Task def init self command parameter workspace semaphore audience version kwargs self command parameter = command parameter self workspace = workspace self semaphore audience version = semaphore audience version super init kwargs defaults from attrs command parameter workspace semaphore audience version def run self command parameter=None workspace=None semaphore audience version=None kwargs SemaphoreTemplate command= home datawork semaphore exp conda conda envs semaphore audience 4 4 9 lib python3 7 bin purify raw logs command parameter=command parameter workspace=workspace semaphore audience version=semaphore audience version run with Flow purify flow executor=DaskExecutor cluster class=PBSCluster cluster kwargs= cores 1 memory 120GB n workers 1 queue omp resource spec select=1 ncpus=56 mem=500gb walltime 05 00 00 as flow workspace = Parameter workspace semaphore audience version = Parameter semaphore audience version day = Parameter date month = Parameter month year = Parameter year PurifyTemplate command parameter= home datawork semaphore exp spool input ftp sync isi eftp1 + year + eftp1 f + year + + month + + day + eftp1 gz workspace=workspace semaphore audience version=semaphore audience version run PurifyTemplate command parameter= home datawork semaphore exp spool input ftp sync isi eftp2 + year + eftp2 f + year + + month + + day + eftp1 gz workspace=workspace semaphore audience version=semaphore audience version run PurifyTemplate command parameter= home datawork semaphore exp spool input ftp ftp + year + ftp f access log + year + month + day + gz workspace=workspace semaphore audience version=semaphore audience version run PurifyTemplate command parameter= home datawork semaphore exp spool input ftp ftp2 + year + ftp2 f access log + year + month + day + gz workspace=workspace semaphore audience version=semaphore audience version run PurifyTemplate command parameter= home datawork semaphore exp spool input http vpublicnew + year + http full vpublicnew f + year + + month + + day + log workspace=workspace semaphore audience version=semaphore audience version run PurifyTemplate command parameter= home datawork semaphore exp spool input http vpublicnew + year + http full vpublicnew f + year + + month + + day + log workspace=workspace semaphore audience version=semaphore audience version run ```

Anna Geller · Answer

you can pass Parameter values to downstream tasks as data dependencies just as any data dependencies in a Prefect flow alternatively you can grab the Parameter values in your custom tasks or state handlers directly from the context ```prefect context parameters get parameter name ```

Anna Geller · Answer

I think you misunderstood my intent I wanted you to move everything after Hi guys I m currently facing an issue with Parameter how can i get the value of a Parameter to the thread could you do that

Thomas Mignon · Answer

Sorry yes i will do this

Anna Geller · Answer

also check this blog post for deep dive on parametrization <https www prefect io blog how to make your data pipelines more dynamic using parameters in prefect >

Thomas Mignon · Answer

i will try the context option because passing downstream a parameter of type Parameter to a task that need a parameter as a string isn t working if there is any chance to do something like `myparameters value` or something like that

Anna Geller · Answer

also you shouldn t do this ```purifyFlow run name= purify flow parameters=dict year=year month=month day=day workspace=workspace semaphore audience version=semaphore audience version ``` if you want to trigger some other flow from this parent flow you can use the `create flow run` task instead explains how you could do that and if you need some examples check

Thomas Mignon · Answer

The problem with `create flow run` is that there is a request done to <http api prefect io|api prefect io> and where i laaunch my script i get an HTTP 443 error that s why i ve done this hack

Anna Geller · Answer

The error you see is not because of Parameters but because you are calling a flow from another flow in a way that is not supported The link above show how you can do that but here is a simple example ```with Flow PARENT FLOW NAME as parent flow normal non subflow task = hello world parent first child flow run id = create flow run flow name=CHILD FLOW NAME project name=PREFECT PROJECT NAME parameters=dict user input= First child flow run task args=dict name= First subflow upstream tasks= normal non subflow task ``` from <https github com anna geller orchestrator pattern>

Anna Geller · Answer

I can totally understand that but this is not how you should trigger a child flow run The proper way of doing this is by registering this child flow and triggering a child flow run from the parent flow via `create flow run`

Thomas Mignon · Answer

Ok thanks for all the info i will rewrite my code with `create flow run` but i think i will face the HTTP error i was getting before see no evil

Thomas Mignon · Answer

I will paste the error when i m done slightly smiling face

Anna Geller · Answer

to explain it a bit more when you do ```purifyFlow run name= purify flow parameters=dict year=year month=month day=day workspace=workspace semaphore audience version=semaphore audience version ``` this gets evaluated at flow registration i e at build time when the computational graph is constructed and at build time `workspace` is of type Parameter task Only once you actually run this flow `workspace` object becomes an actual parameter value does this explanation make sense

Thomas Mignon · Answer

I understood why i was getting a Parameter type this make sense slightly smiling face

Thomas Mignon · Answer

And for tasks in a flow there is an equivalent like `create task run` instead of doing ```PurifyTemplate year=year month=month day=day command parameter= home datawork semaphore exp spool input ftp sync isi eftp1 + year + eftp1 f + year + + month + + day + eftp1 gz workspace=workspace semaphore audience version=semaphore audience version run ```

Anna Geller · Answer

You shouldn t have to call tasks like that The right approach with the imperative API would be to initialize it before the Flow constructor and then call it within the flow like so ```purify template = PurifyTemplate year=year month=month day=day command parameter= home datawork semaphore exp spool input ftp sync isi eftp1 + year + eftp1 f + year + + month + + day + eftp1 gz workspace=workspace semaphore audience version=semaphore audience version with Flow yourflowname as flow purify template ```

Thomas Mignon · Answer

Ok i see but in this way if i initalize PurifyTemplate outside of flow i will not have access to year month dat worskapce semaphore audience version because the flow is receveing theses parameters

Thomas Mignon · Answer

Maybe by read them from the prefect context

Anna Geller · Answer

you will if you put those into the run method

Anna Geller · Answer

it depends on how you built your task This is one of the reasons why the imperative API is so confusing and why it s much easier to just use the functional API With the imperative you need to differentiate between the args passed to the init method and those passed to the run method Here is how you could call this task ```with Flow yourflowname as flow PurifyTemplate init args run method args ``` Does it make things clearer

Thomas Mignon · Answer

Not so clearer ahah because my PurifyTemplate class has init and run method with arguments this is the Imperative API and in the code above ```with Flow yourflowname as flow PurifyTemplate init args run method args ``` PurifyTemplate is initialized inside the flow but you was telling me to initalize it outside i m a bit lost slightly smiling face

Anna Geller · Answer

this ```with Flow yourflowname as flow PurifyTemplate init args run method args ``` is the same as ```purify template = PurifyTemplate init args with Flow yourflowname as flow purify template run method args ``` and run method args can receive data dependencies from upstream tasks

Anna Geller · Answer

the latter is cleaner IMO

Thomas Mignon · Answer

Ok got it slightly smiling face But if i use the latter on i have to pass from this ```class PurifyTemplate Task def init self command parameter workspace semaphore audience version kwargs self command parameter = command parameter self workspace = workspace self semaphore audience version = semaphore audience version super init kwargs defaults from attrs command parameter workspace semaphore audience version def run self command parameter=None workspace=None semaphore audience version=None kwargs SemaphoreTemplate command= home datawork semaphore exp conda conda envs semaphore audience 4 4 9 lib python3 7 bin purify raw logs command parameter=command parameter workspace=workspace semaphore audience version=semaphore audience version run ``` to this ```class PurifyTemplate Task def init self kwargs super init kwargs def run self command parameter=None workspace=None semaphore audience version=None kwargs SemaphoreTemplate command= home datawork semaphore exp conda conda envs semaphore audience 4 4 9 lib python3 7 bin purify raw logs command parameter=command parameter workspace=workspace semaphore audience version=semaphore audience version run ```

Thomas Mignon · Answer

i e clean the init method because it won t have access to the parameters that the flow gonna get

Anna Geller · Answer

don t call run in a run method smile so meta This seems fine though ```class PurifyTemplate Task def init self kwargs super init kwargs def run self command parameter=None workspace=None semaphore audience version=None kwargs SemaphoreTemplate command= home datawork semaphore exp conda conda envs semaphore audience 4 4 9 lib python3 7 bin purify raw logs command parameter=command parameter workspace=workspace semaphore audience version=semaphore audience version ```

Kevin Kho · Answer

The init won t have access to Parameters but you should be able to get them in the run method I don t think you need to remove the `default from attrs`

Kevin Kho · Answer

I think Thomas needs to call the run actually Otherwise the run will return an instantiated task

Anna Geller · Answer

the best solution would be to convert this task to the functional API honestly smile

Anna Geller · Answer

``` task def purify template command parameter=None workspace=None semaphore audience version=None your template class = SemaphoreTemplate command= home datawork semaphore exp conda conda envs semaphore audience 4 4 9 lib python3 7 bin purify raw logs command parameter=command parameter workspace=workspace semaphore audience version=semaphore audience version your template class some method doing sth ```

Thomas Mignon · Answer

If this can help to resolve my problem here is also the code of my SempahoreTemplate that i will have to transform too i think ```class SemaphoreTemplate Task def init self command command parameter workspace semaphore audience version kwargs self command = command self command parameter = command parameter self workspace = workspace self semaphore audience version = semaphore audience version super init kwargs defaults from attrs command command parameter workspace semaphore audience version def run self command=None command parameter=None workspace=None semaphore audience version=None kwargs print os environ print os environ workspace path Path = Path os path join workspace f command os environ PBS JOBID template runner TemplateRunner command=command version=semaphore audience version command parameter=command parameter workspace=workspace path skip signal=True run exit=False ```

Thomas Mignon · Answer

I will try your version Anna slightly smiling face

Anna Geller · Answer

the functional API with ` task` is the preferred way of defining tasks and the only one supported in Prefect 2 0

Anna Geller · Answer

ideally don t create any subclasses of Task like `SemaphoreTemplate Task ` if you convert it to functions this will make it much easier to reason about and debug

Thomas Mignon · Answer

Ok slightly smiling face task come from `from prefect utilities tasks`

Thomas Mignon · Answer

or from `prefect core task`

Thomas Mignon · Answer

Here is the error i got with `create run flow`

Thomas Mignon · Answer

``` 2022 03 31 13 39 19+0000 INFO prefect TaskRunner | Task create flow run Starting task run 2022 03 31 13 40 21+0000 ERROR prefect TaskRunner | Task create flow run Exception encountered during task execution Traceback most recent call last File home datawork semaphore exp conda conda envs prefect lib python3 9 site packages urllib3 connection py line 174 in new conn conn = connection create connection File home datawork semaphore exp conda conda envs prefect lib python3 9 site packages urllib3 util connection py line 95 in create connection raise err File home datawork semaphore exp conda conda envs prefect lib python3 9 site packages urllib3 util connection py line 85 in create connection sock connect sa OSError Errno 101 Network is unreachable During handling of the above exception another exception occurred Traceback most recent call last File home datawork semaphore exp conda conda envs prefect lib python3 9 site packages urllib3 connectionpool py line 703 in urlopen httplib response = self make request File home datawork semaphore exp conda conda envs prefect lib python3 9 site packages urllib3 connectionpool py line 386 in make request self validate conn conn File home datawork semaphore exp conda conda envs prefect lib python3 9 site packages urllib3 connectionpool py line 1040 in validate conn conn connect File home datawork semaphore exp conda conda envs prefect lib python3 9 site packages urllib3 connection py line 358 in connect conn = self new conn File home datawork semaphore exp conda conda envs prefect lib python3 9 site packages urllib3 connection py line 186 in new conn raise NewConnectionError urllib3 exceptions NewConnectionError lt urllib3 connection HTTPSConnection object at 0x2aaaaf59de50 gt Failed to establish a new connection Errno 101 Network is unreachable During handling of the above exception another exception occurred Traceback most recent call last File home datawork semaphore exp conda conda envs prefect lib python3 9 site packages requests adapters py line 440 in send resp = conn urlopen File home datawork semaphore exp conda conda envs prefect lib python3 9 site packages urllib3 connectionpool py line 813 in urlopen return self urlopen File home datawork semaphore exp conda conda envs prefect lib python3 9 site packages urllib3 connectionpool py line 813 in urlopen return self urlopen File home datawork semaphore exp conda conda envs prefect lib python3 9 site packages urllib3 connectionpool py line 813 in urlopen return self urlopen Previous line repeated 3 more times File home datawork semaphore exp conda conda envs prefect lib python3 9 site packages urllib3 connectionpool py line 785 in urlopen retries = retries increment File home datawork semaphore exp conda conda envs prefect lib python3 9 site packages urllib3 util retry py line 592 in increment raise MaxRetryError pool url error or ResponseError cause urllib3 exceptions MaxRetryError HTTPSConnectionPool host= port=443 Max retries exceeded with url Caused by NewConnectionError lt urllib3 connection HTTPSConnection object at 0x2aaaaf59de50 gt Failed to establish a new connection Errno 101 Network is unreachable During handling of the above exception another exception occurred Traceback most recent call last File home datawork semaphore exp conda conda envs prefect lib python3 9 site packages prefect engine task runner py line 880 in get task run state value = prefect utilities executors run task with timeout File home datawork semaphore exp conda conda envs prefect lib python3 9 site packages prefect utilities executors py line 468 in run task with timeout return task run args kwargs type ignore File home datawork semaphore exp conda conda envs prefect lib python3 9 site packages prefect tasks prefect flow run py line 131 in create flow run flow = FlowView from flow name flow name project name=project name File home datawork semaphore exp conda conda envs prefect lib python3 9 site packages prefect backend flow py line 205 in from flow name flows = cls query for flows File home datawork semaphore exp conda conda envs prefect lib python3 9 site packages prefect backend flow py line 289 in query for flows result = client graphql flow query File home datawork semaphore exp conda conda envs prefect lib python3 9 site packages prefect client client py line 452 in graphql result = File home datawork semaphore exp conda conda envs prefect lib python3 9 site packages prefect client client py line 407 in post response = self request File home datawork semaphore exp conda conda envs prefect lib python3 9 site packages prefect client client py line 641 in request response = self send request File home datawork semaphore exp conda conda envs prefect lib python3 9 site packages prefect client client py line 506 in send request response = File home datawork semaphore exp conda conda envs prefect lib python3 9 site packages requests sessions py line 577 in post return self request POST url data=data json=json kwargs File home datawork semaphore exp conda conda envs prefect lib python3 9 site packages requests sessions py line 529 in request resp = self send prep send kwargs File home datawork semaphore exp conda conda envs prefect lib python3 9 site packages requests sessions py line 645 in send r = adapter send request kwargs File home datawork semaphore exp conda conda envs prefect lib python3 9 site packages requests adapters py line 519 in send raise ConnectionError e request=request requests exceptions ConnectionError HTTPSConnectionPool host= port=443 Max retries exceeded with url Caused by NewConnectionError lt urllib3 connection HTTPSConnection object at 0x2aaaaf59de50 gt Failed to establish a new connection Errno 101 Network is unreachable 2022 03 31 13 40 21+0000 INFO prefect TaskRunner | Task create flow run Finished task run for task with final state Failed distributed dask worker INFO Exiting on signal 15``` My machine havent access to HTTP she dialog via TCP with the dask worker

Anna Geller · Answer

from prefect import task

Thomas Mignon · Answer

So if there is a way to use `create flow run` without any HTTP request i would take a solution slightly smiling face

Anna Geller · Answer

This is possible in Prefect 2 0 using subflows but not in 1 0

Anna Geller · Answer

Thomas Mignon · Answer

Ok i will try to pass on prefect 2 0 slightly smiling face

Thomas Mignon · Answer

Hi Is there any example of how to use subflows in prefect orion slightly smiling face < Anna Geller>

Anna Geller · Answer

You can call flows within other flows ```from prefect import flow task import time flow def subflow 1 print Subflow 1 started time sleep 3 return Hello from subflow flow def subflow 2 print Subflow 2 started time sleep 3 return Hello from the second subflow task def normal task print A normal task flow def main flow state subflow 1 = subflow 1 state subflow 2 = subflow 2 normal task wait for= state subflow 1 state subflow 2 if name == main main flow state = main flow ```

Thomas Mignon · Answer

Oh ok simple as that slightly smiling face And for the equivalent of my executor where to put the config i was setting as this before ```with Flow main flow executor=DaskExecutor cluster class=PBSCluster cluster kwargs= cores 1 memory 120GB n workers 1 queue omp resource spec select=1 ncpus=56 mem=500gb walltime 01 00 00 as flow ```

Thomas Mignon · Answer

Found it it was taskRunner slightly smiling face

Thomas Mignon · Answer

But now i m trying to launch my prfect flow with CLI

Thomas Mignon · Answer

before i wa doing this ```prefect run path home1 datawork semexp workspace prefect graph prefect graph main flow py param workspace= home1 datawork semexp workspace dagit running param semaphore audience version= 3 4 9 param day=01 param month=01 param year=2022 watch``` but `prefect run` doesnt exist anymore

Kevin Kho · Answer

I think it s `prefect deployment run` Let me confirm

Kevin Kho · Answer

Yes there is a run <https orion docs prefect io api ref prefect cli deployment prefect cli deployment run|here> but I don t think was versatile as Prefect 1 0 `prefect run` yet

Thomas Mignon · Answer

< Kevin Kho> I can t pass path to and param to `prefect deployment run` as i was doing previously in v1 with `prefect run` Have you a trick to launch a flow with CLI slightly smiling face

Thomas Mignon · Answer

I think i have to use `prefect deployment create` before `prefect deployment run` Here is my file `deployment yaml` i created in order to use ``prefect deployment create deployment yaml`` ``` name purify flow flow location main flow py name main flow flow location main flow py ``` And i got this error ``` 114 def load flow self 115 if self flow location and not self flow 116 self flow = load flow from script self flow location self flow name 117 if not self flow name 118 self flow name = self flow name 119 home datawork semaphore exp conda conda envs prefect lib python3 9 site packages prefect deploy ments py 199 in load flow from script 196 197 198 elif not flow name and len flows gt 1 199 raise UnspecifiedFlowError 200 f Found len flows flows at script path r listrepr flows keys 201 Specify a flow name to select a flow to deploy 202 UnspecifiedFlowError Found 2 flows at home1 datawork semexp workspace prefect graph prefect graph main flow py purify flow main flow Specify a flow name to select a flow to deploy ``` Because i have to specify a flow on `prefect deployment create deployment yaml` but i don t see any options name flow or something like that who would look like this `prefect deployment create deployment yaml name purify flow`

Thomas Mignon · Answer

Maybe i have to spec a ` flow name ` in `deployment yaml` too when i look in the code this seems to be better but my question remain how to pass params to myflow via `prefect deployment create`` or after `prefact deployment run`

Anna Geller · Answer

if you want to trigger a flow run from the CLI and see this flow run in the UI you don t have to run a flow run from deployment you can do ```python yourflow py```

Anna Geller · Answer

At the moment you can t override parameter values from a run from deployment via the CLI but you can override default params using the using `create flow run from deployment` or to do it from the UI is on the roadmap the CLI will likely follow

Thomas Mignon · Answer

The command should be `python home1 datawork semexp workspace prefect graph prefect graph main flow py param workspace= home1 datawork semexp workspace dagit running param semaphore audience version= 3 4 9 param day=01 param month=01 param year=2022 watch`

Anna Geller · Answer

python command is not a prefect command slightly smiling face you would need some argparser Let me give you an example

Thomas Mignon · Answer

for `create flow run from deployment` i will not need it if i can launch the flow via the python command or i misunderstood

Thomas Mignon · Answer

Ok thank you slightly smiling face

Anna Geller · Answer

```from prefect import flow task import sys task def do something with path path str filename = f path some data txt data = Hello world from Prefect 2 0 with open filename w as file file write data flow def parametrized flow with path path str do something with path path if name == main path val = sys argv 1 parametrized flow with path path val ```

Anna Geller · Answer

```python parametrized path flow py Users anna repos orion flows 00 prefect 2 0 ```

Thomas Mignon · Answer

Thank you for the example i will try that blush

Thomas Mignon · Answer

My flow is running now thank you a lot Anna slightly smiling face

Thomas Mignon · Answer

I m facing new erros but it s better i m in dask already

Thomas Mignon · Answer

I ve this strange error ``` File home datawork semaphore exp conda conda envs prefect lib python3 9 site packages prefect graph purify flow py line 9 in purify template semaphore template File home datawork semaphore exp conda conda envs prefect lib python3 9 site packages prefect tasks py line 356 in call return enter task run engine File home datawork semaphore exp conda conda envs prefect lib python3 9 site packages prefect engine py line 505 in enter task run engine raise RuntimeError Tasks cannot be called outside of a flow ``` and my task is called inside the flow sweat smile

Thomas Mignon · Answer

``` task def purify template command parameter=None workspace=None semaphore audience version=None semaphore template command= home datawork semaphore exp conda conda envs semaphore audience 4 4 9 lib python3 7 bin purify raw logs command parameter=command parameter workspace=workspace semaphore audience version=semaphore audience version flow task runner=DaskTaskRunner cluster class= dask jobqueue PBSCluster cluster kwargs= cores 1 memory 120GB n workers 1 queue omp resource spec select=1 ncpus=56 mem=500gb walltime 01 00 00 def purify flow workspace str semaphore audience version str day str month str year str purify template command parameter= home datawork semaphore exp spool input ftp sync isi eftp1 + year + eftp1 f + year + + month + + day + eftp1 gz workspace=workspace semaphore audience version=semaphore audience version ```

Thomas Mignon · Answer

Arg my task called in the flow is calling a task herself i understood

Anna Geller · Answer

is this an instantiated Task looks like it My recommendation from earlier was to just make it a function and decorate with ` task`

Thomas Mignon · Answer

``` task def semaphore template command=None command parameter=None workspace=None semaphore audience version=None workspace path Path = Path os path join workspace f command os environ PBS JOBID template runner TemplateRunner command=command version=semaphore audience version command parameter=command parameter workspace=workspace path skip signal=True run exit=False ``` It is already a function decorated with ` task`

Thomas Mignon · Answer

Maybe i m not supposed to call a task in a task in a flow

Anna Geller · Answer

yup that s true you can call tasks in flows and you can call other flows in flows but calling tasks in tasks is not supported we ve talked about it before

Thomas Mignon · Answer

Ok i will create a flow with one task to prevent this slightly smiling face