Does anyone know how to display information about another broken task or the error that caused the fail in a task? I have very simple etl flow with one alert task. This alert task is executed if any of the etl tasks fail. I want to use this alert task to send a further telegram notification and I want to include meaningful information in this notification than just the "Error" message. Maybe I should use a completely different approach? I tried to get information through context, but this information is not what I need at the moment.

from prefect import task, Flow
from prefect.triggers import any_failed
from random import randint

@task()
def extract():
    # this task generates random integers from 0,1,2,3,4,5.
    random_number = randint(0, 5)
    return random_number


@task()
def transform(random_number):
    # this task is potentially dangerous, since the random number can be zero, and then there will be division by zero
    data = 100 / random_number
    return data


@task()
def load(data):
    # just print data
    print(data)


@task(trigger=any_failed)
def telegram_alarm():
    # this task is needed to send notifications about fails in telegram.
    print('ERROR')  # How can I display a more informative message? For example, which task broke? Or what error caused the fail? (in my case ZeroDivisionError: division by zero)

    # import prefect
    # print(prefect.context.to_dict())  # tried this method but didn't find the needed logs


with Flow('test_etl') as flow:
    random_number = extract()
    data = transform(random_number)
    load(data)
    telegram_alarm(upstream_tasks=[flow.get_tasks()])

flow_state = flow.run()

what should the entrypoint be for custom docker images in a

KubernetesRun

?

