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

Hi all! I have a very specific technical question about prefect related to task configuration.

I would like a series of tasks to feel like a unix pipeline, in the sense of metadata about the previous job in the series could be fed into the next job

My use case is let’s say I have spark jobs A -&gt; B -&gt; C
Each one reads from a data lake, does some processing, then writes back to the same data lake
I would like for the output directory of A to automagically become the input directory of B
How would you do something like this with prefect?

Hi <@U02H33GCQCB> - if I understand your question correctly, you should be able to do this:
```@task
def A(dir: str):
  # .. do stuff
  return dir

@task
def B(dir:str):
  # .. do stuff
  return dir

@task
def C(dir:str):
  # .. do stuff
  return dir

with Flow("Unix-like Pipeline") as flow:
  dir = Parameter("directory")
  
  a = A(dir=dir)
  b = B(dir=a)
  c = C(dir=b)```

Yes! Or maybe even something like
```with Flow("foo") as flow:
  in = Parameter("input_dir")
  out = Parameter("output_dir")
  with Chain(in, out) as chain:
     chain(A, B, C)```

I think my issue is that unix pipelines are not graphs! What I’m wanting might be a fantasy

:laughing: it is really really nice to use, I’m not sure there’s a python equivalent :disappointed:

(in JS we can use method chaining though … :innocent: )

It’s more just me wanting unix pipelines everywhere and also <https://fishshell.com/docs/current/design.html#configurability-is-the-root-of-all-evil|this>
I _hate_ configuration

We have `|` implemented for tasks in Prefect

```    def __ror__(self, other: object) -&gt; "Task":
        """
        Creates a state dependency between `self` and `other`:
            `other | self --&gt; self.set_dependencies(upstream_tasks=[other])`

        Args:
            - other (object): An object that will be converted to a Task and set as an
                upstream dependency of this Task.

        Returns:
            - Task
        """
        self.set_dependencies(upstream_tasks=[other])
        return self```

You could implement a `Task` subclass that binds the return value of the upstream to the first argument of the downstream by overriding this method.

Thx for sharing that <@U01CEUST9B5> it sounds incredibly slick

I typically hate operator overloading because I think it leads to ungrokable code, but I’ll have to mull it over for this case…