Using the functional api how can i set a dependency when the

Question

Using the functional api how can i set a dependency when the outputs of a task dont depend on another for example i have 3 tasks read delete write I want to read some data then delete the old stuff and then right the new stuff deleting does not take any inputs but we want to wait until reading is done before we do it

emre · Answer

After you initialize your task you should pass these non data dependencies as the kwarg `upstream tasks` Any upstream task that shouldn t provide a value for our task should be passed to this kwarg in a list ```mytask = MyTask name= tsk upstream task with data upstream tasks= upstream task without data ```

Marc Lipoff · Answer

Will that show up in the flow visualize

emre · Answer

yeah it wont have any labels on the edge

emre · Answer

since you know there isn t a param being passed

Marc Lipoff · Answer

I get an error

Marc Lipoff · Answer

Here is the code

Marc Lipoff · Answer

``` start date hour end date hour = start end date hour lookback hours start date hour input end date hour input timestamp list = generate list of timestamp ranges start date hour end date hour chunks per hour query = get query map timestamp list unmapped table name df = big query extract map query=query project=unmapped gcp project credentials=unmapped gcp credentials to dataframe=unmapped True success = truncate existing data s3 path=s3 write path timestamp list=timestamp list upstream tasks= big query extract write to s3 map df s3 path=unmapped s3 write path truncate successful=unmapped success ``` and error ```Traceback most recent call last File google analytics import py line 218 in lt module gt FlowManager flow flow params do entry point File Users computerai covid19lab data flows utils deployment py line 286 in do = runner run return tasks=self flow tasks File Users computerai miniconda3 lib python3 8 site packages prefect engine flow runner py line 270 in run raise exc File Users computerai miniconda3 lib python3 8 site packages prefect engine flow runner py line 252 in run state = self get flow run state File Users computerai miniconda3 lib python3 8 site packages prefect utilities executors py line 70 in inner return runner method self args kwargs File Users computerai miniconda3 lib python3 8 site packages prefect engine runner py line 68 in inner raise exc File Users computerai miniconda3 lib python3 8 site packages prefect engine runner py line 48 in inner new state = method self state args kwargs File Users computerai miniconda3 lib python3 8 site packages prefect engine flow runner py line 593 in get flow run state task states task = executor submit File Users computerai miniconda3 lib python3 8 site packages prefect executors local py line 28 in submit return fn args kwargs File Users computerai miniconda3 lib python3 8 site packages prefect engine flow runner py line 745 in run task return task runner run File Users computerai miniconda3 lib python3 8 site packages prefect utilities executors py line 400 in wrapper return func args kwargs File Users computerai miniconda3 lib python3 8 site packages prefect engine task runner py line 320 in run raise exc File Users computerai miniconda3 lib python3 8 site packages prefect engine task runner py line 291 in run state = self get task run state state inputs=task inputs File Users computerai miniconda3 lib python3 8 site packages prefect utilities executors py line 70 in inner return runner method self args kwargs File Users computerai miniconda3 lib python3 8 site packages prefect engine runner py line 68 in inner raise exc File Users computerai miniconda3 lib python3 8 site packages prefect engine runner py line 48 in inner new state = method self state args kwargs File Users computerai miniconda3 lib python3 8 site packages prefect engine task runner py line 856 in get task run state value = prefect utilities executors run task with timeout File Users computerai miniconda3 lib python3 8 site packages prefect utilities executors py line 298 in run task with timeout return task run args kwargs type ignore File Users computerai miniconda3 lib python3 8 site packages prefect utilities tasks py line 449 in method return run method self args kwargs File Users computerai miniconda3 lib python3 8 site packages prefect tasks gcp bigquery py line 136 in run raise ValueError No query provided ValueError No query provided ```

emre · Answer

try `upstream tasks= df ` You need to set the relationship as if you are establishing a data dependency just instead of a normal argument put it in a list and give the list to upstream tasks So use df instead of big query extract

Marc Lipoff · Answer

`truncate existing data` is not being mapped If I do that it will reduce all df before preforming `truncate existing data`

Marc Lipoff · Answer

right i want to wait until they are all done but `write to s3`is mapped and i dont want to reduce all df as I want `write to s3`to be mapped

emre · Answer

Oh ok thats not an issue `truncate existing data` will wait for all df to be ready `write to s3` can still map over each df individually You can use the reduced results in one place and still use unreduced results to map over them try it out

Marc Lipoff · Answer

ok cool