On the installation page, the possibility of installing “extras” is mentioned, but I can’t figure out where the full list is. I guess a link would help. (Further, the reason why I landed on this page is because it seems I need the

[git]

extra to be installed for the Git Storage to work, which is not mentioned in the doc)

Hi, We have a flow that consists of multiple flows. There are two parent flows on which multiple flows depend. The two parent flows are independent of each other. I see the flows execute serially satisfying dependency constraints... but don't see any nondependent flows running in parallel. Should we expect the two independent flows to be scheduled in parallel or they would still run serially one after the other?

Hey guys, I am running the same flow on two different agents both with same configuration, the flow runs successfully on one ages while it fails on the other agent, with this error

Failed to load and execute Flow's environment: FlowStorageError('An error occurred while unpickling the flow:\n  TypeError("int() argument must be a string, a bytes-like object or a number, not \'NaTType\'")')

Any idea why?

Hi Folks, a question about tasks that return a tuple of items and prefect task mapping.

Hey Team, When our agents go down due to a hosed host, we find that we can get an email notification via automations setup (agent is unhealthy). However, when we change it to pagerduty, we find we do not get alerts. How can we resolve this?

Docker doesn't play well with academic clusters because of the root-privilege requirements, but has anyone used Prefect with Singularity? It's an alternative to Docker but can still pull from Docker Hub images. Prefect storage and agent types for Singularity would be really helpful in academic HPC use cases.

is there a simple way, like a pinging some ip address, that can tell me if I have access from inside a production environment to the cloud service ?

What is the purpose of the

mapped

argument in

Task.set_upstream

? is this just another way of applying mapping, or should we always be setting

mapped=True

if the upstream task is mapped? (we usually use

Task.map()

when mapping tasks)

Hi folks, can I have nested tasks?

@task
def f1():
  ...
@task
def f2():
  f1()
  ...

Can I/Is it a good practice to put a result of a task inside the

prefect.context

to be used in all subsequent tasks? I have a stateful variable that many tasks will use and would love to use it inside many tasks without passing in as parameter

Running into an issue where some mapped tasks fail with state TimedOut without an exception or apparent failures in the task run logs. Are there specific conditions that would cause a task to end in TimedOut state? (Feel free to point me to docs or code if needed.) Example log output in thread reply:

I want to use prefect to validate api data from an endpoint; is there a good way to report the results of the validations for example data for this ID was wrong

Hi general question here When running flows lately I have been noticing lately long waits for workers, hour+, and when I do happen to get a worker upon creation of a cluster its only ever one worker. Is this common? I tried just creating coiled clusters and they have no problem getting workers. Any help is appreciated thanks.

Hi, it seems that when running flows locally with the Prefect CLI it will abort if passed parameters that don't exist in the flow, but when running registered flows it will pass them through and run with no errors. Is there a way for the CLI to validate parameters and show an error if a param doesn't exist? We'd like to catch this in case someone passes a misspelled param name, for example.

