Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.

Prefect Community

I am having trouble setting task dependencies where there is not a direct input from one task to the next.

Here is my flow (slightly obfuscated):
```with Flow("My Flow",) as flow:
    parameter = get_parameter()
    objects = get_objects()
    transformed = transform_objects(objects)
    write_data_to_s3(parameter, transformed)
    copy_data_to_snowflake(parameter, upstream_tasks=[write_data_to_s3])```
What I want to happen is to `write_data_to_s3`, and then `copy_data_to_snowflake` after `write_data_to_s3` is finished. However, this code creates a second copy of `write_data_to_s3` in my flow schematic, and does not create the relationship I am looking for. Any advice?

Yeah. Store `write_data_to_s3` in a variable and then point to that.
```with Flow(...) as flow:
    a = first_task()
    b = second_task()
    c = third_task(c_inputs, upstream_tasks=[a,b])```

Or you can do:
```with Flow(...) as flow:
     a = first_task()
     b = second_task()

     c = third_task()
     c.set_upstream(b)
     c.set_upstream(a)```

Does it matter that `second_task` and `third_task` both accept the same `parameter` value?

I don’t think it should. You just get another line in your DAG schematic

In your example, if `a` is an input into `third_task`, is it necessary to also set BOTH `a` and `b` as upstream tasks?

No because the dependency from `a` will already be built so I think you should only need `b`. This is a hard _think_ though

Yea I thought I was doing it right... I was getting an error that I was calling `write_data_to_s3` without providing required arguments, which felt completely wrong. Trying again and will report back. Thank you!

Screen Shot 2021-12-02 at 5.46.32 PM.png

So with:
```with Flow("ex") as flow:
    a = first_task()
    b = second_task()
    c = third_task(a, upstream_tasks=[b])```
I got: