What have people come up with as a best practice for having tasks indicate that they failed? I could not find any example of a task that could fail in the docs or what failure would look like. Returning “none” is not a failure. Are only uncaught exceptions failures?

💃 Just arrived!

We re-deployed our ECS agents with the PREFECT__CLOUD__AUTH_TOKEN env var instead of (PREFECT__CLOUD__AGENT__AUTH_TOKEN), set to the new API key. The agents start up and show green on the agent status page in cloud. But the scheduled flows never start (eventually get resurrected but error out on the third try). What am I doing wrong?

from prefect import Flow, task

@task
def add_ten(x, y):
    return x + y

@task()
def log_result(x):
    print(x)


with Flow('simple map') as flow:
    mapped_result = add_ten.map([1, 2, 3], [10, 11, 12])
    log_result(mapped_result)
if __name__ == '__main__':
    from prefect.run_configs import LocalRun
    flow.run_config = LocalRun()
    flow.run()

I was expecting this code to return 9 results instead of the 3 actually returned. Is there a way to have map do the pairwise enumeration? I was assuming because neither argument was marked as unmapped then they would both be looped over.

Hello. I’m having some issues with mapped task parallel execution. I’m testing locally using the LocalDaskExecutor(). Can you confirm that separate mapped tasks without a dependency will execute serially? In the example below the mapped tasks within ‘a’ and b’ execute in parallel, but ‘a’ and ‘b’ execute serially.

with Flow("testing") as flow:
    a = poll.map(poll_interval=[5,10])
    b = poll.map(poll_interval=[4,9])

flow.run(executor=LocalDaskExecutor())

[2021-05-01 18:20:08-0500] INFO - prefect.FlowRunner | Beginning Flow run for 'testing'
[2021-05-01 18:20:08-0500] INFO - prefect.TaskRunner | Task 'PollDSCCaptureState': Starting task run...
[2021-05-01 18:20:08-0500] INFO - prefect.TaskRunner | Task 'PollDSCCaptureState': Finished task run for task with final state: 'Mapped'
[2021-05-01 18:20:08-0500] INFO - prefect.TaskRunner | Task 'PollDSCCaptureState': Starting task run...
[2021-05-01 18:20:08-0500] INFO - prefect.TaskRunner | Task 'PollDSCCaptureState': Finished task run for task with final state: 'Mapped'
[2021-05-01 18:20:09-0500] INFO - prefect.TaskRunner | Task 'PollDSCCaptureState[0]': Starting task run...
[2021-05-01 18:20:09-0500] INFO - prefect.TaskRunner | Task 'PollDSCCaptureState[1]': Starting task run...
[2021-05-01 18:20:14-0500] INFO - prefect.TaskRunner | Task 'PollDSCCaptureState[0]': Finished task run for task with final state: 'Success'
[2021-05-01 18:20:19-0500] INFO - prefect.TaskRunner | Task 'PollDSCCaptureState[1]': Finished task run for task with final state: 'Success'
[2021-05-01 18:20:19-0500] INFO - prefect.TaskRunner | Task 'PollDSCCaptureState[0]': Starting task run...
[2021-05-01 18:20:19-0500] INFO - prefect.TaskRunner | Task 'PollDSCCaptureState[1]': Starting task run...
[2021-05-01 18:20:23-0500] INFO - prefect.TaskRunner | Task 'PollDSCCaptureState[0]': Finished task run for task with final state: 'Success'
[2021-05-01 18:20:28-0500] INFO - prefect.TaskRunner | Task 'PollDSCCaptureState[1]': Finished task run for task with final state: 'Success'
[2021-05-01 18:20:28-0500] INFO - prefect.FlowRunner | Flow run SUCCESS: all reference tasks succeeded

Thx!

I’m having some trouble understanding whether I should be able to run

python myflow.py

and read Secrets within my flow. I’ve added the Secrets in the Cloud UI and authenticated with the prefect CLI. Is the right model that “running a flow locally without an agent never hits the server” or am I mistaken?

