is there a way to troubleshoot a flow that is stuck in 'pending' and won't run?

HI Guys, I need help with a

dbtShellTask

. The issue I have is that I have a handler which sends notifications with details on the failing flows but when dbt task fails with return code 1, all i get is

'Command failed with exit code 1'

is there a way to capture in a variable the logs of a failed shell task to then process them and display them using the handler?

Hi all, Im learning the basics of Prefect and have a quick question. When I write this and register my flow to the cloud:

with Flow("flow-a") as flow_a:
        a()
        b()
    flow = Flow("flow-a-b", tasks=[a, b])
    flow.register(project_name="tester")

Is the code written in function

a

and

b

saved on the cloud, or does the cloud refers to my system every time it needs to run the flow on an agent? If it does refer to my system, will it make a difference if the agent is running on a different system?

hello i'm experiencing a weird log behaviour when using a MySqlFetch task in a local env (tried with and without CLI). I have a task that does an SQL query to retrieve a parameter value. As soon as this task is executed, the logs are kind of duplicated. I tried with the call of MySqlFetch and without inside the task and the logs are not duplicated when i don't use it. I add the sample logs below

Is there any example code available for calling flows via some python API (flask / fastapi).

FIY -the log viewer in 2.0 struggles to keep up when testing locally with anything other then the

SequentialTaskRunner

. Showing the most recent logs first would help usability.

Hi all. I'm currently trying to refactor some prefect 1 code into prefect 2, and I'm trying to make use of some states in flows and tasks I'm currently developing, as noted here : https://orion-docs.prefect.io/concepts/states/#state-details However, I'm not entirely sure how this is meant to be implmented in, lets say a @flow function object to get information on that particular flow/task. Is there perhaps an example I can work with to expand on this on my side?

I wanted to double check that it's not possible to modify a task name during the task run, right? I have the following tasks, which take an identical DataPayload TypedDict, but am separating them out to allow for human-friendly task names about the load operation:

with case(save_snowflake, True):  # type: ignore
        load_snowflake(
            task_args=dict(name="Load Owners into Snowflake"),
            data_payload=owners_df,
        )
        load_snowflake(
            task_args=dict(name="Load Properties into Snowflake"),
            data_payload=properties_df,
        )
        load_snowflake(
            task_args=dict(name="Load Amenities into Snowflake"),
            data_payload=amenities_df,
        )

        load_snowflake.map(
            data_payload=tuple(reservations_df),
            task_args=dict(name="Load Reservations into Snowflake"),
        )

hey, i'm trying to insert data in mysql using the

MySqlExecute

task. I have no error (i had some but i fixed them) but the row i try to insert isn't showing in the table. I can do it with the same user from mysql workbench and the

MySqlExecute

task return 1 as expected. I don't understand why, maybe i'm doing something wrong. Here is a code example of my task:

@task
def set_new_date():
    logger = prefect.context.get("logger")
    new_date = datetime.date.today().strftime("%Y-%m-%d")
    query = f"""
        INSERT INTO `name`.`table_name` (`name`, `value`) VALUES ('name', '{new_date}');
    """
    result = MySQLExecute(
        db_name="name",
        user="user",
        password="pwd",
        host="host",
        port=3306,
        query=query).run()
    <http://logger.info|logger.info>("New date is {}".format(new_date))
    <http://logger.info|logger.info>("Query results is {}".format(result))

Hey folks, we've been running Prefect cloud now for a while and find it really good. I'm now considering looking in to self hosting the Server and UI on a K8 cluster we have as opposed to paying for cloud. Im a little confused as to what you get for the ongoing monthly cost when you could host for free. I'd be keen to get others thoughts on the community here as to their experiences. I am aware that we would obviously not have to pay for hosting or updates etc, but given the pricing model this feels like a negligible cost in relation to the monthly fees for cloud. I feel like I could be missing something that could be obvious to the community.

