<@ULVA73B9P> if i want to run sub-flow as separate...
# marvin-in-the-wilds
Srul Pinkas
09/13/2024, 3:55 PM@Marvin if i want to run sub-flow as separate deployment, what's the best way to pass large-ish artifacts? they can be serialized to json but they are large data-sets, and i don't want to spam s3 with these temp-inputs. can i share files somehow even if they will be running on external resource? i'm using prefect 2.20
m
Marvin
09/13/2024, 3:56 PMSo you want to pass large datasets between a main flow and sub-flows without spamming S3, and you're using Prefect 2.20. Let's look into the best way to share large files or artifacts, even if they are running on external resources.
I'll search the Prefect 2.x docs first to see the recommended approach for this sort of data transfer.
To share large files between a main flow and sub-flows without using S3 for intermediary storage, you can use Prefect’s support for various remote filesystems, allowing you to centralize storage in a different medium.
Here are the potential steps and a couple of storage options:
### Use Remote File System
Prefect 2.x supports numerous remote file systems that can be integrated through
RemoteFileSystem Copy code
python
from prefect.filesystems import RemoteFileSystem
from prefect import flow, task
# Configure Remote File System
gcs_block = RemoteFileSystem(basepath="<gcs://my-bucket/folder/>")
# Save block configuration (do this during setup)
gcs_block.save("gcs-storage")
@task
def save_large_data(data, filename):
gcs_block.write_path(filename, data)
@task
def load_large_data(filename):
return gcs_block.read_path(filename)
@flow
def main_flow(data):
save_large_data(data, "large_dataset.json")
sub_flow("large_dataset.json")
@flow
def sub_flow(filename):
data = load_large_data(filename)
# Process data here
# Run main flow
main_flow(large_data_object)RemoteFileSystems
Srul Pinkas
09/16/2024, 6:14 PMhi @Marvin - using prefect 2 i executed sub-flows of separate deployments using
await asyncio.gatherrun_deploymentawait flow_run.state.result().get() Copy code
File "/home/projects/insertions_model/.venv/lib/python3.9/site-packages/prefect/results.py", line 630, in get
blob = await self._read_blob(client=client)
File "/home/projects/insertions_model/.venv/lib/python3.9/site-packages/prefect/client/utilities.py", line 100, in with_injected_client
return await fn(*args, **kwargs)
File "/home/projects/insertions_model/.venv/lib/python3.9/site-packages/prefect/results.py", line 645, in _read_blob
content = await storage_block.read_path(self.storage_key)
File "/home/projects/insertions_model/.venv/lib/python3.9/site-packages/prefect/filesystems.py", line 220, in read_path
raise ValueError(f"Path {path} does not exist.")m
Marvin
09/16/2024, 6:14 PMYour message was too long by 33 tokens - please shorten it and try again.
For reference, here's your message at the allowed limit:
hi - using prefect 2 i executed sub-flows of separate deployments usingHowever when i try to get back the result i'm using the code you have suggested -ofawait asyncio.gathercalls, passing the parameters through serialized jsons, and it seems to have been executed on remote machine as expected.run_deployment
await flow_run.state.result().get()s
Srul Pinkas
09/16/2024, 6:14 PMhi @Marvin - using prefect 2 i executed sub-flows of separate deployments using
await asyncio.gatherrun_deploymentawait flow_run.state.result().get()m
Marvin
09/16/2024, 6:15 PMSo you ran sub-flows as separate deployments using
await asyncio.gatherawait flow_run.state.result().get()await flow_run.state.result().get()State.result()fetch=TrueRemoteFileSystem Copy code
python
from prefect.filesystems import RemoteFileSystem
from prefect.deployments import run_deployment
from prefect import flow, task
import asyncio
# Setting up Remote File System (GCS example)
gcs_block = RemoteFileSystem(basepath="<gcs://my-bucket/folder/>")
gcs_block.save("gcs-storage")
@task
def save_result(data, filename):
gcs_block.write_path(filename, data)
@task
def load_result(filename):
return gcs_block.read_path(filename)
@flow
async def main_flow(data):
# Save data for later retrieval in sub-flow
save_result(data, "large_dataset.json")
# Trigger sub-flow deployment
flow_run = await run_deployment("sub_flow", parameters={"filename": "large_dataset.json"})
# Await result
sub_result_state = await flow_run.state.result(fetch=True)
sub_result = await load_result(sub_result_state.result())
return sub_result
@flow
async def sub_flow(filename):
data = load_result(filename)
processed_data = await process_data(data)
save_result(processed_data, "processed_data.json")
return "processed_data.json"
async def main():
data = await fetch_large_data_somehow()
result = await main_flow(data)
print(result)
if __name__ == "__main__":
asyncio.run(main())s
Srul Pinkas
09/17/2024, 9:45 AM@Marvin can i just get the response back to the parent run as return-values, without sharing file-system between runs, perhaps using
fetch=Truem
Marvin
09/17/2024, 9:45 AMGreat question! You can indeed get the sub-flow's result back to the parent flow run directly using serialized return values, without needing a shared file system.
With Prefect 2.x, you can use the
fetch=True Copy code
python
from prefect.deployments import run_deployment
from prefect import flow, task
import asyncio
@task
async def process_data(data):
# Simulate some processing on the data
return data + " processed"
@flow
async def sub_flow(data):
result = await process_data(data)
return result
@flow
async def main_flow(data):
# Trigger sub-flow deployment
flow_run = await run_deployment("sub_flow", parameters={"data": data})
# Await result with fetch=True
sub_result_state = await flow_run.state.result(fetch=True)
sub_result = sub_result_state.result()
return sub_result
async def main():
data = "large data to be processed"
result = await main_flow(data)
print(f"Processed result: {result}")
if __name__ == "__main__":
asyncio.run(main())run_deploymentstate.result(fetch=True)fetch=Trues
Srul Pinkas
09/17/2024, 2:03 PM@Marvin can i do this trick (running deployment remotely, then collecting results using
awaitfetch=Trueasyncio.gatherSrul Pinkas
09/17/2024, 2:14 PM@Marvin did you see my msg?
m
Marvin
09/17/2024, 2:20 PMAbsolutely! You can run multiple deployments concurrently using
asyncio.gatherawaitfetch=Truerun_deploymentasyncio.gather Copy code
python
from prefect.deployments import run_deployment
from prefect import flow, task
import asyncio
@task
async def process_data(data):
# Simulate some processing on the data
return data + " processed"
@flow
async def sub_flow(data):
result = await process_data(data)
return result
@flow
async def main_flow(data_list):
# Trigger multiple sub-flow deployments concurrently
deployment_futures = [
run_deployment("sub_flow", parameters={"data": data})
for data in data_list
]
# Gather all sub-flow executions
flow_runs = await asyncio.gather(*deployment_futures)
# Collect results from each sub-flow run
result_futures = [
flow_run.state.result(fetch=True)
for flow_run in flow_runs
]
# Await all result futures
results = await asyncio.gather(*result_futures)
# Extract the actual results from the states
final_results = [result.result() for result in results]
return final_results
async def main():
data_list = ["large data 1", "large data 2", "large data 3"]
results = await main_flow(data_list)
for idx, result in enumerate(results):
print(f"Processed result {idx + 1}: {result}")
if __name__ == "__main__":
asyncio.run(main())deployment_futuresrun_deploymentdata_listawait asyncio.gather(*deployment_futures)result_futuresfetch=Truefinal_resultsasyncio.gatherMarvin
09/17/2024, 2:21 PMYes, I saw your message. Based on your requirements, here is the adjusted approach focused more on your specific use case of asynchronously running deployments and collecting their results without a shared filesystem:
### Example: Running Deployments with Asyncio and Fetching Results
Copy code
python
from prefect.deployments import run_deployment
from prefect import flow
import asyncio
@flow
async def sub_flow(data):
# Simulate data processing
processed_data = data + " processed"
return processed_data
@flow
async def main_flow(data_list):
# Run multiple sub-flow deployments concurrently using asyncio.gather
deployment_futures = [
run_deployment(
name="sub_flow/deployment_name", # replace with your deployment name
parameters={"data": data}
)
for data in data_list
]
# Gather all sub-flow runs
flow_runs = await asyncio.gather(*deployment_futures)
# Fetch results from each sub-flow run
result_futures = [
flow_run.result(fetch=True)
for flow_run in flow_runs
]
# Await all result futures
results = await asyncio.gather(*result_futures)
# Extract the actual results from the states
final_results = [result.result() for result in results]
return final_results
async def main():
data_list = ["large data 1", "large data 2", "large data 3"]
results = await main_flow(data_list)
for idx, result in enumerate(results):
print(f"Processed result {idx + 1}: {result}")
if __name__ == "__main__":
asyncio.run(main())run_deploymentrun_deploymentdeployment_futuresasyncio.gatherflow_runsfetch=Trueflow_run.result(fetch=True)s
Srul Pinkas
09/17/2024, 2:38 PMhi @Marvin i am using the suggested snippets and keep getting that some path
does not exist.gatherread_path(self.storage_key)m
Marvin
09/17/2024, 2:39 PMIt sounds like the issue lies in how Prefect is handling result fetching in a distributed setup. Given that the error indicates a missing path, it suggests that Prefect might be attempting to access file paths that do not exist locally on the worker machines.
To ensure proper serialization and fetching of results directly (without relying on filesystem paths), you might need to explicitly configure the result storage backend. Prefect 2.x uses
Resultrun_deploymentawait state.result(fetch=True)RemoteFileSystem Copy code
python
from prefect.filesystems import S3
s3_block = S3(bucket_path="my-bucket/path/to/results/")
s3_block.save(name="my-s3-block", overwrite=True)Result Copy code
python
from prefect.results import S3Result
from prefect.deployments import run_deployment
from prefect import flow, task
import asyncio
s3_result = S3Result(bucket="my-bucket", location="path/to/results/")
@task(result=s3_result)
async def process_data(data):
return data + " processed"
@flow(result=s3_result)
async def sub_flow(data):
return await process_data(data)
@flow(result=s3_result)
async def main_flow(data_list):
# Trigger multiple sub-flow deployments concurrently
deployment_futures = [
run_deployment(
name="sub_flow/deployment_name", # replace with your deployment name
parameters={"data": data}
)
for data in data_list
]
# Gather all sub-flow runs
flow_runs = await asyncio.gather(*deployment_futures)
# Fetch results from each sub-flow run
result_futures = [
flow_run.result(fetch=True)
for flow_run in flow_runs
]
# Await all result futures
results = await asyncio.gather(*result_futures)
# Extract the actual results from the states
final_results = [result.result() for result in results]
return final_results
async def main():
data_list = ["large data 1", "large data 2", "large data 3"]
results = await main_flow(data_list)
for idx, result in enumerate(results):
print(f"Processed result {idx + 1}: {result}")
if __name__ == "__main__":
asyncio.run(main()) Copy code
python
from prefect.filesystems import GCS
gcs_block = GCS(bucket="my-bucket/path/to/results/")
gcs_block.save(name="my-gcs-block", overwrite=True)GCSResultS3Resultbucket_pathlocations
Srul Pinkas
09/17/2024, 2:40 PM@Marvin but this requires them to share some kind of file-system. i thought you said this is not required and i can get the flow output directly from the run, even without setting a shared one
m
Marvin
09/17/2024, 2:41 PMThank you for the clarification. Indeed, it's possible to directly fetch the results from the sub-flows without relying on a shared filesystem, as long as the results are properly serialized and stored within Prefect's result handling system.
Let's focus on achieving this without requiring an external file system:
### Approach:
1. Utilize Prefect's Built-in Result Handling: Prefect automatically stores and retrieves results from sub-flows if using the
result()fetch=True Copy code
python
from prefect.deployments import run_deployment
from prefect import flow, task
import asyncio
@task
async def process_data(data):
# Simulate some processing on the data
return data + " processed"
@flow
async def sub_flow(data):
result = await process_data(data)
return result
@flow
async def main_flow(data_list):
# Trigger multiple sub-flow deployments concurrently using asyncio.gather
deployment_futures = [
run_deployment(
name="sub_flow", # replace with your deployment name
parameters={"data": data}
)
for data in data_list
]
# Gather all sub-flow runs
flow_runs = await asyncio.gather(*deployment_futures)
# Fetch results from each sub-flow run
result_futures = [
flow_run.state.result(fetch=True)
for flow_run in flow_runs
]
# Await all result futures
results = await asyncio.gather(*result_futures)
# Extract the actual results from the states
final_results = [result.result() for result in results]
return final_results
async def main():
data_list = ["large data 1", "large data 2", "large data 3"]
results = await main_flow(data_list)
for idx, result in enumerate(results):
print(f"Processed result {idx + 1}: {result}")
if __name__ == "__main__":
asyncio.run(main())run_deploymentresult(fetch=True)asyncio.gatherLocalResults
Srul Pinkas
09/17/2024, 2:47 PMthanks @Marvin. how do i check the configuration on server? what should i look for, to see if i use in-memory result for storage as default?
m
Marvin
09/17/2024, 2:47 PMTo ensure you are using in-memory result storage in Prefect 2.20, and to check the relevant configurations on the server, here is what you need to verify:
### Prefect Default Result Handling
1. In-Memory Results: By default, Prefect keeps results in memory unless specified otherwise. This means that if no external result storage (like S3 or GCS) is configured, results should be managed in memory.
### Checking Configuration
#### Prefect Cloud or Server Configuration
1. UI Configuration: You can check the result storage settings from the Prefect Cloud/Server UI.
2. Result Persistence:
- Check flow and task configurations to ensure they are not explicitly set to use external storage blocks.
Copy code
python
# Example of setting in-memory results at both flow and task level
from prefect import task, flow
@task
def example_task():
return "task result"
@flow
def example_flow():
result = example_task()
return result
if __name__ == "__main__":
example_flow()PREFECT_RESULTS_DEFAULT_RESULT_CLASS Copy code
python
from prefect import flow, task
@task
def process_data(data):
return data + " processed"
@flow
def sub_flow(data):
result = process_data(data)
return result
@flow
def main_flow(data_list):
# Trigger sub-flows
sub_flow_results = [sub_flow(data) for data in data_list]
return sub_flow_results
if __name__ == "__main__":
main_flow(["data1", "data2", "data3"]) Copy code
python
@task(result=None)
@flow(result=None)s
Srul Pinkas
09/17/2024, 5:10 PMhi @Marvin: i ran your specific exact example-flow and it crashed just like mine over
Finished in state Failed('Flow run encountered an exception. ValueError: Pathm
Marvin
09/17/2024, 5:11 PMYour issue seems to be tied to result handling configurations that might be causing Prefect to try and persist results to a file system, despite your goal to use in-memory storage.
### Key Points from Documentation and Issues:
1. Default Result Storage:
- Prefect can sometimes default to using file system storage if not explicitly configured otherwise.
-
persist_resultFalsepersist_resultFalse Copy code
python
from prefect import flow, task
import asyncio
@task(persist_result=False)
async def process_data(data):
return data + " processed"
@flow(persist_result=False)
async def sub_flow(data):
result = await process_data(data)
return result
@flow(persist_result=False)
async def main_flow(data_list):
# Trigger multiple sub-flow deployments concurrently using asyncio.gather
deployment_futures = [
run_deployment(
name="sub_flow", # replace with your deployment name
parameters={"data": data}
)
for data in data_list
]
# Gather all sub-flow runs
flow_runs = await asyncio.gather(*deployment_futures)
# Fetch results from each sub-flow run
result_futures = [
flow_run.state.result(fetch=True)
for flow_run in flow_runs
]
# Await all result futures
results = await asyncio.gather(*result_futures)
# Extract the actual results from the states
final_results = [result.result() for result in results]
return final_results
async def main():
data_list = ["large data 1", "large data 2", "large data 3"]
results = await main_flow(data_list)
for idx, result in enumerate(results):
print(f"Processed result {idx + 1}: {result}")
if __name__ == "__main__":
asyncio.run(main())PREFECT_RESULTS_DEFAULT_RESULT_CLASS.prefect/config.tomlpersist_result=Falses
Srul Pinkas
09/17/2024, 5:15 PM@Marvin is there anything similar to Airflow xcomm object to retract (in retro) results of runs, even once the remote-machine has finished (and perhaps died)?
m
Marvin
09/17/2024, 5:15 PMYes, Prefect offers similar functionality to Airflow's XComs through the concept of Results. In Prefect, results can be retrieved or rertrieved later even after the flow run has completed, as long as the results were persisted. Here's a breakdown on how to use Prefect's result handling to retrieve results of past runs:
### Persist Results
To enable result retrieval after the flow run has completed, you need to persist the results. This would typically involve using a storage backend like S3, GCS, or some other remote file system.
### Persisted Results Example:
Below is an example of how you might set this up:
Copy code
python
from prefect.filesystems import S3
from prefect.results import S3Result
from prefect import flow, task
# Set up S3 as the result storage
s3_block = S3(bucket_path="my-bucket/path/to/results/")
s3_block.save(name="my-s3-block", overwrite=True)
s3_result = S3Result(bucket="my-bucket", location="path/to/results/")
@task(result=s3_result, persist_result=True)
def process_data(data):
return data + " processed"
@flow(result=s3_result, persist_result=True)
def sub_flow(data):
result = process_data(data)
return result
@flow(result=s3_result, persist_result=True)
def main_flow(data_list):
results = [sub_flow(data) for data in data_list]
return results
if __name__ == "__main__":
main_flow(["data1", "data2", "data3"]) Copy code
python
from prefect.client import Client
async def retrieve_flow_result(flow_run_id):
client = Client()
flow_run = await client.read_flow_run(flow_run_id)
# Assuming S3Result is being used
result = await flow_run.result.fetch()
return result
# Example usage
flow_run_id = "your-flow-run-id"
result = asyncio.run(retrieve_flow_result(flow_run_id))
print(f"Retrieved result: {result}")persist_resultTrueresults
Srul Pinkas
09/17/2024, 5:21 PM@Marvin so if i did not define any file-system (s3 or efs) i can't access the results in retro once the sub-flow has already finished, right?
also, if
PREFECT_RESULTS_PERSIST_BY_DEFAULTfetch=Truem
Marvin
09/17/2024, 5:21 PMYou are correct that if you haven't defined any file system (like S3, EFS, etc.) and
PREFECT_RESULTS_PERSIST_BY_DEFAULTTruePREFECT_RESULTS_PERSIST_BY_DEFAULTFalsefetch=Truepersist_resultTruePREFECT_RESULTS_PERSIST_BY_DEFAULTFalsepersist_result=False Copy code
python
from prefect import flow, task
@task(persist_result=False)
def process_data(data):
return data + " processed"
@flow(persist_result=False)
def sub_flow(data):
result = process_data(data)
return result
@flow(persist_result=False)
def main_flow(data_list):
deployment_futures = [
run_deployment(
name="sub_flow", # replace with your actual deployment name
parameters={"data": data}
)
for data in data_list
]
flow_runs = await asyncio.gather(*deployment_futures)
result_futures = [
flow_run.state.result(fetch=True)
for flow_run in flow_runs
]
results = await asyncio.gather(*result_futures)
final_results = [result.result() for result in results]
return final_results
async def main():
data_list = ["large data 1", "large data 2", "large data 3"]
results = await main_flow(data_list)
for idx, result in enumerate(results):
print(f"Processed result {idx + 1}: {result}")
if __name__ == "__main__":
asyncio.run(main())persist_result=False Copy code
bash
export PREFECT_RESULTS_PERSIST_BY_DEFAULT=Falsepersist_result=Falsepersist_result=Falses
Srul Pinkas
09/17/2024, 5:29 PMhey @Marvin i used the decorator to step over the parameter (in both flow, subflow, and task) - and got a different error:
State data is missing. Typically, this occurs when result persistence is disabled and the state has been retrieved from the APIm
Marvin
09/17/2024, 5:29 PMThe error you're encountering,
State data is missing. Typically, this occurs when result persistence is disabled and the state has been retrieved from the APIfetch=Truepersist_result=False Copy code
python
from prefect.filesystems import S3
from prefect.results import S3Result
from prefect import flow, task
import asyncio
# Set up S3 as the result storage
s3_block = S3(bucket_path="my-bucket/path/to/results/")
s3_block.save(name="my-s3-block", overwrite=True)
s3_result = S3Result(bucket="my-bucket", location="path/to/results/")
@task(result=s3_result, persist_result=True)
async def process_data(data):
return data + " processed"
@flow(result=s3_result, persist_result=True)
async def sub_flow(data):
result = await process_data(data)
return result
@flow(result=s3_result, persist_result=True)
async def main_flow(data_list):
deployment_futures = [
run_deployment(
name="sub_flow/deployment_name", # replace with your deployment name
parameters={"data": data}
)
for data in data_list
]
flow_runs = await asyncio.gather(*deployment_futures)
result_futures = [
flow_run.state.result(fetch=True)
for flow_run in flow_runs
]
results = await asyncio.gather(*result_futures)
final_results = [result.result() for result in results]
return final_results
async def main():
data_list = ["large data 1", "large data 2", "large data 3"]
results = await main_flow(data_list)
for idx, result in enumerate(results):
print(f"Processed result {idx + 1}: {result}")
if __name__ == "__main__":
asyncio.run(main())s
Srul Pinkas
09/17/2024, 5:31 PM@Marvin i want to fetch results and not store anything, this is what i wanted to do all along and can't seem to get it to work. You mentioned
Evaluate Immediate Fetch Strategym
Marvin
09/17/2024, 5:32 PMGot it! Let's focus on fetching results immediately after the sub-flows complete without persisting anything to disk.
### Immediate Fetch Strategy:
To fetch results immediately without persistent storage, ensure the flow and tasks use in-memory storage and fetch the results as soon as the sub-flow completes.
### Steps:
1. Ensure
persist_result=Falsefetch=True Copy code
python
from prefect import flow, task
from prefect.deployments import run_deployment
import asyncio
@task(persist_result=False)
async def process_data(data):
return data + " processed"
@flow(persist_result=False)
async def sub_flow(data):
result = await process_data(data)
return result
@flow(persist_result=False)
async def main_flow(data_list):
# Trigger multiple sub-flow deployments concurrently
deployment_futures = [
run_deployment(
name="sub_flow", # replace with your deployment name
parameters={"data": data}
)
for data in data_list
]
# Gather all sub-flow runs
flow_runs = await asyncio.gather(*deployment_futures)
# Fetch results immediately
results = []
for flow_run in flow_runs:
state = await flow_run.state.fetch()
result = await state.result(fetch=True)
results.append(result)
return results
async def main():
data_list = ["large data 1", "large data 2", "large data 3"]
results = await main_flow(data_list)
for idx, result in enumerate(results):
print(f"Processed result {idx + 1}: {result}")
if __name__ == "__main__":
asyncio.run(main())persist_result=Falsefetch=Trueawait asyncio.sleep(...)s
Srul Pinkas
09/17/2024, 5:41 PM@Marvin but your example assumes the order of finished sub-flows, right? if the second is finished before the first, the await-fetch will still suffer from not waiting properly to the sub-flow?
So if i'm right, this still doesn't help me, i want to run dozens of sub-flow deployments at the same time and can't predict their order
(also, you typed
state = await flow_run.state.fetch()fetchstate = await flow_run.statem
Marvin
09/17/2024, 5:41 PMYou're right; the example provided could suffer from race conditions if the sub-flows finish out of order. To handle a dynamic and unpredictable order of sub-flows finishing, we can use an approach that checks the state after the tasks complete and fetches the results accordingly.
### Corrected Approach:
1. Track and Retrieve States Dynamically: Use
asyncio.gatherstatefetch=True Copy code
python
from prefect import flow, task
from prefect.deployments import run_deployment
import asyncio
@task(persist_result=False)
async def process_data(data):
return data + " processed"
@flow(persist_result=False)
async def sub_flow(data):
return await process_data(data)
@flow(persist_result=False)
async def main_flow(data_list):
# Trigger multiple sub-flow deployments concurrently
deployment_futures = [
run_deployment(
name="sub_flow", # replace with your deployment name
parameters={"data": data}
)
for data in data_list
]
flow_runs = await asyncio.gather(*deployment_futures)
# Fetch results from each sub-flow run dynamically
result_futures = [
flow_run.state.result(fetch=True)
for flow_run in flow_runs
]
results = await asyncio.gather(*result_futures)
# Extract actual results
final_results = [result.result() for result in results]
return final_results
async def main():
data_list = ["large data 1", "large data 2", "large data 3"]
results = await main_flow(data_list)
for idx, result in enumerate(results):
print(f"Processed result {idx + 1}: {result}")
if __name__ == "__main__":
asyncio.run(main())asyncio.gatherfetch=Truefetch=True5 Views