Hey Prefect community! 👋 Is there a way to control the order of mapped executions (I use local dask executor)? I noticed that once mapped (and scheduled) the actual execution is quite random… Whereas in my case, I need to be consecutive in the order I have provided it). Any help would be much appreciated! 🙂

Got an issue when I use shelltask. I have a long lasting task which creates tons of log. If I enable stream_out, it overwhelms my disk. If I don’t, prefect attempts to kill the task after several hours due to

heart beat check

.

Hi everyone, I'm new to prefect and I had a question for which I couldn't find an answer online: Can we run multiple separate flows at the same time, in parallel? I understand that we can parallelize tasks within a Flow. But I'm asking about running Flows in parallel. Thanks

Hi community 🙂 I'm new and I need some help for a very simple problem. I'm testing prefect since a few days, and I try to pass environment variables to a flow. It seems easy but it doesn't work for me. I have configured the flow like this flow.run_config = LocalRun(env={"GREETING": "Hello"}). And when I call flow.run, os.environ doesn't contain the GREETING env. Do you know why ? It is for local test and I don't want to register the flow right now. Thank you!

Hi Prefect, Is there a way to create a custom Task that can be visualized in the Prefect UI similar to Parameter tasks? I currently get most of my parameter variables from environment variables being set (and want to continue doing this). Ideally I want to group related parameters into groups (e.g. by using data classes) so the resulting groups can be passed around to wherever they are needed. For example, I currently fetch my parameters to communicate with my Kafka service via this custom task:

class GetKafkaConfig(prefect.Task):
    """
    Get Kafka configuration
    """
    def run(
                self,
                timeout: timedelta
            ) -> KafkaConfig:
        return KafkaConfig(
            broker_address=os.environ['KAFKA_BROKER_ADDRESS'],
            topic_name=os.environ['KAFKA_TOPIC_NAME'],
            timeout=timeout
        )

I would like to visualize these parameters in my prefect UI (and ideally make them editable), similar to how Parameter tasks can be visualized (e.g. as is demonstrated in ) Is it possible to write custom visualization layers for custom Tasks?

Hey @merlin , will respond in the server thread.

Hello guys, I have a question related to the task RAM usage - it is possible to save results using S3 storage or other storage without keeping all results in RAM? This is needed because I would like to process a large file, that is split into x subfiles, load each subfile in a different task, perform necessary operations, save it in the pickle file and then load that pickle file in the other task. The goal is to keep organised tasks in the prefect flow(I would like to track each subfile task operation in the prefect tasks) and keep as minimal RAM usage as possible by not storing all data in the RAM(need to store just 1 subfile at the time in the RAM). For now I'm trying to use S3 storage results(https://docs.prefect.io/orchestration/execution/storage_options.html#aws-s3), but it seems it is not free up RAM memory when result is saved into the pickle file. Any ideas related with this problem?

Hi all, I'm using Docker storage to store my flow within a Docker image on azure container registry. However when I change my flow and push the image to the registry (with the same tag as the previous version) it appears that when I run the flow with the Cloud UI, the flow is not run with the latest image containing the changes. Any ideas on why the agent is not using the latest image?

Hi everyone, When a task is failed, what happens to the downstream task? I was expecting that the downstream tasks would be failed. But it was just staying on the pending state. Is that expected from Prefect Flow?

Is it possible to name a flow run with the flow.run() function? We are currently using the client.create_flow_run() function but we need this call to be synchronous as to not overload our data warehouse. Appreciate any advice!

Hi everyone I'm having trouble registering my flow to cloud and am getting this docker error.

docker.errors.NotFound: 404 Client Error for <http+docker://localhost/v1.41/images/create?tag=0.14.0-python3.9&fromImage=prefecthq%2Fprefect>: Not Found ("manifest for prefecthq/prefect:0.14.0-python3.9 not found: manifest unknown: manifest unknown")

