did someone try module as storage for prefect flows?

Just wondering if anyone has seen this in Prefect Server, we accidentally trigger the flow for the wrong date and when we try to cancel the flow from the UI it didn't work. The task continue to run under the Dask and we were not able to stop that task until we restart the Dask. Would appreciate if anyone know how to fix this without restarting the Dask?

Quick question, we usually achieve caching by using the checkpoint i.e. result are stored on the file. At times one of our task might fail and it might be possibly due to one of the upstream task had some issue and wanted to know if there is a way to remove the cache from the UI rather than going to backend and remove the file manually.

Do folks have preferences/advice for using

nohup

instead of Supervisor? I’m thinking something simple like this

nohup prefect agent local start 2>&1 > /tmp/prefect_local_agent.log &

Can prefect cloud support PickleSerializer on

PrefectResult

? I’m currently encountering this error:

TypeError: PrefectResult only supports JSONSerializer or DateTimeSerializer

Join us for the Prefect Happy Hour!

How do I tell Prefect to run exactly one copy of a Flow, starting immediately, until the Flow ends (variable amount of time), then immediately start another single copy, and so forth? ... by "copy" I just mean I wouldn't want the Flow runs to overlap

any tips on what to do when an agent gets "stuck?" I have one agent that is responsible for kicking off flows every minute, and it often ends up where the agent is querying for flows, and there are submittable runs, but they never start.

Hi all, given that during flow_run, task_A will return a list from a sql query downstream to task_B. task_B will then start at index[0] and send that value downstream to task_C. Now, x minutes later, flow_run will run again, and task_A will return the same list, but task_B will now return index[1] instead of index[0] and send the value of index[1] to task_C and so on and so forth. What would be the best approach for this, where I want a given flow, between scheduled runs, to remember its location in an iterable?

Hello there, to avoid wasting time and if anyone has tried it, is it possible to customize the prefect cloud UI or the beta feature (artifacts) to add things like progress bars or plots from tasks run on the UI...anyone ?

Hi, What is prefect orion and how is it different from prefect core and prefect server. How will prefect cloud get affected by this? Can we deploy jobs to prefect cloud from orion like we are able to do from core? Thanks.

Flow registration within the script using `flow.register()`: If I use

flow.register(project_name='my-project', indempotency_key=flow.serialized_hash())

on prefect server will the version number still get bumped each time?

When running prefect docker agents it seems that the a running docker container does not automatically inherit the context from the agent. Is that intended behavior? How best to set the context inside the docker container?

Prefect agent that is deployed on gke cluster when deploying server using helm chart and the prefect agent that we deploy on kuberntes using this commnad

prefect agent kubernetes install -k API_KEY | kubectl apply --namespace=my-namespace -f -

are they both the same, like i have one deployed with prefect server on gke but i need to use prefect cloud so is one agent enough for both server and cloud or i have to deploy seperate one for cloud?

I’m wondering on how to remove completed prefect jobs on k8s?

Hi all - I am wondering what the best way is to determine what agent a flow is running on while its in progress

I’m having some trouble wrapping my head around designing a conditional flow based around Great Expectations Validation and was wondering if any one had any advice. To summarize- we have an ETL Task, after which we want to run the Validation Task. If the Validation Task sets its state as Failed we want to run a Notification Task that alerts us, but otherwise we want the subsequent tasks to continue. Where I’m getting stuck is that it seems like the routing tools that I’ve found don’t quite fit right: • Using Triggers doesn’t feel correct, because I don’t want to run my Notification Task on

all_failed

or

any_failed

- if the ETL Task fails, it shouldn’t run. What I believe I would need is something like an

on_x_task_failed

option- it seems like the available options are way too broad to be useful. • I then looked into some of the ideas on the Conditional Logic page, but this seems clumsy for a few reasons. I need a Task specifically to check the State of the Validation Task, and then on our Schematic View we have additional Tasks showing up for each Case Task as well as the Merge Task. All combined, it makes our Schematic look like the image below which seems crazy for what is effectively

