Hello all 🙂 Is it possible to have dynamic cache keys? Currently they can be templated, so they are semi-dynamic, but I haven't found any way to make data that's passed through the pipeline part of the key. Usecase: caching mapped tasks where the order can be non-deterministic, so

prefect.context.map_index

might not be enough

Hi again, I want to run my first flow linked to my local API and I am trying to run

client.create_project()

to create a project. But each time I get the following error :

requests.exceptions.HTTPError: 400 Client Error: Bad Request for url: <http://xxx.xxx.xxx.xxx:4200/graphq>

My code :

client = Client(api_server="<http://xxx.xxx.xxx.xxx:4200/graphql>")
client.create_project(project_name="weather")

I don't understand what port / URL I should put in

Client()

. Thank you for your help.

Hi, I am trying to make a flow run on my API but I get the following error from the UI :

Failed to load and execute Flow's environment: ModuleNotFoundError("No module named '/root/'")

I don't understand the problem as I don't have any relative import in my code and the latter is pretty simple : https://github.com/flavienbwk/prefect-docker-compose/blob/main/scripts/weather.py

Hi, is there a proper way to launch an on demand k8s dask cluster i.e. KubeCluster and then run prefect tasks on it using DaskExecutor? I tried doing it but getting a

ERROR - prefect.FlowRunner | Unexpected error: TypeError('code() takes at most 15 arguments (16 given)')

not sure if it’s a supported mode of operation.

hi Team, I’m migrating Prefect core from 0.12.6 to 0.13.10, and get the error when i try to register my flow. Please help, thanks.

