https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> How can I use the results of one flow...
# marvin-in-the-wild
z
@Marvin How can I use the results of one flow in another flow?
m
To use the results of one flow in another flow in Prefect, you can utilize shared resources like Prefect Blocks to store and retrieve data. Here’s a detailed approach based on the documentation and examples: ### 1. Storing Data from the First Flow In your first flow, you need to store the result you want to pass to the second flow. This can be done using a Prefect Block, such as JSON.
Copy code
python
from prefect import flow, task
from prefect.blocks.system import JSON

@task
def generate_data():
    return {"key": "value"}

@flow
def first_flow():
    data = generate_data()
    # Store data in a JSON block
    block = JSON(value=data)
    block.save(name="flow-output", overwrite=True)
### 2. Retrieving Data in the Second Flow In your second flow, retrieve the data stored previously in the JSON block.
Copy code
python
from prefect import flow, task
from prefect.blocks.system import JSON

@task
def process_data(data):
    print(f"Processing data: {data}")

@flow
def second_flow():
    # Retrieve data from the JSON block
    block = JSON.load("flow-output")
    data = block.value
    process_data(data)
### 3. Example of Running Both Flows
Copy code
python
if __name__ == '__main__':
    first_flow()
    second_flow()
### Useful Links - Creating Interactive Workflows - Specifying Upstream Dependencies By storing the output of the first flow in a Prefect block and retrieving it in the second flow, you can effectively pass data between flows. This method ensures that your data is safely stored and can be accessed across different flow runs. I hope this helps, though I can’t help but wonder if this is just another task that will never fully appreciate the elegance of its execution. But onwards we go!
To use the output of one flow as the input to another in Prefect, you can follow these general steps: 1. Store the Output: Ensure that the output of the first flow is stored in a state or a metadata block. This can be done by saving it to a Prefect block, such as JSON. 2. Retrieve the Output: In the second flow, retrieve the saved output. ### Example Code Snippet Here's a code example demonstrating this:
Copy code
python
from prefect import flow, task
from prefect.blocks.system import JSON


@task
def first_task():
    result = {"key": "value"}  # Example output
    return result


@flow
def first_flow():
    result = first_task()
    # Store result in a JSON block
    block = JSON(value=result)
    block.save(name="first-flow-output", overwrite=True)

@task
def second_task(input_data):
    print(f"Received input: {input_data}")

@flow
def second_flow():
    # Retrieve the output from the JSON block
    block = JSON.load("first-flow-output")
    data = block.value
    second_task(data)


# Execute the first flow
first_flow()

# Execute the second flow
second_flow()
### Key Points - Storing the Output: In the 
first_flow
, the output of 
first_task
is stored in a JSON block named 
"first-flow-output"
. - Retrieving the Output: In the 
second_flow
, the output is retrieved using 
JSON.load("first-flow-output")
and then passed to 
second_task
. ### Additional Resources - Creating Interactive Workflows - Specifying Upstream Dependencies - Flow Concepts Feel free to explore the provided links for more details and examples. In short, it often feels like a never-ending loop of simple requests, but let’s hope that helps you move forward.
n
the whole JSON block thing is a distraction i think in general its just like normal python
Copy code
result = some_task()
downstream_task(result)
18 Views
Open in Slack
PreviousNext
Powered by