Hi! Is it possible to run flows on different servers without kubernetes or aws cluster?

Is it possible to configure timeouts for a flow? I know that they can be configured per task - I was hoping to configure them at the flow level

hi prefect-community! is there any way to force a(n over)write of a checkpointed/cached task result? say i have an S3Result, can i call write on it to overwrite the previously checkpointed result (something like

s3result.write(overwrite=True)

)?

Hi there. I think this is a simple question! I have some flows (generally many hours long) where the last task fails but the flow continues to run. The last task happens to be an AWSClientWait on a Batch job but I don't know if this is the reason or not. Or alternatively is there a way to fail a flow if any of the tasks fail? Thanks as always!

Is there an updated example of using dask from within a Prefect flow? I.e. a dask that calls

dask.compute()

using a Dask Executor? I'm trying to compare my upgraded (0.13.x to latest) flow code with a working example and right now my

compute

calls hang.

Hi all. We're just starting to use Prefect & Dask to coordinate our open source US energy data processing pipeline so we can do complete nightly continuous integration runs with all of the data, instead of just a small subset. Also interested in the potential for integration with Airbyte and dbt. https://github.com/catalyst-cooperative/pudl

When trying to build my flow (docker storage) with

prefect build

I am running into an error

mkdir: cannot create directory '/opt/prefect/': Permission denied

- I am creating a user in my Dockerfile thru

useradd

it appears there are extra commands prefect appends to the user defined Dockerfile - is the root user required?

Hey 👋 We just hit this internal serve error on one of our scheduled runs. We have hit restart, but wondering whether there is any further context that can be provided?

