Any idea how to prevent prefect from building up large amounts of working memory? This is seen on a flow with 7 Tasks (6 of which are mapped over a list of ~40,000 elements), and running on a kubernetes agent

hi folks, getting this non-speaking error when i want to run a flow:

INFO - prefect.FlowRunner | Flow run FAILED: some reference tasks failed.

does somebody know, what might be a reason for that?

Hi guys, I have a scheduled prefect flow (runs once a day) that I register once. The flow has several Parameters, two of them being some dates which I would like to have default values with respect to the flow runs, not the flow registration. Now, the default values are set correctly upon flow registration, but on the next day, the same default dates appear, even though I expect them to be one day later. I've set the dates like this:

schedule = Schedule(clocks=[IntervalClock(timedelta(days=1))])
with Flow(name=name, schedule=schedule) as flow:
    start_date = Parameter(name="start_date",
                           required=False,
                           default=(datetime.now() - timedelta(days=2)).strftime("%Y-%m-%d"))
    end_date = Parameter(name="end_date",
                         required=False,
                         default=(datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d"))

Hey community, we are getting the weird and low informative error message

{'type': ['Missing data for required field.']}

as single log directly in the beginning of a flow that directly fails. What could be possible reasons and did you encounter the same? 😮 PS: the flow runs locally but throws this error on Prefect Cloud 🤔

Hi Prefect! Quick question: PyCharm keeps telling me that the documentation-approved way of using the

@task

decorator is lacking a positional argument. For example:

@task
def add_one(x: int) -> int:
    return x + 1


@task()
def add_two(x: int) -> int:
    return x + 2

The

@task

decorator over

add_one

is how I’ve seen it in the docs, but it seems like the decorator for

add_two

is how PyCharm likes it. Am I missing something?

Hello! I am running the hello-world flow and trying to get it running in the cloud. However when I do I get the following error

{'type': ['Unsupported value: UniversalRun']}.

It works locally. Here is my code:

import prefect
from prefect import task, Flow

@task
def say_hello():
    logger = prefect.context.get("logger")
    <http://logger.info|logger.info>("Hello, Cloud!")

with Flow("hello-flow") as flow:
    say_hello()

flow.register(project_name="tutorial")

Hi guys, where is the best place to make license questions?

Hey all, I am running into a weird issue. We are trying to use Prefect to build a docker container that contains a module that depends on Pandas. We have been successfully able to build it and log "hello" to the UI. However, when we begin to use the module, we are unable to rebuild the docker container because it is looking for Pandas. The error it gives is: ModuleNotFoundError: No module named 'pandas' how can we resolve this?

Hi - I am trying to get some direction for the following error when a

StartFlowRun

is run from within another flow. Unexpected error: ClientError('400 Client Error: Bad Request for url: https://api.prefect.io/graphql\nThis is likely caused by a poorly formatted GraphQL query or mutation but the response could not be parsed for more details') Traceback (most recent call last): File "/home/ilivni/miniconda3/envs/py38lmap/lib/python3.8/site-packages/prefect/client/client.py", line 361, in _send_request response.raise_for_status() File "/home/ilivni/miniconda3/envs/py38lmap/lib/python3.8/site-packages/requests/models.py", line 943, in raise_for_status raise HTTPError(http_error_msg, response=self) requests.exceptions.HTTPError: 400 Client Error: Bad Request for url: https://api.prefect.io/graphql The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "miniconda3/envs/py38lmap/lib/python3.8/site-packages/prefect/engine/runner.py", line 48, in inner
    new_state = method(self, state, *args, **kwargs)
  File "miniconda3/envs/py38lmap/lib/python3.8/site-packages/prefect/engine/task_runner.py", line 865, in get_task_run_state
    value = prefect.utilities.executors.run_task_with_timeout(
  File "miniconda3/envs/py38lmap/lib/python3.8/site-packages/prefect/utilities/executors.py", line 299, in run_task_with_timeout
    return task.run(*args, **kwargs)  # type: ignore
  File "miniconda3/envs/py38lmap/lib/python3.8/site-packages/prefect/utilities/tasks.py", line 454, in method
    return run_method(self, *args, **kwargs)
  File "miniconda3/envs/py38lmap/lib/python3.8/site-packages/prefect/tasks/prefect/flow_run.py", line 177, in run
    flow_run_id = client.create_flow_run(
  File "miniconda3/envs/py38lmap/lib/python3.8/site-packages/prefect/client/client.py", line 1094, in create_flow_run
    res = self.graphql(create_mutation, variables=dict(input=inputs))
  File "miniconda3/envs/py38lmap/lib/python3.8/site-packages/prefect/client/client.py", line 298, in graphql
    result = <http://self.post|self.post>(
  File "miniconda3/envs/py38lmap/lib/python3.8/site-packages/prefect/client/client.py", line 213, in post
    response = self._request(
  File "miniconda3/envs/py38lmap/lib/python3.8/site-packages/prefect/client/client.py", line 445, in _request
    response = self._send_request(
  File "miniconda3/envs/py38lmap/lib/python3.8/site-packages/prefect/client/client.py", line 374, in _send_request
    raise ClientError(f"{exc}\n{graphql_msg}") from exc
prefect.utilities.exceptions.ClientError: 400 Client Error: Bad Request for url: <https://api.prefect.io/graphql>
This is likely caused by a poorly formatted GraphQL query or mutation but the response could not be parsed for more details

Thanks in advance

If I have a task (for example winrm) that uploads and executes some scripts, what is the best way with prefect server to get them into the flow. In prefect core, I load the scripts from the local filesystem into a dictionary and pass this into the flow as a parameter when doing flow.run() For prefect server I'd like to use a cron job so I don't have a local script and the scripts must already exist. Store them in bitbucket? S3?

hi guys, how do i define a user friendly logging task? without making the user define a separate task just to concatenate strings?

@prefect.task
def log(item):
	<http://prefect.context.logger.info|prefect.context.logger.info>(item)

this works when you pass in the string output of an upstream task as the entirety of the input to the logging task

log(output)

but what if i want to do something like

log("the output of the previous task was: " + output)

here we have an issue because

+

is not a task and so we will actually be logging something like

the output of the previous task was: Task<blah blah>

Hi, I want to limit my agent to running only a single flow / task. I thought I read about this functionality somewhere, but am unable to find it. Do you know where this is explained?

Hi folks, we're using pyvirtualdisplay for one of the flow, running container in k8s, and got intermittent failure

Failed to load and execute Flow's environment: StorageError("An error occurred while unpickling the flow:\n DisplayNameError('0.0')")

Hello all, I am looking through the documentation and I am struggling to find information on passing volume information to Docker Storage. My flow outputs some files to disk and I would like to mount a volume so that we can see the files that were created on the filesystem, outside of the docker container. How would I accomplish this?

Hey There -- One of the reasons we are considering moving to prefect is to get better testing coverage. I am looking for examples how how to test Flows including any approach to mocking out Tasks. Anyone have any pointers?

Hrm, can you turn a Prefect Flow into a Snowflake External Function? https://docs.snowflake.com/en/sql-reference/external-functions-introduction.html

Hey folks, super excited for the new flow readme feature. Is there documentation anywhere for how to add the readme in code?

I want to deploy prefect UI and Backend on azure app service using docker and docker compose. is it possible? If yes, can someone guide me how I'll be able to deploy ? I'm totally new on prefect. Thanks,

Dear Prefect people, can you please help me find out how to make Prefect (Cloud?) work for my use case? 🧑‍💻 We want to process thousands of satellite images 🛰️🌍 on a Kubernetes cluster, scaling to ~100 TB of RAM and thousands of jobs running in parallel. We already successfully did this with our own little event-based workflow manager (REST APIs, Postgres, Docker, Python K8s API), but it is lacking a lot of the bells and whistles that Prefect has, choking a bit on all the messages, and we would rather like to join the awesome community you have here! ❤️ Using your nice docs and the open source code, we already ran a single processing job on our K8s cluster via a Prefect flow 👍. Now we want to scale this up. Intuitively, I would have one flow process a single image (using 3-8 K8s tasks, ~10 min per task). But we want to scale to 1000s of images getting processed in parallel. Seeing that Prefect Cloud Teams plan only allows for 2️⃣(!) concurrent flows, I get in doubt whether Prefect is built for this number of concurrent flows and all the logs and messages generated from that. ❓ Can you please let me know whether Prefect is meant to scale to 10ths of 1000s of tasks running in parallel? And if so, whether I can run 1000 flows concurrently or should rather have one flow branch out into 1000 tasks with each their subtasks (if that is possible)? Please note that this is burst compute once a month or so, with lower (10s of images) daily loads. I am happy to work with Prefect Core, but Cloud might also be very nice if we can afford it.

Hi, is it possible to run LocalDaskExecutor or local DaskExecutor on Kubernetes agent? I tried it but it seems it doesn't use local Dask cluster at all.

Hi, Is it possible to restart a failed flow from a specific task ? Considering I've handled the failed state with a state handler for that task.

Hi all, I am trying to run a Prefect agent on ECS and have been following this tutorial. I used the below configurations (along with exporting the runner token, aws creds, etc to environment variables). However I am continuously receiving a

ValueError: Failed to infer default networkConfiguration, please explicitly configure using --run-task-kwargs

error. I have tried all teh different combinations I can think of and been through the docs (both prefect and ECS) but cant find a solution that works. Has anyone else been able to solve this?

export networkConfiguration="{'awsvpcConfiguration': {'assignPublicIp': 'ENABLED', 'subnets': ['<EC2 instance subnet>'], 'securityGroups': ['<EC2 instance security group>']}}"
prefect agent ecs start --token $RUNNER_TOKEN_FARGATE_AGENT \
 --task-role-arn=arn:aws:iam::<AWS ACCOUNT ID>:role/ECSTaskS3Role \
 --log-level INFO --label prod --label s3-flow-storage \
 --name prefect-prod

Hi. I was wondering if anyone has noticed that some task-runs get resubmitted when running with prefect-cloud/dask. For example, in the screenshot below. I have a task that completed within 51 minutes. However, the reported completion time shows 9 hours. It seems that sometime during the flow, the task run was resubmitted and hit the cached state, which updated the logs. Does anyone have any ideas why the task resubmitted?

Hi all! I was wondering how we can change the UI port from 8080 to something else. I have changed the

PREFECT__SERVER__UI__PORT

and

PREFECT__SERVER__UI__HOST_PORT

to try and follow the config.toml, but the UI service is still launching and saying “👾👾👾 UI running at localhost:8080 👾👾👾” and I dont know what else to change in the UI service 😞 As a follow up, what is the difference between the UI port and the UI host port? EDIT: I see echo

UI running at localhost:8080

is hardcoded in ./start_server.sh in the UI service, is this meant to be the case?

Does anyone have a recent example of getting DaskExecutor running on my local environment? It seems that most of the conversations in slack are all about using Environments instead of RunConfig.

Hi there, I was wondering if it's possible somehow to

map

on

case

?

with case.map(should_do_the_thing, True):

does not agree...

Gday all! Does prefect server currently support flow concurrency? Heres the scenario - we have a flow that polls for data every 10 minutes and stores it. Works great. However, we want to backfill the data, and need one flow run executed per day back a year (365 flow runs). Now, if I simply use the prefect client to create_flow_run 365 times, the agent schedules them all at once, the API times out, and everything crashes (see fun image below). On top of this, it would be ideal if I could have a priority of flows to be run if theres concurrency so that the regular polling doesnt get stuck in an execution queue (not an issue if we have concurrency available via agents, I can have a backfill agent and a real time agent, but unsure if theres support like that). I can see task concurrency in the doco, but have missed flow concurrency itself. Is this supported?

Is there a good way to set a predefined prefect task to have an upstream dependency? I would like the sql_task to wait on the password, but that's got the graph that I'm getting.

sql_task = SqlServerFetch(
        db_name=prefect.config.sql_server.database,
        user=prefect.config.sql_server.user,
        host=prefect.config.sql_server.server,
        query=get_manual_override_rows,
        fetch='many',
        fetch_count=3,
        result=result_formatter,
        name="Database Query"
        # commit: bool = False,
)

with Flow("config_flow") as f:

    mode = Parameter("Mode", default="dev")

    password_var  = set_creds(mode)
    password_task = EnvVarSecret()
    password = password_task(password_var)
    s = sql_task(password=password)

    v = view_sql(sql_task)

Hi Everyone, I am prototyping an ETL flow where I want to read a bunch of files, do some transformation, and write the results in parallel using a

LocalDaskExecutor

My flow looks something like this:

list_of_files = Parameter()
list_of_frames = extract_fn.map(list_of_files) # This is indeed getting executed in parallel

How can I pass both the result of the previous mapped fn and also the input parameters to my transform fn? I understand that I could probably refactor the transform_fn to avoid using the

list_of_files

but wanted to know if there is any other way of handling this.

transform_fn.map(list_of_frames, list_of_files)

As you can tell I am very new to prefect. Thanks a lot!

Is there a way to define a function at runtime? I have a sql task that I would like to pass a username and password at runtime, and I'm struggling pretty hard to get it moving. More in the thread