Hello Prefect community! I'm starting out with Prefect, I have many similar tasks that been a pleasure to automate, -> I attach a sample code snippet of my flow in the thread 🧵 to avoid spamming here. My questions are: 1. I couldn't figure out if instead of the three for loops I could/should be using map/iterated mapping (I've made .map() work well with the equivalent of 1 for loop but not the three nested ones to go over all my parameters) 2. I couldn't figure out how to execute 'depth first' ( = execute the task_inference task after each long task_training is complete). I've read in the prefect docs that Depth First Execution is with Dask executors. In my case I don't want parallel execution since each task takes the entire available GPU memory of my local machine i'm running these on. Many thanks for any pointers and corrections of my misunderstandings

Hi, I'm working with a larger than memory dataset. The operations needed can be easily parallelized and be split into chunks and I have done that with

.map()

. The problem is that when I run it the results are all still stored in memory (

bigger_than_mem

runs twice and keep resuslt in memory). Is it possible to have the flow the the data as a file, and clear the memory? Here is a sample flow

@task

def get_chunks():

return [[1,2],[3,4,5]]

@task

def bigger_than_mem(x):

return x*100000000

@task

def dump_to_db(x)

dumb(x)

with Flow("my_flow") as flow:

x = get_chunks()

x_trans = bigger_than_mem.map(x)

dump_to_db.map(x_trans)

As far as I found out,

state.TriggerFailed

is a child of

state.Failed

. When one of my tasks fails, then I get the

state.Failed

from this task and then from the next task, which actually didn't break, I get

state.TriggerFailed

with message _TRIGGERFAIL('Trigger was "all_successful" but some of the upstream tasks failed.',)._ I would like to receive information only about the failed task. At the moment I have implemented the following logic, is this the best practice?

if isinstance (new_state, state.Failed) and not isinstance (new_state, state.TriggerFailed):

I feel like we need a #kudos channel but anyway: Big #kudos to @Tyler Wanner and @Kevin Kho for putting up with all my constant questions and queries and being really quick to help out 💯 🥇

can I assign task to two different agents.

I mean, One task1 is running on local agent and task2 is running on the docker agent

and put some conditions if the any task fails, the process fail ??

Hey all! I am having trouble understanding the change from personal access tokens to api keys. What I understand is that they act in the same way. We have multiple projects that are using prefect. We are using it to run our ELT pipeline with docker, meltano, redshift, and dbt. The proper way to use them is to create a service account for each of the projects that will handle the authentication and store the keys in env variables we call 

PREFECT__CLOUD__AUTH_TOKEN

. Can we just replace references to the personal access tokens (

prefect_auth_token

) to the new name

PREFECT__CLOUD__AUTH_TOKEN

? Any guidance would be appreciated. Thanks!

Hi, I'm trying to use the Docker storage method. The image gets correctly saved in my registry but I get the following error in my agent :

docker.errors.APIError: 400 Client Error for <http+docker://localhost/v1.40/containers/create>: Bad Request ("invalid IP address in add-host: """)

Could anyone help ? Thanks

Hi! I’m trying to trigger a flow from another flow using

StartFlowRun

and running into an issue where the flow cannot be found, even though that flow exists. I’ve attached the stack trace of the error as well as a screenshot of our flows, to show that we do have one named

final_stats-staging

. Any help here would be great- thanks!

Hi, Any other users of MLFlow? I am trying load a model from S3 using MLFlow in a tasks, but it does not seem to be picking up the AWS credentials (passed through Prefect using PREFECT__CONTEXT__SECRETS__AWS_CREDENTIALS env var). Thanks!

Can anyone point me to a way to pass a priority to a Dask task from map()? I can work with one priority for all the tasks being mapped, but it would be really cool if I could assign a priority to each of the mapped values.

Does anyone know if there is a way to query cloud api for late flow runs?

We have this massive script that is super important to our day to day operations, and I'm looking to convert this into a prefect-cloud oriented setup so that I can connect it with a slack webhook and also have access to logs anywhere/everywhere. But I'm having trouble setting this up correctly, for a few reasons: 1. These tasks in the graph are inter-twined, meaning that a lot of these tasks are called within a "task" itself. Example:

def fnRunTask(parameter1,  parameter2, parameter3 .... parameter10):
    df1, df2 = fnRunSubTask(parameter3, parameter5)
    dfTmp = fnRunSubTask2(df1, parameter1, parameter7, parameter9)

2. As shown above, each subfunction requires different parameters, and the number of parameters for each function can vary. Essentially this is all headache from converting a bunch of stuff that was defined under "if name == _main_:" into a prefect setup. I've played around with passing in one large dictionary at the start, but wanted to ask here before I continued debugging that (didn't work on tries #1-20 lol) Also wanted to ask about the task interconnectivity, as I was a bit confused about upstream/downstream tasks since some of them are "at the same time" within the task that calls them.

Hi everyone, Does Labels could be utilized as the agent pointing only? can we use it as general usage of label? Because I tried labelling my flow as ie: label1, label2, label3 And my agent was: label1 The flow won't run unless I remove label2 and label3

Regarding Approval steps in Flows, are there any restrictions on which type of account can Approve the continuation of the flow? Do you need to be a User?

Also I'm banging my head against the wall

# fnPortfolioOptimization()
from prefect import Client
client = Client(api_token = %TOKEN%)
client.login_to_tenant(tenant_slug='pwcm')

with Flow('test2') as flow:
    lookback_periods = Parameter("lookback_periods", default=[3, 5, 7])
    percent_change = Parameter("percent_change", default = [.05, 0.07, 0.1])
    lookback_days_rsi = Parameter("lookback_days_rsi", default = 20)
    rsi_overbought = Parameter("rsi_overbought", default=0.75)

    fnRunAlgo(lookback_periods, percent_change, lookback_days_rsi, rsi_overbought)

prefect.utilities.exceptions.ClientError: Malformed response received from Cloud - please ensure that you have an API token properly configured. I've tried a custom service-account token, as well as my user api key, neither work when calling flow.register(project_name) or client.register(flow, project_name)

Does multiprocessing library work with Prefect flows?

Hi everybody, when i tried to rerun my failed mapped task from prefect cloud by clicking on the restart button, it prompts me this error

Unexpected error: TypeError("object of type 'NoneType' has no len()")

. However, when i rerun a fresh one, it works! Is my way of caching wrong? prefect = 0.14.15

Is there a recommended way of exporting

flow_run

data (or any other entity) into a data-warehouse? We had a poor man's workflow engine that we are replacing by prefect. But we need to get the

flow_run

data to our data-warehouse, in order to then use the reporting tools. We were thinking of doing some sort of ETL that queries Prefect's GraphQL API, and loads it into our data warehouse. But we were wondering if there is a recommended way of achieving this? GraphQL's subscriptions do no work yet, right?

Hello everyone! Unfortunately, I did not find it in the documentation, is it possible to implement authentication in the Prefect Server UI? Even the simplest one. I understand that this can be done in the Prefect Cloud, but what about the Server version?

Hey Folks, I'm running into this exception `ValueError: Cannot set

mapped=True

when running from inside a mapped context`` As a dummy example, I'm trying to do something like this:

from prefect.tasks.core.function import FunctionTask
from prefect import Flow, apply_map


def task_block(z):
    R = FunctionTask(lambda x: range(x))(z)
    M = FunctionTask(lambda x: x + 1).map(R)
    return M
    
with Flow('nested_map') as flow:
    L = [1, 2, 3]
    apply_map(task_block, L)

is there a way to do this kind of nested mapping differently?

Hi all, unfortunately I couldn't join the Livestream - Data science on Prefect last Friday. Is there a recording somewhere that I can watch?

I have a task that is using the trigger “any_failed” as a way to “clean up” if any upstream tasks fail. My expectation is that this task would be “skipped” if the upstream tasks succeed. In actuality the task is marked as failed and my flow failed. I got around this by setting a reference task, but it seems strange to mark a task as having failed when it didn’t run. Is there a way to change this to skip? Thx!

Hello every one,  This is my example flow that I want to run with docker as Storage, KubernetesRun as run_config and DaskExecutor in an existing cluster as executor.

import prefect
from prefect import task, Flow
from prefect.storage import Docker
from prefect.run_configs import KubernetesRun
from prefect.executors import DaskExecutor

STORAGE = Docker(registry_url="<http://prefect.azurecr.io|prefect.azurecr.io>", image_name="prefect/test")

RUN_CONFIG = KubernetesRun(image="<http://prefect.azurecr.io/prefect/test|prefect.azurecr.io/prefect/test>")

EXECUTOR = DaskExecutor(address="<tcp://my-ip:8786>")


@task
def hello():
  logger = prefect.context.get('logger')
  <http://logger.info|logger.info>("Hello!")


with Flow("changeme", storage=STORAGE, run_config = RUN_CONFIG, executor = EXECUTOR) as flow:
  hello()

 When I run this flow, the flow is successfully submitted but never got executed and the status is forever pending. Do I miss something? Do I need to specify

flow.run()

in my code? Many thanks already for helping me 🙂🙏

Hi all, just reposting what I had in the server channel. I hope someone can help. We’re deploying an ECS agent as a container into a Fargate ECS cluster. As far as I can tell, the load balancer etc are all configured sensibly - they use the same settings as other containers which work. Prior to the agent starting, we can see logs on Cloudwatch (for instance, if we send a broken command), but once it’s launched, there is nothing. The container health fails, and the task goes into a health check loop. We’re seeing lots of messages telling us that a task failed to load the image, but again, we know the image is accessible and is being loaded initially, because we can see logs from it when we change the command in the Dockerfile. I want to understand more about what’s going on here. We have no NAT or internet gateway configured, due to security restrictions, which we know will prevent the agent from communicating with Prefect Cloud, but we would expect some sort of message to tell us so. Could someone explain why we’re getting the messages about ECR? Is the agent trying to load additional instances of itself into the cluster?

Hi All I am trying to set up slack notifications to a channel i created from prefect, I have the SLACK_WEBHOOK_URL saved and it doesn't seem to be working, i'm getting this error:

Caused by NewConnectionError('<urllib3.connection.HTTPConnection object at 0x000002C254411C10>: Failed to establish a new connection: [WinError 10061] No connection could be made because the target machine actively refused it'))

Hi all, I have written a Flow to perform an ETL process. Throughout the course of the Flow thousands of API requests are being made (using the .map function). Consequently, my Flow takes over 24 hours to make the API calls and parse all the data. Also, the Flow is using a local Dask executor and it’s running on an ECS container. Does anyone have any ideas on how I might be able to shave some time off my Flow execution?