hi, i got a flow success event before the time all task really successed. in the bellow pic, i receive a flow success at 1.09 pm , but still got a task success at 1.13 pm. is this a bug?

Hi what is the alternative for upstream_tasks in prefect 2.0

Hi! I have a flow that executes a

RunNamespacedJob

task. Because of a known bug the way I'm obtaining the logs are via

ReadNamespacedPodLogs

. All that working well. Recently I changed the pod spec to run the application as an

initContainer

then some docker image to capture results as the actual

container

. So I have an image that generates content and logs in the

ìnitContainer

and a

container

that process the produced content (upload to gcs basically). Problem: the logs from the initContainer are not being captured by prefect, only the logs from the container. Any recommendation? 🙏

Good morning 🙂 So I have been working on this really extensive ETL and finally uploaded it to the cloud but there it is failing... I have a parameter like this:

weeks = Parameter("weeks", required=False, default=None)

Afterwards there is a case block where either the last actual weeks is getting pulled or set to the provided Parameter. Locally it's working like a charm but when I want to execute the flow within the cloud, I get an error for the week task:

The following error messages were provided by the GraphQL server: INTERNAL_SERVER_ERROR: Variable "$input" got invalid value null at "input.states[0].task_run_id"; Expected non-nullable type UUID! not to be null. The GraphQL query was: mutation($input: set_task_run_states_input!) { set_task_run_states(input: $input) { states { status message id } } } The passed variables were: {"input": {"states": [{"state": {"context": {"tags": []}, "cached_inputs": {}, "message": "Starting task run.", "_result": {"__version__": "0.14.10", "type": "NoResultType"}, "__version__": "0.14.10", "type": "Running"}, "task_run_id": null, "version": null}]}}

Task 'weeks': Finished task run for task with final state: 'ClientFailed' Do you know what might be the problem here? Why is it stated that I got a null value? It should be None.

Hi, I'm trying to list files from a S3 bucket but don't know how to get the result of the task. Can anyone help here? I would have thought to get an array of file names instead of a task object.

Hi Prefect Community, I am currently receiving an error when using a local (on my pc) prefect agent which pulls from a work queue in the the prefect cloud. The error only occurs when I try to use the RayTaskRunner:

(begin_task_run pid=141324) ImportError: cannot import name 'SubprocessFlowRunner' from partially initialized module 'prefect.flow_runners' (most likely due to a circular import) (/home/christian/Documents/ray_and_prefect/env/lib/python3.9/site-packages/prefect/flow_runners/__init__.py)

I am using the following dependencies: prefect==2.0b7 prefect-ray==0.1.0 ray==1.13.0 Apparently I am doing something wrong with my dependencies or when I am importing them. Do you have any idea?

Hi, i'm trying to run flow locally from ide (vs code) and i got this: what have I done wrong ?

Hi, quick question, why

map

function fail? and works when

map

was replace with

bind

?

from prefect import task, Flow, unmapped


@task
def first():
    return range(10)


@task
def second():
    return True


@task
def third(numbers):
    print(numbers)


with Flow(name='test') as flow:
    numbers = first()
    second_result = second()

    # third.bind(numbers=numbers, upstream_tasks=[second_result])
    third.map(numbers=unmapped(numbers), upstream_tasks=[second_result])

flow.run()

Is there a docker-compose.yml file out there for running prefect server. We are currently on 0.15 and I want to upgrade to 1.2.3. When I first set up server there was just a single compose file, I can’t seem to find that anymore

Hello! Running into an issue I don't fully understand. I have a series of flows that have been running for a couple of months without incident. We're running all of our production on 1.2.0-python3.8 for the ECS Agent and for all ECSRun instances. Our flows use S3 storage. Nothing has been purposely or visibly changed with our Prefect installation- no new flow deployments, or changes to our AWS infrastructure. Our flows, which do little more than kick off AWS Batch jobs via the BatchSubmit task are now consistently failing with the following error on the BatchSubmit invocation:

Unexpected error: AttributeError("'S3Result' object has no attribute 'upload_options'")
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/site-packages/prefect/engine/runner.py", line 48, in inner
    new_state = method(self, state, *args, **kwargs)
  File "/usr/local/lib/python3.8/site-packages/prefect/engine/task_runner.py", line 930, in get_task_run_state
    result = self.result.write(value, **formatting_kwargs)
  File "/usr/local/lib/python3.8/site-packages/prefect/engine/results/s3_result.py", line 89, in write
    ExtraArgs=self.upload_options,
AttributeError: 'S3Result' object has no attribute 'upload_options'

Attached is one of the flows that are failing. Does the task running code record execution state back to the storage bucket? Thanks in advance!

Hi, in 2.0 I was wondering if it’s possible for the scheduler to prevent a single flow from “Running” concurrently? For example, if I use an IntervalSchedule with an interval of 5 minutes, and my first flow run takes 8 minutes, can the scheduler skip or wait until the first flow finishes before starting the new flow run?

