Philip MacMenamin
06/23/2020, 8:50 PMurllib
, and saves it in $RUN_LOC/my_file
• execs a task which runs a bash command - eg wc -l $RUN_LOC/my_file > file_len
Obviously the tasks here are silly. My aim is to have an example of something which I can have a series of tasks do work on files. The task results are almost not important, the result might be nothing more than a RAN_OK/ NOT_OK. What matters to me is I can have a mechanism to operate on files, and shell out to other utils to operate on files and check their return status.
I've been looking at https://docs.prefect.io/core/concepts/results.html#how-to-configure-task-result-persistence
from prefect import task, Flow
from prefect.engine.results import LocalResult
@task(result=LocalResult(location="initial_data.prefect"))
def root_task():
return [1, 2, 3]
@task(result=LocalResult(location="{date:%A}/{task_name}.prefect"))
def downstream_task(x):
return [i * 10 for i in x]
with Flow("local-results") as flow:
downstream_task(root_task)
Zachary Hughes
06/23/2020, 9:03 PMPhilip MacMenamin
06/23/2020, 9:05 PMfile_a
and it knows that it needs to open file_a for that specific run, and I wouldn't have to keep explicitly pass around full paths. But that's for neatness, that's not a deal breaker I suppose.open file_a
as opposed to having an argument getting passed in at the top of the task, and prepending this to the file path, and then looking up where this tasks set of files would be written, and then return that path on for the next task.
I hope I'm making sense here. 😕~/.prefect/config.toml
fileZachary Hughes
06/23/2020, 9:20 PMShellTasks
, potentially with a Python task sandwiched in between to download your file.Philip MacMenamin
06/23/2020, 9:21 PMZachary Hughes
06/23/2020, 9:23 PMfrom prefect import task, Flow
from prefect.tasks.shell import ShellTask
directory = "testing456"
mkdir_task = ShellTask()
# in your use case you'd download the file
# rather than using touch_task
touch_task = ShellTask()
with Flow("shelltaskflow") as flow:
mkdir_task(command=f"mkdir {directory}")
touch_task(command=f"touch {directory}/test.txt")
out = flow.run()
testing456
for a UUID or a timestamp or something along those lines that floats your boat.Philip MacMenamin
06/23/2020, 9:25 PMtesting456
in... the ~.prefect/results dir?Zachary Hughes
06/23/2020, 9:25 PMtesting456
in whatever directory you ran your flow in.helper_script
argument.
https://docs.prefect.io/api/latest/tasks/shell.html#shelltaskPhilip MacMenamin
06/23/2020, 9:26 PMZachary Hughes
06/23/2020, 9:29 PMPhilip MacMenamin
06/23/2020, 9:29 PMZachary Hughes
06/23/2020, 9:31 PMPhilip MacMenamin
06/23/2020, 9:32 PMZachary Hughes
06/23/2020, 9:32 PM