Hi folks, this is a question of passing a custom context to a prefect flow run. More specifically about passing a custom context with a nested dictionary.

I’m experiencing an issue with unpickling when a flow is registered from inside a conda environment. Registering the flow directly from a

virtualenv

which contains the necessary dependencies works and the flow runs as expected. But registering the flow from a conda environment I receive this unpickling error when running.

I have a quick question, I was thinking about using preemptible machines. If I was using checkpointing and caching the results I guess it would just retry the flow?

Is there a way to get the agent to only run flows from a particular project?

Hi Prefect, Running a flow from another flow can be done through

StartFlowRun

. However, this needs a flow being registered prior to calling

StartFlowRun.run()

, which requires Prefect Cloud or server to register the flow. Is there an alternative to use dependent flows when no prefect Cloud or Server is available (for example when running a unit-test testing the full flow and its dependent flows)?

Hi - I ran into a problem recently with the Task Concurrency Limiting feature. I have a task tag that is supposed to throttle the number of tasks that can interact with a particular database. In the past, this tag was only applied to a task that is mapped in a single flow. Yesterday, I tried applying this same tag to a separate mapped task in a different flow that interacts with this same database. At first, it worked as I expected. However then I saw that any task with this tag applied to it would just sit in state of "Queued" and never move into "Running". I made sure that no other tasks were running with tasks using this tag, but the behavior remained the same. I ended up having to delete the tag to get the tasks to run. My questions are: 1. is it appropriate to have a single task tag shared between different flow's tasks that have a common resource dependency limitation like the one I've described? 2. Any idea why all these tasks were getting stuck in a "Queued" state when there were no actively running tasks with that tag?

Hi, I am looking for something to either run to look at long running flows and send alerts based upon these flow running over let's say an hour for example. Does anyone have a good reference to something similar or have any ideas for these types of issues I am having? Thanks in advance!

Does anyone have an example of how to use KubernetesSecret? We are using Bitbucket for storage, and the password is stored in Kubernetes as “bb-flows-pull”. Everything works if we hardcode the secret into the script but this isn’t an option for us. I’m a complete beginner on this so apologies if this is something simple.

Hi everyone — I’m doing a PoC of using Prefect at my company and I’m running an ECS Agent as an ECS Service. I’m wondering how much memory does the ECS agent needs, in terms of vCPU and GB. I assume this can be quite minimal..

Is there a migration guide somewhere for moving from Airflow to Prefect?

Second question: How's the static typing/MyPy support in Prefect?

Is there a way to

prefect execute flow-run

an already-completed run? My use case is that I’m building a custom container to execute flows stored in script storage on my kubernetes agent, but when the flow runs it can’t find python dependencies (

import pandas

fails). I’d like to repro it locally, but when I try and run the flow I see

Flow run has already finished.

. Other advice for debugging would be welcome; I’ve run a shell in the container and I do see my dependencies FWIW.

howdy id like to have an action in my tasks/flows happen upon cancellation from the prefect ui - is this doable?

Hey all! I'm really struggling to register a very large flow (~1200 tasks) to cloud- I'm hitting an

API_ERROR

on flow.register(). I had this issue intermittently when I last worked on this project (late March), related to this post, but it used to register after 1 or 2 retries. However now I can't register at all - I've tried 6 times and still can't register the flow. I'd be very grateful for any advice please! Prefect version 0.14.14, docker storage pushing to ECR, full error message in thread!