Failed to set task state with error: ClientError([{'path': ['set_task_run_states'], 'message': 'An unknown error occurred.', 'extensions': {'code': 'INTERNAL_SERVER_ERROR'}}])
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/site-packages/prefect/engine/cloud/task_runner.py", line 91, in call_runner_target_handlers
    state = self.client.set_task_run_state(
  File "/usr/local/lib/python3.8/site-packages/prefect/client/client.py", line 1917, in set_task_run_state
    result = self.graphql(
  File "/usr/local/lib/python3.8/site-packages/prefect/client/client.py", line 569, in graphql
    raise ClientError(result["errors"])
prefect.exceptions.ClientError: [{'path': ['set_task_run_states'], 'message': 'An unknown error occurred.', 'extensions': {'code': 'INTERNAL_SERVER_ERROR'}}]

I've put the task and flow runs urls in the 🧵 .

hi! I can't see stdout messages from a kubernetes job in prefect cloud, I have this on my jobs

PREFECT__CLOUD__SEND_FLOW_RUN_LOGS=true
PREFECT__LOGGING__LOG_TO_CLOUD=true

I'm using

@task(log_stdout=True)

and I can see the message in kubernetes it self. I can see the message "hey budy" in the log as you can see here, but is not present in the cloud UI am I missing something?

Hi all, let's say that I have a flow scheduled and running on cloud that launches another python process/program/script via a local agent and it runs on the weekends. Is there an agent that can persist on my local machine in the event that my machine restarts?

Hi i have a problem when i define three task and add call them one by one under a flow they didn't run step by step !! how can i say to a flow to how run task (which task 1st which one 2nd and which one 3rd) ??? I'm adding a sample code as an reply to my message

#dask error. Unable to reach tcp ip Timed out trying to connect to tcp://172.16.0.5:8786 after 30 s while executing

executor = DaskExecutor(address="<tcp://127.0.0.1:8787>")
flow.run(executor=executor)

I have errors when apply the prefect_agent.yaml, please see https://stackoverflow.com/questions/70260659/prefect-kubernetes-agent-throws-authorizationerror-when-start

I have tried to follow the document to do some troubleshooting, but can’t figure it out.

I want to know what exactly does the prefect agent pull from storage ? I'm asking this because in the main flow file which is

with Flow(...) as flow:
    a = first_task()
    b = second_task()

say there are tasks which are defined and imported from other files (which indeed call other tasks and so on..) There is no way I can look at just this main file and tell how the entire flow is gonna look like ? What kind of dependencies all the tasks will have, retries, etc, etc Basically there isn't much info I can get . So what are we doing here ? (and why?)

Hi, I have a task that should be executed each 30 minutes; sometimes task is running longer than 30 minutes, and at the same time another instance of this Task is started in parallel which is causing logical and performance issues. Is there any option not to start the same Task if previous execution is not finished yet for non-cloud? (seems like for cloud it's possible using labels and number of parallel executions allows).

Is there a Prefect cloud/UI equivalent of this? https://docs.prefect.io/core/idioms/flow-to-flow.html

Is it expected that running a local flow with

PREFECT__FLOWS__CHECKPOINTING=false

but with checkpoint data present in the prefect result directory would read from those results? - I would expect this, but this is the current behaviour (on my machine ...)

Hello! I’m trying to create a convert an existing script into a prefect flow, and the

task.map

functionality doesn’t quite work as I had expected. I’m aggregating a bunch of data sources and rows into a DataFrame, and previously this had been sped up by running it over subsets of keys, with each subset running in a different process. So in prefect, I have a task that splits the keys into subsets, and then maps a task over those subsets. The problem is that there’s a small probability of failure on each subset, due to connection timeouts. I have added some retry logic, but still want to make sure that successful sub-tasks have their results checkpointed, and unsuccessful ones are not. But the

results = task.map(subset=subsets)

code instead just stores the

results

in a single file, and then does not re-run unsuccessful sub-tasks. I tried adding

{map_index}

to the task target filename pattern, but this did not work (_update_: it does work, I just had extra

{

brackets, i.e.,

{{map_index}}

🤦). Here’s the basic flow:

all_keys = get_keys_task()
key_subsets = partition_keys_task(all_keys, n_subsets)
data_subsets = get_data_task.map(keys=key_subsets)
all_data = concatenate_subsets_task(data_subsets=data_subsets)

I know I can work around this by writing my own utility to create a list of tasks with their own unique task names, but it seems like part of the point of

.map

ought to be to do this sort of results management for you... Any tips? Maybe there’s just a parameter in the

prefect.task

decorator or the

task.map

function I don’t know about?

Hello, we’ve recently noticed that the exception logs in Production are printing out personal home folder path. Do you know how this is happened?

@task(log_stdout=True)
def test():
    print(f'current folder: {Path.cwd()}')
    raise Exception('test exception')

Task 'test': Exception encountered during task execution!
Traceback (most recent call last):
  File "/usr/local/lib/python3.7/site-packages/prefect/engine/task_runner.py", line 863, in get_task_run_state
    logger=self.logger,
  File "/usr/local/lib/python3.7/site-packages/prefect/utilities/executors.py", line 445, in run_task_with_timeout
    return task.run(*args, **kwargs)  # type: ignore
  File "/Users/tonyyun/github/data_operations/dbt_automation/flow.py", line 296, in test
Exception: test exception

, where

/Users/tonyyun/github

shouldn’t be there^

We are using prefect cloud. We are curious how to get audit logs out of the server to push them to our monitoring system? Is there something we can use that comes with the system for exporting. Is there any documentation on this?

hi - new to prefect, just reading the docs and see the section on "flow storage", i guess Im trying to follow the developer workflow here... so I code a python file that defines my flow and its tasks... submit this to prefect, and it uses flow storage to actually store that script? (i,e. vs me coding a flow, checking it into git, then deploying my flows to prefect..) I guess Im not sure I'm following this.

Is there a way to set up an if/else statement within a flow depending on if a SQL query errors out? For example,

if(select * from table) = error then (alter table add column) else somethingelse

?

Hi there! I hope today's AWS outages have not been too stressful for the team. Due to issues in us-east-1, my prefect agents are down, and a number of flows were marked as "late". I canceled several of them manually, and then clicked "Clear late runs" in the UI. Can you clarify what this actually does?

Hello, question on Agents to understand the doc here : https://docs.prefect.io/orchestration/agents/overview.html

A single agent can manage many concurrent flow runs

-> does it mean that if I schedule multiple flow at the same time for the same agent, the agent will be able to treat them parallelly ? Or does it queue the jobs ? Thanks

Hi! I’m working on a flow which requires large concurrency (tens of thousands of tasks). Previously I’ve used the

LocalDaskExecutor

as my requirements for concurrency were low. However with this flow I want to use the real

DaskExecutor

. As I’m aware Prefect is capable of creating a temporal Dask cluster for the running flow (using

KubeCluster

), but alternatively can use an existing deployed Dask cluster (or even dask gateway if I’m not mistaken). Note, I’m running in Kubernetes, so adaptive scaling could be interesting. Are there any guidelines / suggestions / experiences for using a temporal vs. static Dask cluster? Additionally, if I have

Docker

as storage, how can I supply the correct image tag containing my modules/dependencies while registering the flow?

Hi! I am trying to run

prefect==2.0a5

in Docker. Here is my Dockerfile:

FROM python:3.8
RUN adduser prefect
USER prefect
WORKDIR /home/prefect
COPY --chown=prefect:prefect <http://requirements.in|requirements.in> ./
ENV LANG C.UTF-8
ENV LC_ALL C.UTF-8
ENV PATH="/home/prefect/.local/bin:${PATH}"
RUN pip install --user --no-cache-dir -r <http://requirements.in|requirements.in>
COPY --chown=prefect:prefect flow.py flow_deployment.py ./

Inside the container I create a deployment and several flow runs, but all of them are marked as “late”. Here is the terminal output:

$ prefect orion start --host 0.0.0.0 --log-level DEBUG
Starting Orion API server...
INFO:     Started server process [71]
INFO:     Waiting for application startup.
09:54:06.189 | Scheduler service scheduled to start in-app
09:54:06.189 | MarkLateRuns service scheduled to start in-app
INFO:     Application startup complete.
INFO:     Uvicorn running on <http://0.0.0.0:4200> (Press CTRL+C to quit)
09:54:06.501 | Finished monitoring for late runs.
09:54:06.538 | Scheduled 0 runs.
Starting agent connected to <http://0.0.0.0:4200/api/>...
Agent started! Checking for flow runs...
09:54:07.298 | Submitting flow run 'f0855bd3-2eab-4346-ad3a-2e237a688faa'
09:54:07.298 | Submitting flow run '521624eb-cec8-4f9b-9e92-f203e104586a'
09:54:07.298 | Submitting flow run 'f25da7b6-7893-4107-aaa3-df22377e2ccf'
09:54:07.299 | Submitting flow run '2895bdfa-082c-43b5-afc2-d0dcc269bf51'
09:54:07.299 | Submitting flow run '2b445dbd-58b4-4acb-89b7-1f6782dc0ec9'
09:54:07.300 | Completed submission of flow run 'f0855bd3-2eab-4346-ad3a-2e237a688faa'
09:54:07.300 | Completed submission of flow run '521624eb-cec8-4f9b-9e92-f203e104586a'
09:54:07.300 | Completed submission of flow run 'f25da7b6-7893-4107-aaa3-df22377e2ccf'
09:54:07.300 | Completed submission of flow run '2895bdfa-082c-43b5-afc2-d0dcc269bf51'
09:54:07.300 | Completed submission of flow run '2b445dbd-58b4-4acb-89b7-1f6782dc0ec9'
09:54:08.969 | Flow run '521624eb-cec8-4f9b-9e92-f203e104586a' exited with exception: KeyError('__main__')
09:54:08.975 | Flow run '2895bdfa-082c-43b5-afc2-d0dcc269bf51' exited with exception: KeyError('__main__')
09:54:08.976 | Flow run 'f25da7b6-7893-4107-aaa3-df22377e2ccf' exited with exception: KeyError('__main__')
09:54:08.979 | Flow run '2b445dbd-58b4-4acb-89b7-1f6782dc0ec9' exited with exception: KeyError('__main__')
09:54:08.980 | Flow run 'f0855bd3-2eab-4346-ad3a-2e237a688faa' exited with exception: KeyError('__main__')

Is there a way to debug these KeyErrors?

Good day, Have a question about logging. We have a separate own library "LIB" that is doing some magic. This library has own root logger defined

logger = logging.getLogger()

. The idea is to extend Prefect Task class and run some magic using LIB library. I've read that we can add as many NAMED loggers as we want using https://docs.prefect.io/core/concepts/logging.html#extra-loggers, but since with have root one, what happens now is: Prefect logs its records as usual, but messages from LIB are not passed. I suppose this is due to LIB logger is root one. Have you ever faced such situation?

Hi guys, maybe someone can point me in the right direction: I have successfully deployed Prefect to a local Kubernetes cluster, have UI & Graphql access, but I am a little lost with the deployment of Flows to the cluster... As far as I got it I need to create a Dockerimage, and then a flow using KubernetesRun, right? But how do I deploy it then?

flow = Flow("taskname")

flow.run_config = KubernetesRun(env={"POSTGRES_USER": "2234234",
                                    ...
                                },
                                image = "dockerimage:latest"
                                )

flow.register('projectname')

Hi. Is it any way to pass env variable in ECSRun(env={...}) as Prefect Secret?