if x do y

. I feel like I have to be approaching this in completely the wrong way- if anyone had any ideas or suggestions I would be extremely grateful.

Theory question: Is it possible to have a (non-task) abstract class in which the methods are tasks? I have a flow structure that gets repeated frequently for different data sets, I'm trying to create an abstract class in which the repetitive tasks are methods with the necessary text inputs, and then the 1-2 non-repetitive methods are abstract methods in the parent class and are thus implemented in child classes. The problem though is that when I do this, the @task decorators are messing with calling the methods internally. I have to pass in "self" manually, and if I need to specify any non-data dependencies (e.g. add an upstream_tasks input for the decorator) the methods seem to break whether "self" goes before or after the decorator input. Is there a way to do this? Or will this approach just not work? Really hoping there is a way since this will enable me to MASSIVELY reduce my code replication.

Hi, I have a list of flows that are very similar and I wanted to create a factory function in order to make things easier to maintain. The code looks very similar to this:

from prefect import Flow, Parameter, task

from etl import prepare, load

from crawlers import crawler1, crawler2

from typing import Dict, Callable, String

prepare = task(prepare)

load = task(load)

def make_standard_pipeline(flow_name: String, func :Callable, func_params:Dict):

    

with Flow(flow_name) as flow:

        

params = {k: Parameter(k, default = v) for k,v in func_params.items()}

        

df = func(**params)

        

df = prepare(df)

        

df = load(df)

    

return flow

pipe1 = make_standard_pipeline('flow1', task(crawler1), {})

pipe2 = make_standard_pipeline('flow2', task(crawler2), {'type' : 'xxx'})

Locally this code runs fine and I can also register these on the prefect server. However, when I try to run the flows, only the first flow defined in the file runs successfully (I tested reordering the flows). For the other flows, I get Key Error saying it couldn't found task slug crawler2. Does anyone has hints on what could be causing this problem? Thank you.

Hi，I'm new to Prefect, but I'm using airflow for a while, we got an issue in airflow, so I wonder if Prefect could handle it better than airflow. We tried to run 40,000 tasks (very simple tasks, just print commands, with 1000 dags, 10 dag runs per dag, 4 task per dag run. ) at same time, we found airflow took a very long time to finish them (about 1 hours) , our server resource is good enough, it looks like scheduler is the bottleneck, so I wonder would Prefect be better than airflow to handle many tasks at same time. What's the max amount of task run at the same time you have seen? How long Prefect takes to finish them? Thanks

Hi everyone, first question, please be gentle 🙂 I am sending a parameter called video to my "Process video" workflow. The first task in this workflow is to upload the video to an s3 bucket. The upload task is transcribed in the thread: My main problem seems to be that s3 requires a string filename, but my video is of the type <Parameter: video>, as illustrated by the error message

ValueError: Filename must be a string

So there is clearly some sort of.. unwrapping/unpacking that I'm missing? my flow is likewise in the thread, as is the invocation. So... would someone please tell me in which exact angle I should smack my forehead and the correct octave of the "duh!"... plus maybe what I should be doing instead?

Hello everyone! I'm new here. I'm handling a new DE role, and after trying the others (Airflow, Luigi, Dagster) I chose prefect for being that weird characteristic of being the most fun to use. It's absolutely amazing. Would anyone know how to capture the exact full python traceback when a task fails? I've set up state_handlers but the State.message only captures an abbreviated form of the error (e.g. ZeroDivisionError) . Thanks!

Hi all, i wanna create a api, that use to call to execute a task in prefect. How to do that? thanks all

Hey 🙂 I'm trying to send a notification on Slack in case of task failure, but unfortunately I don't receive anything 😕 • First, I followed the instructions on : https://docs.prefect.io/core/advanced_tutorials/slack-notifications.html#customizing-your-alerts with

slack_notifier

, without success. • Then I tried with

SlackTask