Hi folks. We’re trying to get a better handle on the module and version compatibility between Flow registration -> Prefect Agent -> Flow Runner -> Dask Cluster so that we can ensure consistency without having to frequently update versions our Agent is running. We’re trying to understand where the flow deserialization happens during a flow run on a K8s cluster. If my understanding / recall is correct 1. Agent receives the Flow and launches the Flow run pod. 2. The Flow run pod deserializes the flow from storage and launches the Dask KubeCluster. 3. The Dask scheduler pod launches worker pods based on scaling settings. • If my understanding is correct, our Agent container needs only to have Prefect and Python versions which match the environment where we registered our Flow? As an example would an Agent running Python 3.8 and Prefect 0.14.22 work with a Flow registered using Python 3.8 and Prefect 0.15.11? • Does the Flow run pod need to have matching dependencies for all the modules used in the environment where the flow was registered. If so, how is the image used for the Flow run pod configured? Are there dependency mismatch issues we should be aware of for Flow deserialization? • The Dask scheduler and worker pods definitely need matching dependency versions from the registration environment to avoid any code inconsistencies. Some of this information seems to be spread around the docs and discourse so I was hoping we could get some definitive guidance about this. Cheers

Hi all, I'm wondering if someone would be able to clarify how flow registration works in the case of

LocalStorage

and

KubernetesRun

, as shown in this example. We're trying to follow the same pattern in that example, and our setup registers flows in CI via the Prefect CLI when a PR is merged into the main branch. When doing this, the flow is actually run while building in our CI environment (in this case, I just have a print statement in the flow to verify that it's running). Is there a recommended pattern to register flows in CI without running them?

Hello! 2.0 question: I am trying to build a flow that uses other flows that have already been registered in Prefect. Is there a way to define the new flow referencing the existing flows? All the examples I have seen declare the main flow and subflows at the same time and in the same file. Thanks!

I have a task that outputs a tuple of data_frames, and other tasks that output individual data_frames. I'm trying to think of a good way to reduce them into one tuple of data_frames so I can use a single map task over them. So far, I've tried extending a list and coercing back to a tuple, and a few other methods that have failed. There's likely a parsimonious way to do this that I'm missing

Hi, I wrote this code:

@task
def slicer():
    time.sleep(2)
    op = []
    for each in range(0, random.randint(1, 10)):
        op.append(each)
    return op

@task
def model_slide():
    logger = prefect.context.get("logger")
    time.sleep(2)
    <http://logger.info|logger.info>(f'model c: {datetime.now().strftime("%H:%M:%S")}')


if __name__ == "__main__":
    with Flow("processing-flow", executor=LocalDaskExecutor()) as flow:
        list_of_slices = slicer()
        for each in list_of_slices:
            create_flow_run(flow_name="other-processing-flow", project_name="poc")
        model_slide(upstream_task=[create_flow_run])

I'm getting the following error:

for each in list_of_slices:
TypeError: Task is not iterable. If your task returns multiple results, pass `nout` to the task decorator/constructor, or provide a `Tuple` return-type annotation to your task.

What am I doing wrong?

hi folks, I'm curious to know how I would go about triggering a flow via prefect cloud from my local machine. I have the following use-cases: 1. run a prefect job in our CICD process: Here I'm envisioning having some sort of API Key present in our CICD, and perhaps making a post request to prefect.io to trigger the flow I'm interested in 2. trigger failed jobs manually, but rather than logging into prefect cloud and hitting the rocket ship icon, I'd prefer to not leave my terminal. maybe there is a client for prefect-cloud already, so we wouldn't have to post requests using curl, but I'm comfortable curl-ing my way through this. Thanks in advance folks!

I am trying to access agent information using prefect client but it is throwing error. Am i doing it correctly? I have tried using both

access_token

and key created via service accounts. Both of them did not work.

Whenever I register a flow that has a cron schedule it seems to immediately run all of the tasks upon registration. Is this expected behavior since my assumption was it should wait until the cron schedule condition is met before running.

Hello everyone. I have a setup with aws fargate that was working great until 2 days ago. For some reason since yesterday we are getting a weird error message and we haven't changed anything. In prefect the error says "A Lazarus process attempted to reschedule this run 3 times without success. Marking as failed" In the agent we can see the following error: "Error: no such option: --token (Possible options --env --key) Last night, process on prefect tried to run 3 times and failed. He gets stuck on "submitted" So, today, I changed the schedule and after stuck for 19 minutes at "submitted" it was able to run. I have no idea what's going on, does someone understands what's going on?

