Quick question about schedules. I want a “once per quarter” schedule, so I came up with this:

schedule = Schedule(
    clocks=[IntervalClock(interval=timedelta(days=1)),],
    or_filters=[
        between_dates(start_month=1, start_day=1, end_month=1, end_day=1),
        between_dates(start_month=4, start_day=1, end_month=4, end_day=1),
        between_dates(start_month=7, start_day=1, end_month=7, end_day=1),
        between_dates(start_month=10, start_day=1, end_month=10, end_day=1),
    ],
)

Is that idiomatic Prefect? Or is there a more natural way of doing this? Another potential solution is to use

IntervalClock

with an interval of 3 months, except you can’t say

timedelta(months=3)

and get the interval to fire on the first of the month easily.

Hi! Quick question about the equality operator. Noticed that

prefect.core.task

understandably does not override

__eq__

__neq__

methods to be able to suppoort task comparisons as explained in

is_equal

method docstring: https://github.com/PrefectHQ/prefect/blob/1959eecf1bbbb8e3b194b288c197a6108deb8693/src/prefect/core/task.py#L924-L938 Just to confirm, this would be an appropriate way to resolve task result equality within a flow, right?

with Flow('my-flow') as flow:
    task_a = get_task_a_result()
    task_a_is_value = task_a.is_equal('value')

When creating a state handler, how can I tell when a task fails the last retry? tried

def my_state_handler(obj, old_state, new_state):
    if new_state.is_finished() and new_state.is_failed():
        # send notification

but this did not work I want to send a notification only if the last retry fail

Hi 🙂 - trying to run the intro tutorial i.e. https://docs.prefect.io/core/tutorial/01-etl-before-prefect.html and I've having troubles with the OpenSky API by the look of it (presumably all APIs have bad days?) Should I just wait til it gets "better" to run this part of the tutorial or is something else going on? Cheers, Michael

(prefect) mjboothaus@Michaels-Air tutorial % python 01_etl.py 
...TRUNCATED - See thread

As a follow-up -- I ran the 2nd part of the tutorial

02_etl_flow.py

and I guess this illustrates the more helpful messages/info provided by the prefect library when tasks fail

(prefect) mjboothaus@Michaels-Air tutorial % python 02_etl_flow.py
...TRUNCATED... See thread

SOLVED - seems API was down - now OK

Just wondering if the full graphql schema is hosted anywhere? I’m trying to use a codegen (https://www.graphql-code-generator.com/docs/getting-started/codegen-config) to generate bindings for a typescript/nodejs application. I tried to use https://api.prefect.io/graphql but it wasn’t able to download the schema.

hi team, wondering if we could override/enrich the prefect context? i.e. I would like to have a task that does

from prefect import context as prefect_context, task

@task
def enrich_context():
    prefect_context["custom_key"] = 1

@task
def main_task():
    value = prefect_context["custom_key"]
    print(value)

with Flow("test-enrich_context") as flow:
   _enrich_context_task = enrich_context()
   main_task(upstream_tasks=[_enrich_context_task])

Hy every one, I am using Docker storage for my flow and get this error

ModuleNotFoundError: No module named 'utils'

when running

python file.py

It seems like that my local package utils is not found during docker build. Here is the structure of my folder. 

src
	utils/
	file.py
	__init__.py

I am using a custom docker image where I already copied every thing in src in the image. Please help me on this. Any help is appreciated 🙏Many thanks already

Hello everyone, it's possible to execute multiple flows where each one uses a specified virtualenv (I'm using conda by the way)? For example, let's suppose that I have two envs with different libraries in each one, one in Python 3.6 (env_1) and another in Python 3.8 (env_2), and I want to run two Flows, one using the env_1 and another the env_2. I know that a possible approach would be using a containerized application, but since I do some file/folder manipulation on the host, I think that would be tricky to do with docker (please correct if I'm wrong). Obs: I'm using the Prefect Cloud

Hi guys, is there any guide of monitoring Prefect Agent using tools like New Relic or Datadog?

Hi people, I'm getting this weired error while using custom Docker image with DockerRun on Prefect Server, the error pops up immediately after the agent pulled the image successfully, and no task has been executed yet. I didn't use

rstrip

anywhere in my code and the flow runs smoothly with LocalRun. I wounder is it possible for me to get a full stacktrace on this error... Thanks for your help!

How do I change logs to debug when running a flow with

prefect run

?

Hi - can someone point out what I'm missing in the following set up - I have a Server stood up, and separate machine running:

prefect agent local start --api http://<server_IP>:4200