I have a mapped task and wanted to set the task name dynamically with

@task(task_run_name=name_fn)

where name_fn() dynamically generates the task name form the kwargs that are passed to it. This all works great when I'm running with the UI. When I run directly by calling flow.run() the

set_task_run_name()

in engine/task_runner.py is stubbed out and doesn't call my name_fn(). I can see that in TaskRunner.run() getting the call to

set_task_run_name()

isn't totally straight forward. What would it take to get

set_task_run_name()

to work when running directly without the UI?

https://docs.prefect.io/core/task_library/control_flow.html is raising a 404, linked from here: https://docs.prefect.io/core/examples/conditional.html#conditional-tasks

happy monday everyone!

Does anyone know what is going on here? In every task, I use logger = prefect.context.get('logger') and then call logger.info(), logger.debug(), etc.

D:\venv\poetry\.venv\lib\site-packages\prefect\utilities\logging.py:123: 
UserWarning: Failed to write logs with error: 
ClientError('400 Client Error: Bad Request for url: <https://api.prefect.io/graphql>\n\n

The following error messages were provided by the GraphQL server:\n\n    
INTERNAL_SERVER_ERROR: Variable "$input" got invalid value null at\n        
"input.logs[0].flow_run_id"; Expected non-nullable type UUID! not to be null.\n    

INTERNAL_SERVER_ERROR: Variable "$input" got invalid value null at\n        "input.logs[2].flow_run_id"; Expected non-nullable type UUID! not to be null.\n    

INTERNAL_SERVER_ERROR: Variable "$input" got invalid value null at\n        "input.logs[4].flow_run_id"; Expected non-nullable type UUID! not to be null.\n

are there any plans to add webhook as an action for automations?

I have a project with this structure. The two flows have some shared functions and configs.  project/ • flow1.py • flow2.py • utils/ • config/ I’d like to pack two flows in one docker image but I don’t see a prefect command which can help me do so. e.g if I speficy storage for both flow1 and flow2 with the same image name, they will overwrite each other when I call

prefect register

So I try to replicate the behavior of building storage by writing my own dockerfile. A question here is how does the file

healthcheck.py

and

flow1.flow

come into scope? Below is the output after I call register command with docker storage

Step 7/12 : COPY flow1.flow /opt/prefect/flows/flow1.prefect
 ---> ea7572a6c0e7
Step 8/12 : COPY healthcheck.py /opt/prefect/healthcheck.py

Is there any way to see in the cloud ui what executor the flow is registered with? I'm not seeing the task concurrency I am expecting and want to confirm the executor settings.

Does anyone know what is going on here? when getting api responses from google-ads, I keep getting this error

Unexpected error: ReferenceError('weakly-referenced object no longer exists')

on prefect Run logs api res count is almost 2000, error no occurs in local env someone please help me..!

Hi. I was looking for more information on tenants on the docs site, but only found brief mentions here and there. I found someone with a similar question. Are there any plans to expand on the docs? What exactly is the role of a tenant, do I still need to run

prefect server create-tenant

when using prefect server?

Hey team, somehow today our agent is not picking up any flow runs to run. The last run happened around 24 hours ago, and since then it's gone quiet, I restarted it and it just sits there reporting

Waiting for flow runs...

We are cloud users, and we don't think we touched anything at all since it was last healthy (it was a bank holiday here yesterday in fact)

Hi,,

Is it possible in Prefect to invoke R script with-in a task. I have a code in R script and that can only run in the R-studio, I don't want to convert it into python. How can I invoke that R code with prefect. Is there an interface between R and Prefect

Hey 🙂, I have a question regarding caching. There are some tasks in my flow for which I cache the results using GCSResult. If I set the 

cache_for

 parameter to 30 minutes, does this mean that every 30 minutes this task is rerun and cached again? My downstream task fails because this upstream task contains a token (for 1 hour), so it seems as if this this is not happening in my case.