Issue with new version of register. Previously (and currently although I'm getting warnings) I was able to register flows with this:

prefect register flow --file flow.py --project proj_name

This works without issue, however the recommendation is to use the new format:

prefect register --project my-project -p myflows/

Is there a way to pass an existing dask client to Prefect (instead of pass a DaskExecutor which launches a cluster on its own) - asking in particular to be able to run coiled cluster commands to launch cluster, register some worker plugins, trigger a scale, then pass it to Prefect for execution.

Greetings everyone , I am facing issue while connecting my gitlab storage for flow registration. I did try the steps mentioned in docs but i guess I am missing on something. Can anyone please share in simple steps of resolving the same.

Hello Prefectians 🙂, just starting out on Prefect, at our Organisation we are trying to use Prefect with Elasticsearch just wanted to know if anyone had any experience to guide me with this

do y'all know how to propogate canceled states from the flow to task? it seems lik the task just keeps running when i cancel a flow... retrying the flow also doesnt work until i explicitly set the state of the active task to "cancelled" (even after having cancelled the flow)

Hello everyone, I’m using an ECS Agent and S3 storage and I’d like to know how I can specify the docker image in the Run Config (ECSRun) at runtime. I tried to get the image to use from an environment variable but it does not work. I need that because I’d like to use different docker images in staging and production environments. Any suggestion?

Is there a way to get the logs from a single element of a mapped task?

Hey folks, we’re experiencing some intermittent issue with Fargate Agent in EC2 mode. we are on

prefect==0.13.13

From time to time, tasks fail to start and we get the following error:

f6edaa77-6dc6-4b85-953a-25ccdb4fb366 finished in state <Failed: "list index out of range">

where

f6edaa77-6dc6-4b85-953a-25ccdb4fb366

is the task run id. We can’t find any meaningful log on ECS, but I suspect this error is caused by the fact that the container fails to start. Any suggestions? Thanks!

Hey everybody. My name is Zack Novak, and I’m a data engineer at Second Front Systems. We are building a platform as a service to support bringing commercial technology into the Department of Defense and other government agencies. We want to use Prefect, but can’t as it is not in the Department of Defense’s hardened container repository called Ironbank. Has Prefect gone through this Ironbank onboarding previously / would this be something Prefect is interested in? Currently the only two DoD approved workflow solutions are Airflow and Argo, and Prefect is the preferred solution over those two.

Hey guys, curious if someone can help me understand task mapping / flattening for an iterated for loop. This loop calculates tracking error for each account in several strategies, and as such takes quite some time. I know task mapping would be of use here, but can't quite seem to get it right from the docs.

strategies = ['strat1', 'strat2', 'strat3']

for strat in strategies:
    strat = [strat]
    
    # this gets the list of applicable accounts
    lstAccnts = GetRiskModelAccnts(strategy = strat)

    # this gets the tracking error risk on an account level (each account has between 1 and 3 strategies)
    dfAccntsTEc, dfAccntsTE, dfAccntsMoments = RiskWrapper(
                                                           lstAccnt = lstAccnts,
                                                           benchmark = 'SPY',
                                                           lstAllowStrats = strat,
                                                           AggregationLevel = ['accnt']
                                               )

    # this gets the tracking error risk on an account-strategy level 
    dfAccntsTEc, dfAccntsTE, dfAccntsMoments = RiskWrapper(
                                                           lstAccnt = lstAccnts,
                                                           benchmark = 'SPY',
                                                           lstAllowStrats = strat,
                                                           AggregationLevel = ['accnt', 'strategy']
                                               )

You'll notice the aggregation level is also something that is iterated, and there is a third aggregation level ['accnt', 'strategy', 'strategyLevel'] that isn't shown above. I know mapping could be of use here as well but I'm not sure the correct syntax to implement it.

Are there any prohibitions/caveats associated with a non-task function in a flow, particularly with respect to how the framework might serialize the function and flow? As an example:

def my_non_task_util_func(some_args):
    return util_results

@task
def task_a(a_args):
    return a_result

@task
def task_b(b_args):
    return b_result

with Flow('test') as f:
    util = my_non_task_util_func(some_args)
    util_2 = my_non_task_util_func(some_other_args)
    a = task_a(util, other_a_args)
    b = task_b(util, util_2, a)

Should I be concerned about how Prefect treats this? For further context, this util func is a factory for a collection of

related prefect.tasks.core.constants.Constant

s. Thus far, it has been easier to return these from a vanilla func rather than a task that returns a collection of tasks.

Hey, I was wondering what kind of solutions people have found to event-driven flow execution. I saw a there was a PIN for the issue but I wanted to know how people have been solving this issue without built in functionality (or if there is some functionality ive overlooked)

Hi - I have a flow that runs under a minute and is part of another flow. I was thinking of wrapping the flow in an aws lambda to speed up the overall processing of the larger flow. Any thoughts?

Hello, I am hoping for some guidance here. After a mapped task runs in parallel I need to run a different task sequentially a dynamic number of times with Parameters. Is this possible? The below is in the flow context ... with Flow

post_runner.set_upstream(all_pushes_mapped_results)
    post_runner.bind(mapped_run_name,
        mapped_gpudb_user,
        mapped_gpudb_pass,
        mapped_gpudb_host,
        mapped_collection_name)

post_runner is a task itself so that I can actually get access to the parameters. Internally the runner task loops and creates a dynamic number tasks (if I do the same algorithm below directly in the flow context I dont have access to the parameters)

class RenameTaskRunner(Task):
    def __init__(self, *args, **kwargs):

        super().__init__(*args, **kwargs)

        # other constructor logic

    def run(self, run_names, gpudb_user, gpudb_pass, gpudb_host, collections_list):

        <http://logger.info|logger.info>("Start Rename Runner")

        all_tasks = []
        iterations = len(run_names)

        all_tasks.append(RenameTask(name="push1_post"))
        all_tasks[0].bind(gpudb_user, gpudb_pass, gpudb_host[0], collections_list)

        for i in range(1, iterations):
            all_tasks.append(RenameTask(name=f"push{i + 1}_post"))
            all_tasks[i].set_upstream(all_tasks[i - 1])
            all_tasks[i].bind(gpudb_user, gpudb_pass, gpudb_host[i], collections_list)

        <http://logger.info|logger.info>("Finish Rename Runner")

        return all_tasks

Below is the error. Is it possible to have RenameTaskRunner register the task with the given outer flow context? Can I pass a reference in to the constructor or some other idea?

[2021-05-25 01:31:50+0000] ERROR - prefect.TaskRunner | Unexpected error: ValueError("Could not infer an active Flow context while creating edge to <Task: push1_post>. This often means you called a task outside a `with Flow(...)` block. If you're trying to run this task outside of a Flow context, you need to call `RenameTask(...).run(...)`")
Traceback (most recent call last):

Hello everyone. Prefect looks wonderful, but I am unable to get past the “https://cloud.prefect.io/welcome/name-team” screen on first login. Support appears to be offline, so I was wondering if anyone in the community had an idea for bypassing this screen?

Hi everyone. We are a small data company based in Europe. We use prefect since months without going on cloud. We built all our automation flows on prefect and it's an amazing tool, thanks a lot for your work. We decided to move some flows on the prefect cloud backend and we were surprised to see the total duration time increase by a factor of two compared to our local prefect backend server. The flow on which we ran the tests is one of the longest in our stack (191 tasks, 26 parameters). Keeping the same agent, executor, computer, and parameters. We move from 65s to 135s. Each task takes on average +100% time increase (even the parameters collection). So, we understand that there is some time cost linked to the exchange of state information and metadata but could you, dear fellow community, please help us with the following questions? 1. Is the time gap between runs on the prefect cloud backend and the prefect server backend proportional: The longer the flow, the longer the time gap? 2. Are there any good practices to reduce this time gap? 3. Upgrading to the standard or enterprise offer will that make things faster? Thanks a lot :) Extra information: Our internet connection at the office 127 mbps/s download, 7.8 mbps/s upload.