# Running the flow locally
aric@aric-dev:~/src/conductor$ prefect run -p conductor/samples/hello_world_flow.py --param foo=bar
Retrieving local flow... Done
Configured local flow run
└── Parameters: {'foo': 'bar'}
Running flow locally...
Flow runner encountered an exception!
  Traceback (most recent call last):
    File "/home/aric/.local/lib/python3.8/site-packages/prefect/cli/run.py", line 590, in run
    result_state = flow.run(parameters=params_dict)
    File "/home/aric/.local/lib/python3.8/site-packages/prefect/core/flow.py", line 1250, in run
    raise ValueError(
  ValueError: Flow.run received the following unexpected parameters: foo

Flow run failed!

# Running the same flow registered on Prefect Cloud
aric@aric-dev:~/src/conductor$ prefect run -n "Hello World" --param foo=bar
Looking up flow metadata... Done
Creating run for flow 'Hello World'... Done
└── Name: monumental-bat
└── UUID: 39c7896b-0bee-4015-be99-648d90da201b
└── Labels: []
└── Parameters: {'foo': 'bar'}
└── Context: {}
└── URL: <https://cloud.prefect.io/><URL>

Hi all, has anyone toyed with deployment of flows from S3? I’d like to speed up our CI/CD process, and wondering if it’d make sense to have a “meta” Flow fetch a python package of other Flows from S3, download and register them, etc. Airflow has had this on their to-do list for years (https://cwiki.apache.org/confluence/display/AIRFLOW/AIP-5+Remote+DAG+Fetcher), and I’d love for Prefect to beat them to the punch. Reducing deployment from 20-30 minutes to seconds would be a game-changing differentiator

What is the best approach to calling an async function within a Prefect task? I have to call the Azure Eventhub client which is async all the way down, from my straight synchronous Prefect task. I've tried calling the async functions like so;

asyncio.run(client.send_batch(event_data_batch))

but I get errors like

RuntimeError: Event loop is closed

Any advice?

Hi all, I'm triggering my flow with AWS lambda and this great example answered most of my questions: https://github.com/cicdw/prefect-cloud-lambda . But I'm just wondering, is there an easy way to get latest flow_id for my flow_group, rather than hardcoding flow_id in there (and changing every time I deploy a new version)? Is there another env var I can specify maybe?

Hi everyone. I have a flow that triggers another flow using

create_flow_run

and

get_task_run_result

. I am using these two tasks inside of another task so this can be retried as a unit (see code in thread). Last night, the child flow was triggered three times (triggered by the retries in the parent flow) and all three times the flow succeeded so it looks like there was an issue where the parent flow misread the child flow's status. I am attaching the logs in the thread. Am I misusing these tasks? Any ideas as to why this could be happening? This has been running every night for a couple of weeks and it's the first time this problem happens. Thanks!

Hey everyone, do you know any resources for best practises / guides regarding production deployment of agents? I'm looking at https://docs.prefect.io/orchestration/agents/overview.html#agent-types and am wondering whether a single vm using a docker agent would be fine. I am looking for something with very little overhead as our team is very small (3 people, so I feel kubernetes agent is not really an option). Thanks!

Hi all! Can someone rub me into the documentation where I can find information on how prefect works with long-run jobs. We are interested in what will happen if there will be temporary network issue. What will happen with the job and do we have any choices to configure behaviour for that.

Has anyone had trouble with a localized

CronClock

schedule skipping

Flow

runs, entirely?

Does prefect support intersphinx for documentation purposes?

Has anybody used Arrow for anything with Prefect/Dask? https://arrow.apache.org/docs/python/index.html It seems appealing, I’m just wondering if there are sharp edges.

Archiving this myself: SSL Errors when using Prefect agents can be fixed with updating the package “certifi”, running:

conda update certifi

Follows the overflow post  where I found a similar problem with this solution: https://stackoverflow.com/questions/41691327/ssl-sslerror-ssl-certificate-verify-failed-certificate-verify-failed-ssl-c

I had a quick question about the

.map

within prefect. Is it guaranteed that the order is preserved across multiple maps, like a python

multiprocessing.pool.map

? Or is it not a guarantee that the order is preserved, like

multiprocessing.pool.imap_unordered

. I couldn't find whether this was guaranteed or not in the

prefect docs

.

folks, I'm coming from a luigi background, where we can add custom loggers on top of luigi loggers:

[core]
logging_conf_file = logging.cfg

then

[loggers]
keys=root,app

[handlers]
keys=consoleHandler,StackDriveHandler
...

Is there anything similar in Prefect? I do not want to add additional logs inside my tasks, but make prefect log using additional loggers.

Hey I am deploying a Kubernetes Agent using Cloud backend. I am confused on how the agent knows which flows belong to it. Is there a value in the metadata that tells cloud server that this flow belongs with this agent?

helo! Beside the logger is write in normal mode, I wanna write logging into table in Database. How to do that?

Hello, trying to run a workflow using ecs agent - getting this error while starting ecs start ValueError: Failed to infer default networkConfiguration, please explicitly configure using

--run-task-kwargs

I’ve seen some confusing examples, possibly representing different versions of Prefect over time. Is there a canonical example .prefect/config.toml? For example I saw

host = foo

in one video, and

[server] endpoint = baz

in another