, this is my piece of code (didn't work) :

def post_to_slack_on_failure(task, old_state, new_state):
    if new_state.is_failed():
        if isinstance(new_state.result, Exception):
            value = "```{}```".format(repr(new_state.result))
        else:
            value = str(new_state.message)
        msg = (
            f"The task `{prefect.context.task_name}` failed "
            f"in a flow run {prefect.context.flow_run_id} "
            f"with an exception {value}"
        )
        SlackTask(
            message=msg,
            webhook_secret="<https://hooks.slack.com/services/*******/*******/*******>"
        ).run()
    return new_state

• Finnally, I tried with a custom solution (didn't work, again) :

def post_to_slack_on_failure(task, old_state, new_state):
    if new_state.is_finished():
        msg = "Task {0} finished in state {1}".format(task, new_state)
        # replace with your Slack webhook URL secret name
        secret_slack = cast(str, Secret("<https://hooks.slack.com/services/*******/*******/*******>").get())
        <http://requests.post|requests.post>(secret_slack, json={"text": msg})
    return new_state

This is my test flow :

@task
def task_error():
    raise Exception("Test")

with Flow("Test Slack", state_handlers=[post_to_slack_on_failure]) as flow:
        task_error()

    flow.run()

(I also tried EmailTask with smtp_type="INSECURE", and once again it didn't work) Need some help please 🙏

Hey guys! Is there a way to hand in an endpoint_url to Prefect when using s3 storage? We use a containerized localstack instance for local testing/dev in place of s3.

Hey All, I'm setting up a new instance of Prefect and I seem to be misunderstanding something about flows. I've got a bunch of methods with @task decorators and then I set them up in what should be a single a flow. When I load this into Prefect though I get two split DAGs that seem to function independently. Flow setup here and picture of the result attached.

with Flow(flow_name) as flow:
    <http://logger.info|logger.info>(f"{flow_name} Task Flow initiated, running in {file_path}")
    df = pull_oracle_data_via(oracle_query_sql=oracle_query_sql_path,prod=use_oracle_prod)
    df = set_data_types(df)
    upload_to_table(df, destination_table = data_destination_table_name)

    if summary_view_name is not None and history_table_name is not None:      
        <http://logger.info|logger.info>("Initiating history upload process.")
        summary_df,summary_data_empty = pull_summary_data_via(sql=f"SELECT * FROM {summary_view_name}")
        if summary_data_empty:
            delete_today_from_history_if_exists(df=df,history_table=history_table_name)
            upload_to_history_table(df=summary_df, destination_table=history_table_name, append=True)
    else:
        <http://logger.info|logger.info>("Skipping summary view run: summary view name and/or history table name missing.")

To address this I tried to make the dependencies explicit, by adding "upstream_task" flags to two of the above lines as so:

summary_df,summary_data_empty = pull_summary_data_via(_upstream_tasks_=[upload_to_table],_sql_=_f_"SELECT * FROM {summary_view_name}")

delete_today_from_history_if_exists(_upstream_tasks_=[pull_summary_data_via],_df_=df,_history_table_=history_table_name)

This doesn't seem to fix the issue though; when I run the flow, later tasks still seem to initiate before the Oracle pull, which should occur before everything. Anyone see what I'm doing wrong? The documentation would seem to indicate that feeding result data from one task to another would make dependencies work correctly, but that doesn't seem to be happening here.

hi, maybe I’m tired today, but I can’t figure out if I can set / change the context variable inside of the Flow block.

Hi, anyone has tried to use shelltask to run powershell scripts?

Does prefect have a mobile app?

Is there a good way to read in custom SQL, then delete 14 days of data from the target table, then deposit that custom SQL for the last 22 days into the target table, ensuring no duplicate records? Something like

extract => transform => delete 14 days in target => append last 22 days w/ no dupes

. Right now I"m doing 2 parallel ETLs, one which ends in replacing the existing data and the other which appends. This seems very slow since I have to perform 2 full extracts.