Prefect 2 question here: Are an agent and work queue necessary in order to run a k8 job? For instance if i am using prefect cloud and create a deployment with kubernetes flow runner. Running the flow via the GUI or using "deployment run" from vm/local machine without an agent and queue setup results in no flow being run and a state of late. The prefect docs just say you need: Remote storage and

kubectl

configured to connect to a cluster

Hi all, how do you access the labels of the agent running a flow at runtime? I assumed there would be something in the context, but that doesn’t seem to be the case? Is there any agent level information you can access from inside a flow or task?

I have a running flow where a number of tasks are in a paused state. I’d like to programmatically approve all of them. What’s the best way to do that? I find that when I programmatically set all of their states to resume, I get varying behavior. Sometimes the tasks stay resumed and never get running. Any ideas why?

Hi, I'm going through v2 docs and was wondering if module storage from v1 (

prefect.storage.Module

) would be available in the future releases. We're running Prefect + Dask setup on k8s from a monorepo (i.e. agents and dask workers deployments run the same docker image & tag), where flows are sourced from Python modules (and registered before agents start up during a new release), therefore I'd prefer to not use any object storage for storing flows or results if possible

hey folks, I have a hopefully rather simple question also prefect noob. I’m trying to test the cacheing/restart features. For simplicity, I have 5 csv’s I’m downloading within 1 task. If the first csv is successful but then the task fails on the 2nd csv, I would like to restart the task where it left off and not need to re-download the 1st csv. From my understanding I need to leverage the results feature? I understand there’s a parameter for max retries but when retrying, is there any configuration needed to inform the task to start from the 2nd csv? Also if the max retries fails, I’d like the ability to re-start manually (presumably after fixing x bug that stopped the 2nd csv from downloading) again without having to re-download the 1st csv

So I've been using Prefect Cloud 2.0 on my personal account for a few weeks now, and we got our company account setup earlier this week. I can't figure out to login to beta.prefect.io using the company account. I can switch between mine and the companies with normal prefect.io. Does it need to be enabled and/or is it like that because its still in beta?

Is the new 2.0 release coming today?

Hello all, I'm having difficulties figuring out what this error means