File ".../.pyenv/versions/alert/lib/python3.8/site-packages/prefect/core/flow.py", line 1608, in register
    registered_flow = client.register(
  File ".../.pyenv/versions/alert/lib/python3.8/site-packages/prefect/client/client.py", line 734, in register
    serialized_flow = flow.serialize(build=build)  # type: Any
  File ".../.pyenv/versions/alert/lib/python3.8/site-packages/prefect/core/flow.py", line 1451, in serialize
    self.storage.add_flow(self)
  File ".../.pyenv/versions/alert/lib/python3.8/site-packages/prefect/environments/storage/local.py", line 140, in add_flow
    flow_location = flow.save(flow_location)
  File ".../.pyenv/versions/alert/lib/python3.8/site-packages/prefect/core/flow.py", line 1520, in save
    cloudpickle.dump(self, f)
  File ".../.pyenv/versions/alert/lib/python3.8/site-packages/cloudpickle/cloudpickle_fast.py", line 55, in dump
    CloudPickler(
  File ".../.pyenv/versions/alert/lib/python3.8/site-packages/cloudpickle/cloudpickle_fast.py", line 563, in dump
    return Pickler.dump(self, obj)
TypeError: 'NoneType' object is not callable

Hi, I am trying the S3 storage option. I have two questions about it : • Does the S3 storage option uses Docker (does it saves images or just flows data ?) • How can I specify the endpoint of my own S3 server (I am using Minio) and specify its credentials ? (Am I forced to use Webhook storage with non-AWS S3 servers ?) Thank you !

Is there a way to get an IntervalSchedule to pass (a Parameter?) the interval to the resulting flow handler so it knows what time span it is responsible for?

the docs show how to pass a fixed label of the Schedule itself (e.g., "Daily schedule"), but not the span ("10/1/2020--10/2/2020")

I think the ideal is Parameter so we can reuse for other flow handlers, but right now looking for anything 🙂

Hi Team, i see GraphQL API change between 0.12.6 and 0.13.10, such as there is schedule instead of schedules in flow attributes. The changes break some functions in my system, so it’s better to have a brief about the change for me to fix them. where can I see some introduction about all the changes between the two versions? Thansk

Hi folks! Are there any training courses or workshops to learn and master Prefect?

Hi, I have a question about this error :

ValueError: Interval can not be less than one minute when deploying to Prefect Cloud.

Why was that limit set ? Why can't I decide to schedule a job every second ?

Hi, about Prefect flows and agents : does 1 flow == 1 agent, or are tasks in flows parallelized over the available agents ? Should I use Dask for that ?

hi Team, with the latest core 0.13.10, seems I have to get tenant and project ready before creating any flow. Is there any workaround to get tenant and project ready easily instead of creating them manually?

Hi there! I am trying to get started, and I'm wondering if there are any good architecture diagrams to get oriented within the Prefect ecosystem?

Looking around in the GitHub issues but I don't see anything for it. Has there been thought to adding something like:

results, errors = my_task.safe_map(**iterables)

is there a way to parameterise the timeout of a flow somehow?

I have a task to grab some data, but depending on the size i want to specify on the command line how long the timeout should be

I tried to set the timeout before I run the task, for example flow.slow_task.timeout = 100 for @task(timeout=9) def slow_task(opts, scripts): sleep(10) return 1

Does anyone know is this will be implemented in Prefect anytime soon? https://docs.prefect.io/core/PINs/PIN-14-Listener-Flows-2.html How do people currently handle event-based flows I wonder? 🤔

I think I need to do the class slow_task(Task): run(self, opts, scripts): thing, but the example in the documentation hints at this but not enough detail for me to work out how to do it.

Hi Guys, Is there a way to supply MySQL connection to multiple tasks? Currently, I am getting this error : "TypeError: can't pickle _mysql_connector.MySQL objects"

Does anyone know if it’s possible to either disable or increase the time interval for the Lazarus process?

Hi 👋 Is mapping only possible on the functional API and if not, how can something be mapped with Imperative API?

Hi, I am working on FargateTask sample provided in the documentation "https://docs.prefect.io/orchestration/execution/fargate_task_environment.html”. I am getting following error:

[2020-10-13 13:56:50,113] ERROR - agent | Error while deploying flow: ClientException("An error occurred (ClientException) when calling the RegisterTaskDefinition operation: Invalid setting for container 'flow'. At least one of 'memory' or 'memoryReservation' must be specified.")

. I have followed the sample given the documentation and have defined all the attributes including "memory" and "cpu". Please suggest on the same.

Hi guys! We have successfully deployed a Kubernetes Agent in Amazon EKS following your great tutorial on Medium. I executed some stress tests by starting around 30 flow runs for the new k8s agent, and I noticed that it picked 6 runs at a time (I don't know if 6 is a setting, or the agent picks this number based on the node capacity), and queued the rest. It's not bad! but since our flow runs can take as long as 10 hours each, now I'd like to setup an autoscaler to avoid having a long queue of runs waiting for so long. As you might already inferred, I'm new to Kubernetes. My guess is that HPA is not the right answer here. Maybe a cluster-autoscaler is the way. Could you guys give me a hint on where should I start for setting up an autoscaler on EKS that works with the Kubernetes Agent? Thanks!

Hi 👋 🙂 Does anyone have experience with Dask Executor? I guess to the nature of a flow the DFE can be blocked sometimes… 🤔 I have detailed my question here: https://github.com/PrefectHQ/prefect/discussions/3491

Hey friends, I'm looking for some guidance on versioning flows. I like the way flows are up-versioned when they get registered to Prefect Cloud and older versions are automatically archived. However, I want to be able to trace a flow version to its code in github and I'm not sure the best way to do that. We tag our commits with version strings, so I just need to be able to add that information to the flow somehow. Options I've thought of: • Can I add some kind of tag when registering my flow? I know there are labels, but as far as I understand the labels are just for linking flows with agents. I haven't seen anything about being able to add other types of tags. • Can I override the auto-incremented version number so it matches our version string? Again I haven't seen how to do this in the docs. • Can/should I add our version string to the flow name? It sounds like I can set a common version group id for each new version of my flow so that even though they're named differently, Prefect Cloud will still consider them to be versions of each other. However, the docs also say that version group ids are deprecated, so I'm not sure that's the right thing to use. How are other people handling this?

Hello, I am running into a strange failure at the

Cloudpickle serialization check

step of the health-check during flow.register(). Could you help? The error message is below

Beginning health checks...
System Version check: OK
Traceback (most recent call last):
  File "/opt/prefect/healthcheck.py", line 147, in <module>
    flows = cloudpickle_deserialization_check(flow_file_paths)
  File "/opt/prefect/healthcheck.py", line 40, in cloudpickle_deserialization_check
    flows.append(cloudpickle.load(f))
  File "/usr/local/lib/python3.8/site-packages/environs/__init__.py", line 334, in __getattr__
    return functools.partial(self.__custom_parsers__[name], self)
  File "/usr/local/lib/python3.8/site-packages/environs/__init__.py", line 334, in __getattr__
    return functools.partial(self.__custom_parsers__[name], self)
  File "/usr/local/lib/python3.8/site-packages/environs/__init__.py", line 334, in __getattr__
    return functools.partial(self.__custom_parsers__[name], self)
  [Previous line repeated 994 more times]
RecursionError: maximum recursion depth exceeded

Removing intermediate container 98613182a8f2
The command '/bin/sh -c python /opt/prefect/healthcheck.py '["/opt/prefect/flows/20-06-0-standard-ag-stage.prefect"]' '(3, 8)'' returned a non-zero code: 1
Traceback (most recent call last):
  File "build_and_register.py", line 174, in <module>
    flow.register(
  File "/usr/local/lib/python3.8/site-packages/prefect/core/flow.py", line 1588, in register
    registered_flow = client.register(
  File "/usr/local/lib/python3.8/site-packages/prefect/client/client.py", line 734, in register
    serialized_flow = flow.serialize(build=build)  # type: Any
  File "/usr/local/lib/python3.8/site-packages/prefect/core/flow.py", line 1439, in serialize
    storage = self.storage.build()  # type: Optional[Storage]
  File "/usr/local/lib/python3.8/site-packages/prefect/environments/storage/docker.py", line 346, in build
    self._build_image(push=push)
  File "/usr/local/lib/python3.8/site-packages/prefect/environments/storage/docker.py", line 412, in _build_image
    raise ValueError(
ValueError: Your docker image failed to build!  Your flow might have failed one of its deployment health checks - please ensure that all necessary files and dependencies have been included.

Please see more context and the details in the thread.

Hi folks, I am trying to schedule a flow to run every 4th business day - so I created my own filter function following a similar format as how prefect schedule filter functions are implemented but I am getting a marshmallow.exceptions.ValidationError