Hi there. I am using the new

get_task_run_result

and

create_flow_run

tasks described here. I'd like to treat the child flow as a single unit that can be retried or restarted if the flow fails. Currently, these are two separate tasks and I haven't been able to set them up this way. How could I change my flow to support retries/restart when the child flow fails?

Hi folks - we’re setting up a new prefect agent in our Kubernetes cluster on Google Cloud and I could use some pointers getting the correct permissions configured. I’m starting with this:

prefect agent kubernetes install -k $KEY --namespace=staging --rbac | kubectl apply --namespace=staging -f -

I then submit a flow that uses the

KubernetesRun

config and

GCS

storage configured as:

run_config = KubernetesRun(cpu_request=2, memory_request="2Gi", image='<http://gcr.io/carbonplan/hub-notebook:c89f7f1|gcr.io/carbonplan/hub-notebook:c89f7f1>', env={'TZ': 'UTC'})
storage = GCS("carbonplan-scratch", project='carbonplan')

This results in an error message like this:

└── 23:15:59 | INFO    | Submitted for execution: Job prefect-job-d8a8a648
└── 23:16:05 | INFO    | Entered state <Failed>: Failed to load and execute Flow's environment: Forbidden('GET <https://storage.googleapis.com/storage/v1/b/carbonplan-scratch?projection=noAcl&prettyPrint=false>: Caller does not have storage.buckets.get access to the Google Cloud Storage bucket.')

So, I gather my agent doesn’t have the correct IAM privileges to read the flow from GCS. Next I tried adding a service account to my agent:

prefect agent kubernetes install -k $KEY --namespace=staging --service-account-name pangeo --rbac | kubectl apply --namespace=staging -f -

Here I’m pointing to my kubernetes service account called

pangeo

which has been given

storage.objectAdmin

permissions. However this results in the same error as above. So now I’m wondering if I’m missing something more fundamental here. If anyone has suggestions on where to look for more details on setting up prefect on GKE, I’d certainly appreciate it.

Hi folks, how can I write a file to the prefect working directory? I have defined my own path using the

prefect_directory

param in the

Docker

storage. I am able read files from that defined path but I cannot write files to that directory - these are files that will be used in the flow execution. Getting

[Errno 13] Permission denied: my/path/filename

Hi All, What is right way to access a returned result of a task which is a python class object? In my case I have a task that returns a python object; but attempting to access it with simple variable assignment seems to output NULL as a value for that value.

Hi all, is it safe to say that the job template inherits the image from the dockerfile used for Docker storage when passed to

KubernetesRun

at runtime as an image is not required for the job spec?

how can we control the size of the logs to make sure they do not get to large on one hand, but to make sure it keeps enough history ?

Hi Team, I have Prefect cloud setup on GKE cluster with Dask Gateway. I would like to leverage all the power of Dask (I’m assigning 12-20 workers, each worker is set to have 16GB memory) when reading/writing data to/from BigQuery. I have

pandas.read_gbq()

or

<http://pandas.to|pandas.to>_gbq()

defined within my tasks but the execution of flow seems to be using only one Dask worker instead of distributing the load and I’m getting into memory/timeout issues. I’m getting errors and warnings like this in the GKE dask worker logs:

INFO - Event loop was unresponsive in Worker for 4.40s. This is often caused by long running GIL holding functions or moving large chunks of data. This can cause timeouts and instability. 

WARNING - Memory use is high but worker has no data to store to disk. Perhaps some other process is leaking memory? Process memory: 6.40 GB -- Worker memory limit: 8.59 GB.

WARNING - Worker is at 80% memory usage. Pausing worker. Process memory: 6.89 GB -- Worker memory limit: 8.59 GB

WARNING - Worker exceeded 95% memory budget. Restarting.
/opt/conda/lib/python3.7/multiprocessing/semaphore_tracker.py:144: UserWarning: semaphore_tracker: There appear to be 1 leaked semaphores to clean up at shutdown.

distributed.utils_perf - INFO - full garbage collection released 842.28 MB from 584 reference cycles (threshold: 10.00 MB)
distributed.nanny - INFO - Worker process 10 exited with status 1

Do I have to use Prefect’s

BigQueryTask

class if I want the

DaskExecutor

to be utilizing all of its workers and cluster options as set? or Do I have to change something in the Dask Gateway config? How do I tell Prefect to use all the assigned Dask workers when running tasks?

@Kevin Kho just curious, do you think there's a potential way to leverage the PandasSerializer from https://github.com/PrefectHQ/prefect/pull/3020 in conjunction with the question here? the