Unexpected error: KeyError(0)
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/site-packages/prefect/engine/runner.py", line 48, in inner
    new_state = method(self, state, *args, **kwargs)
  File "/usr/local/lib/python3.8/site-packages/prefect/engine/flow_runner.py", line 569, in get_flow_run_state
    executors.prepare_upstream_states_for_mapping(
  File "/usr/local/lib/python3.8/site-packages/prefect/utilities/executors.py", line 682, in prepare_upstream_states_for_mapping
    value = upstream_state.result[i]
KeyError: 0

I’m trying to run the following code snippet

snowflake_query = SnowflakeQuery(
    max_retries=cfg.retry_max,
    retry_delay=datetime.timedelta(seconds=cfg.retry_delay_seconds),
)

sql_queries = cfg.sql_query.split(';')

with Flow(
    getenv('FLOW_NAME')
) as flow:
    run_snowflake_query = snowflake_query(
        account=PrefectSecret("SNOWFLAKE_ACCOUNT"),
        user=PrefectSecret(f"{environ['SNOWFLAKE_USER_VAR']}"),
        password=PrefectSecret(f"{environ['SNOWFLAKE_PASSWORD_VAR']}"),
        warehouse=cfg.snowflake_warehouse,
    ).map(query=sql_queries)

sql_queries is getting put into a list since I can loop through it….Any thoughts on what I am doing incorrectly?

We want to be able to turn any function into single-task flows. I’ve created a wrapper to do this (as pictured). Calling

.run()

on the flow that gets returned works fine, and registering the flow seems to be fine too (using k8s run). However, when we try to actually run it, we get the following error:

Failed to load and execute flow run: FlowStorageError('An error occurred while unpickling the flow:\n  ModuleNotFoundError("No module named \'build_index\'")\nThis may be due to a missing Python module in your current environment. Please ensure you have all required flow dependencies installed.')

Am I missing something obvious?

Hey Prefect community, I’m working with Orion and trying to create a deployment from the API client. I’m able to create it from the CLI, but for my use-case would prefer calling with the api client to ease of use / response parsing. It appears we haven’t updated this yet on the docs page: https://orion-docs.prefect.io/concepts/deployments/#with-the-api Playing around, I have something like this:

from typing import List
from uuid import uuid4

from prefect.client import get_client
from prefect.flow_runners import SubprocessFlowRunner
from prefect.flows import Flow
from prefect.orion.schemas.data import DataDocument
from prefect.orion.schemas.schedules import CronSchedule
from workflow_etl.flows.flow import hello_world

def main():
    prefect_client = get_client()

    schedules = [] # TODO

    for schedule in schedules:
        flow_id = await prefect_client.create_flow(hello_world)
        prefect_client.create_deployment(flow_id=flow_id,
                                         name=schedule.id,
                                         schedule=CronSchedule(cron="0 * * * * *"),
                                         parameters={
                                             **schedule.workflow_params
                                         },
                                         tags={
                                             'owner_id': schedule.owner_id,
                                         },
                                         flow_runner=SubprocessFlowRunner(),
                                         flow_data=# UNKNOWN WHAT TO PUT HERE
                                         )

I’m a bit stuck on what to put in the flow_data argument. Anyone tried this / have a link to sample?

Hello, a bit of an odd one, as I am not sure what I am doing wrong here. Effectively, a very simple task seems to be running forever when run via Prefect Cloud (using Local Agent, Prefect 1.0). The same flow when run on the machine that runs the local agent seems to work fine. Code:

from prefect import task, Flow, Parameter

@task
def task_function(in_list_size=10, out_list_size=4):

    in_list = list(range(in_list_size))
    print(f">>> Input list: {in_list}")
    
    out_list = list(range(out_list_size))
    print(f">>> Output list: {out_list}")
    
    diff_list = [i for i in in_list if i not in out_list]
    print(f">>> Result list: {diff_list}")
    
    return diff_list


with Flow("Test Flow") as flow:

    in_list_size = Parameter("input_list_size", default=10)
    out_list_size = Parameter("output_list_size", default=4)
    result = task_function(in_list_size=in_list_size, out_list_size=out_list_size)


flow.register(project_name="tutorial")

if __name__ == "__main__":
    flow.run()

Trying to figure out where the problem is - guessing something is wrong on the Local Agent side?

Hello the community. I’m stuck in my code since days. It’s really annoying because It was supposed to be the easiest flow I develop with prefect. So, the algo: I want to read the content of a big redis database (redis.scan_iter()), filter to keep key which not matching a pattern and then remove record (easy !). My problem, I don’t want to load all my redis in memory in the first step task (my dask workers will not handle) and start filtering after. I want to generate batchs of keys and process each batch directly (map feature so). If I run my code without .map(), it work but It’s sequential (I load all my key before filtering). When I run it with .map() the filtering task return directly a failed status without any print error. I have no idea why it fail ! Someone have a clue?

Hello 🙋 If we wanna use a

LocalDaskExecutor

for our flow, but to be able to limit the parallelisation (due to each task requiring a lot of resources when being run) to - for example - only two tasks at a time, is that possible? I read the docs but still don’t fully understand if we have to use a

DaskExecutor

for this, and if so - would it by default run locally (if we don’t give it any other config)? and kind of tangenial (and not directly related to prefect) but is there some advantage of something like the AWS fargate cluster for dask of a k8s dask cluster? is the former just easier to set up or something?

hello, our cloud colleagues want to deploy prefect 2.0/orion to kubernetes, then need this information: helm chart repository and url where i could find this information?

Hi! Prefect 2.0 has the ability to enforce pydantic checks for flow input which is a really nice feature. So, when we pass invalid parameters to a flow it goes straight from a pending state to a failed state as expected and we get an error in the form of:

09:47:51.189 | INFO    | prefect.engine - Flow run 'shiny-falcon' received invalid parameters and is marked as failed

However this information is quite limited. Is there a way to get more information which of the parameters failed to pass the pydantic check?

Hi All! I have been loving my experience with Prefect but now that I am deploying to Kubernetes (GKE) I am having the all too common problem of it running fine locally but not in the cloud. Initially I was running into

OOMError

but got around that by increasing the memory requested from k8s. Now I am stuck on having the ZombieKiller stop tasks that should be reporting back status since they continue to produce logs with the message

No heartbeat detected from the remote task; marking the run as failed.

I have attempted to set the

HEARTBEAT_MODE

to

thread

and

off

via the

config.toml

file as well as in the

KubernetesRun

environment variables but no matter the combination I setup I still run into heartbeat errors killing a process. I am curious if there is a way to send a heartbeat from within the code base or if there is another approach I should take to get around my time out issue. Cheers!

Hello everyone! I am planning to build an image processing pipeline which consists of both ML and non ML processing. I am planning to build each processing step as an independent function (Prefect Task) and successively apply these functions on each image. I will be pulling the images from a cloud storage container. The ML processing will require some GPU acceleration. My question here is twofold: 1. How do I load and share the ML models to be used in running the inference? I have some pytorch models right now. 2. Is it possible to specify on what type of node (GPU/non-GPU) each task will run on? I want that the ML inference functions to run on GPU nodes, and non-ML on CPU nodes. I would really appreciate your answers! Thanks!

Maybe someone will need this: So after some investigation my conclusion is that docker integration with wsl2 is the root cause. I got some successes after reinstalling wsl, but then it was failing again 😕 strange thing that it sometimes works and sometimes not. Even stranger, that when I try to run it in wsl2 docker, then it breaks wsl and I cannot even run prefect as "pure python" (without docker, just local python script) disabling wsl2 docker integration helps, but then I cannot then run docker in wsl 😞 So for now I'll stick to running it in windows docker ...