Hi everyone, I’m interested to hear how you’ve handled scaling up Prefect and dealing with a large code-base. We have around ~40 data pipelines to manage, around ~15 using Prefect. We’re building new pipelines all the time and old pipelines are being migrated too. Currently these pipelines run the ELT process end to end and we have helper libraries that are reused across pipelines. We’re considering how best to architect these pipelines to encourage code-reuse and effectively maintain them. We’ll have 3 teams maintaining them so code ownership is a consideration too. One idea is to split up the pipelines so they don’t run the full process end to end. We could use an event driven architecture, so smaller flows are triggered to run based on an external event handler. It gives us more choices for team ownership and could make it a little easier to replace or add steps in the ELT process. The alternative is to keep doing what we’re doing and make the most of shared libraries to encourage code reuse. In either case, we’ll do more to use configuration so similar datasets are processed using the same data pipeline. I’d be keen to hear your approach to handling large code-bases.

Hello everyone! I am trying to add colour using html tags in the artifacts but it does not work. In the documentation it says it supports Github-flavored Markdown, so inline html should be rendered. Am i doing something wrong or we cannot add html formatting in artifact yet? Thanks!

Hi Prefect community, I also have problem with redeploy, please help. I'm using Prefect 2.0b3, and I want to make changes to my deployment. Maybe to change the code, maybe to remove schedule... But when I run

prefect deployment create <deployment_name>

for an existed deployment, I get error

prefect.exceptions.ObjectAlreadyExists

and no changes are made. Please help me on how exactly do I make changes to my deployments. Thank you.

Hey, just saw the announcement for

2.0b4

, some great stuff in there. One thing that caught my eye was:

Futures from async tasks in sync flows are now marked as synchronous

The docs section for async tasks just says ‘Coming soon’. Is this a feature that’s already available and the docs are out of date or am I misunderstanding what is meant here?

Hey peeps! I’m currently trying it out some tests here to be ran with prefect cloud. When i’m using the flow.run() in my machine just using a LocalRun and Local storage, everything seems to run fine, but when i try to register the flow and run through the UI i get back the ModuleNotFoundError(“No module name ‘/Users/username’“). I’ve already searched for it’s solution back in the discourse page,(https://discourse.prefect.io/t/when-i-run-my-flow-i-see-an-error-failed-to-load-and-exe[…]derror-no-module-named-users-username-what-is-happening/33) but none of the solutions seems to work, even when i specify the module within the agent. The module that i’m using is in the same folder structure and everything seems to work when running directly from my IDE. Can’t figure it out what’s happening 🤔

We'd like to integrate Sentry with our Prefect flows as described here. However, the approach that has been taken here has been to have the sentry initialisation done in a separate task that is called at the beginning of each flow. Am i correct in believing that this will only work for LocalExecutor and not for LocalDaskExecutor or DaskExecutor? If so, what would be the best practice to have the sentry initialisation done for those executors? Calling the initialisation as a function at the beginning of each task is technically feasible but clunky. Would a Task subclass help here? Could the initialisation be put in a task state handler?

Seeing this error: can't pickle CompiledFFI objects , can someone help what might be the issue?

can GraphQl be used to invite users to your tenant?

Hi all, totally random question but are there any books written about Prefect? I know some popular (like pytest) python libraries have books written on them. Not sure if Prefect had one or not.

Morning! When I raise a

RETRY

, is it possible to queue an upstream task that had already passed? For example:

drop_glue_database = Parameter()
case(drop_glue_database, True):
    drop_stuff()

save_to_parquet()

If save to parquet fails, I'd like the task to drop and rebuild the database, which is usually unnecessary and parameterized.

Hello, have been seeing this problem intermittently, any ideas? Prefect endpoint returning 500?

prefect.exceptions.ClientError: [{'path': ['secret_value'], 'message': 'No value found for the requested key ("xxx") in tenant xxx', 'extensions': {'code': 'INTERNAL_SERVER_ERROR'}}]