to_dict()

approach that you mentioned in this thread is similar to what we're doing (we return

list(df.itertuples())

instead) but it's not ideal for the same reasons that

PandasSerializer

is nice in general compared to using cloudpickle. I can't really think of anything off the top of my head though that wouldn't require messing with the .map internals (cc @Chris White @Zanie too in case y'all have any clever ideas). can also make a discussion on github if that's preferable

Is it possible to separately apply a job_template with kubectl and register a flow without passing the job_template? I think that would be done by passing the image explicitly in the

KuberneetsRun

?

has registering flows been slow for anyone else today? For me it seems to hang intermittently at this line

Cloning <https://github.com/PrefectHQ/prefect.git> (to revision master) to /tmp/pip-install-2dvduvf6/prefect_9381dd189c7849aa9cd40f1a3fa2afae

at various points today.

Hey guys, do you have any examples or do you even suggest using Prefect to manage ML pipelines. Currently we have a very naive pipeline that creates 100's of features in order. It takes an hour or so, and I am thinking of a way that each feature could be created in a parallel as a flow. Would you suggest this as a good use case for Prefect?

Hi Folks, I would like to know how Prefect reacts to

sys.exit(1)

. I have this line inside of a task and if the execution reaches this line, does it exit with

1

and mark the

task

as successful or failed? Or does it think the

flow

failed and try to restart the

flow

?

Hello everyone, I see that a task as an arbitrary function can be skipped by raising SKIP signal.

from prefect import task
from prefect.tasks.shell import ShellTask
from prefect.engine import signals

shell_task_invoke = ShellTask()

@task
def skip_if_non_prod(env):
    if env == "PROD":
        return shell_task_invoke(command="ls")
    else:
        raise signals.RETRY()

Can I skip subclasses of class Task i.e. ShellTask, DBTShellTask by passing parameter something like below?

is_not_prod = False if env == "PROD" else True
ShellTask(skip=is_not_prod)

I have one flow importing modules from a different directory on a cloud Prefect.io I have something like:

import os
import sys

PATH = 'Module path'
sys,path.append(PATH)

from <my module> import <my func>

when doing a test function, without a Prefect flows, I can run it from the command line. when running the same function from a flow, I can run it when using

flow.run()

, but when registering it in the cloud Prefect and trying to run it, I get

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

I am running a local agent on my machine,

prefect agent local start

I do not use Docker containers any suggestions ?

Hello everyone, I want to map a task, inside a task, but whenever I try it gives me the following error: `ValueError: Could not infer an active Flow context while creating edge to <Task: map_fn>. 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

map_fn.run(...)

Below is the replication of what I want to do:

from prefect import Flow, task

numbers = [1, 2, 3]

@task
def map_fn(x):
    return x + 1

@task
def reduce_fn():
    res = map_fn.map(numbers)
    print(res)

    return res + [1]

with Flow('flow1') as flow1:
    print(reduce_fn())

flow1.run()

Any suggestions or workaround will be appreciated.

Hey guys, I'm seeing some pretty substantial delays in my jobs where my agent appears to be having SSL connection issues going from AKS with (prefect:python37-latest) to Prefect cloud with my flows registered on Azure Storage. I've seen in previous comments that this could be to do with rate limiting? urllib3.exceptions.ReadTimeoutError: HTTPSConnectionPool(host='10.0.0.1', port=443): Read timed out. (read timeout=None) Details in thread

Hello, we always use checkpoints and want to enable checkpointing in Python by default to avoid potentially error-prone extra steps for correct environment setup. It seems that the following approaches do not work (before creating a flow):

prefect.config.flows.checkpointing = True

or

os.environ["PREFECT__FLOWS__CHECKPOINTING"] = "True"

Instead, we need to ensure that no

prefect

import statement (direct or indirect) occurs before setting this environment variable. It seems impractical that prefect reads such configuration in top-level code (executed on import) and not in functions. Are there any better solutions?

Hello, I'm looking for a way to pause a task if a certain condition is met, and resume it after the user confirms. Here is a minimal example:

from prefect import task, Flow, Parameter
from prefect.engine.signals import PAUSE

@task
def dummy_task(x):
    confirmed = ???
    if x > 100 and not confirmed:
        raise PAUSE("Are you sure ?")
    return x + 1

with Flow("dummy") as flow:
    x = Parameter('x', default=1)
    dummy_task(x)

Is there a clean way of getting previous states of a task, maybe from

prefect.context

, such that I know confirmation happened ? Or do I have to add some branching to handle this ?