Hi everyone! Does dask executor support versioning? Let's say I have two versions of a codebase for my tasks, each lives in a separate docker container. I'd like to be able to run both versions, e.g. flow1 runs inside container1, and flow2,3 run inside container2. This could be achieved without dask AFAIK, but dask is really handy for resources management, such as GPU memory.

Hello folks! The Prefect team has the day off in recognition of Election Day in the US, and so might not get to all questions today. We appreciate your patience and understanding and we’ll be back tomorrow!

Hello Everyone! We are using Prefect cloud. We have dev and prod environments on separate EC2 instances. Each env has an agent configured and we can see 2 agents/ 2 flows for dev and prod in Prefect cloud. My question is how do we know for sure the prod flow will always run the job in the prod EC2 instance? Is there a possibility of the Prod flow running the job on dev EC2 and vice versa. cc @Adrien Boutreau

Hello Prefect team 👋, I'm adding

S3Result

to my flows and I have encountered a weird error (template issue?), which I tried to replicate in a minimal example fashion. I'm running the flow through Prefect server which I set up using

prefect server start

and using local agent

prefect agent start

. More info in thread

Hey, what's the standard way of storing Secrets for Prefect Core?

We've got our own cluster that we're running it on. Is there an easy for it to talk to EKS or something? Or what are Best Practices here?

Also, just wanted to say I'm super happy that I got a new gig where we're using Prefect! 🙂

I’m wondering about using

apply_map

with functions that create parameters. Here’s a minimal example:

from prefect import Flow, Parameter, unmapped, apply_map

def func(x, y):
    z = Parameter('z')
    w = x + y + z
    return w

with Flow('flow') as flow:
    x = Parameter('x')
    y = Parameter('y')
    w = apply_map(func, x, y)

---------------------------------------------------------------------------
ValueError: Parameters must be root tasks and can not have upstream dependencies.

The error comes when

apply_map

tries to add edges from the root tasks of the subgraph to the argument tasks passed to

apply_map

. This fails when it tries to make an edge from

x

to

z

since

z

is a parameter. This error seems to preclude ever using

apply_map

with a function that creates a parameter. I’m wondering if that constraint is necessary. My intuition is that if

apply_map

sees a parameter that is in the subflow but not in the parent flow, it should assume that any edges from that parameter are unmapped edges that should be added to the parent flow. Does this seem right?

Hi folks, I know flow cancelling was recently introduced in prefect, but I am curious why *prefect.engine.state.State.is_cancelled*() wasn't also introduced ?

Hi, a newbie question… If the trigger for all tasks in a flow is “all successful”, and one of the upstream task fails, all of the downstream tasks are “trigger failed” (even if the upstream failed task is restarted and succeeds). This forces me to restart the trigger-failed tasks one by one. Is this the intended behavior? or am I missing something? Should I change the trigger to “all finished” in order to avoid this manual restarting of the downstream tasks when one of the upstream tasks fails at least once?

Hi Folks, Did anyone receive the prefect 5k stars goodies? - https://prefect-community.slack.com/archives/CKNSX5WG3/p1601480661016400

Hello all, Sorry in advance for this vague issue but I couldn't be more specific 😕 In my case, flows are split in different groups (different sub-folders) based on their similarity and for my convenience, I have created a shell script to register specific groups of flows each time. so far, so good. When I register a flow I have noticed that the scheduling of another flow is deregulated. Let's say I have flow

A

and flow

B

. cron for flow B is

0 11 * * 3

(1 event every Wednesday at 11:00am). Last time I registered flow

A

I noticed that two runs had been scheduled for flow

B

for the upcoming Wednesday (both at 11:00am). When I re-registered flow

B

, scheduling was fixed. Have you any idea why this happening? Thanks. python:

3.7.3

prefect:

0.13.12

OS:

Debian GNU/Linux 10 (buster)

Had anybody tried to pass results/states between tasks that are in different flows? Thank you! https://prefect-community.slack.com/archives/CL09KU1K7/p1604173648410700

I have written mysql connection function in different file and imported in my flow it works well when i use flow.run() But when I run using prefect server UI it says module not found . I have attached code .Please help

Hi, any pointer on how to use

map

with

StartFlowRun

?

problems with connecting Prefect agent Hey, I have a prefect server running which I am having problems getting an agent connected. The agent is running on the machine, and starts up with the message:

Agent connecting to the Prefect API at <http://localhost:4200>

but, I'm unable to see or use the agent. (If I run curl on localhost:4200, I can connect without issue.)

Hello, I wonder how a task could process the result of a failed upstream task. Please see the example below, I have a

task_a

and then I have a

task_b

which takes the

task_a()

as upstream task.

task_b

always runs and process the task_a result even when the task_a fails

def task_a(threshold: float):
  result = random()
  if result > threshold:
   return result
  else:
    # question 1: how do I get task_a to return the result of random() even when the FAIL signal is raised. 
    raise signals.FAIL("failed, result is less than the threshold")


@task(trigger=triggers.always_run)
def task_b(result_a: float):
  # question 2: how could task b check if the task_a is a success or failure? 
  if result_a.is_failed():   # ???
    print ('task_a failed with result {}'.format(result_a))
  else:
    print ('task_a suceeds with result {}'.format(result_a))


with Flow('task_b always handles task_a failure') as flow:
  result_a = task_a(0.3)
  result_b = task_b(result_a)

I have two questions: 1. how do I get

task_a

to return the result of

random()

even when the

FAIL

signal is raised. 2. how do

task_b

check if the

task_a

is a success or failure?

Hello, I would like to raise a feature request to allow ShellTask to return stdout when it fails.

https://github.com/PrefectHQ/prefect/issues/3614

please see this link or the thread for the details

Hey guys, new here to Prefect. The DevOps team at my company created a server for us but i can't seem to connect.

curl -I <http://myserver>

seems to return

HTTP/1.1 200 OK
Date: Wed, 04 Nov 2020 20:57:57 GMT
Content-Type: text/html
Content-Length: 12382
Connection: keep-alive
Server: nginx/1.18.0
Last-Modified: Tue, 13 Oct 2020 23:49:16 GMT
ETag: "5f863cfc-305e"
Accept-Ranges: bytes

But i can seem to connect on the UI.

Hi everyone, first thanks for all work on Prefect, good job! 👏 We are in the phase of testing Prefect as a potential replacement of Airflow, but I am not sure if I understand the Agent part of concepts correctly. In Airflow we have workers and there are many of workers across our infrastructure (servers) - we can scale just by adding workers - to be able to run more and more tasks. And worker is asking if there are some tasks what worker can do. As I understand, workers are called Agents in Prefect and they have similar function - am I right? If yes, how can I scale the Agents (distribute them across servers) or how the scaling is done and is it possible to scale community version? Thanks for any explanation! cc @Radek Tomsej

Hey Folks, I'm hopeful that someone can give me some tips as to what I am doing wrong with the BigQueryLoadGoogleCloudStorage Task. I have built a flow to export data from a MySQL database and then upload it to GCS (all works fine). When the flow attempt to run the BigQueryLoadGoogleCloudStorage task, I get the following Error:

Unexpected error: PicklingError('Pickling client objects is explicitly not supported.\nClients have non-trivial state that is local and unpickleable.',)

I'm configuring the security credentials for GCP in

.prefect/config.toml

under the

[context.secrets]

section using the

GCP_CREDENTIALS

field. Doing

flow.run()

, the flow completes sucessfully. I then do

flow.register(...)

and then start the agent with

flow.run_agent()

and manually kick off a run in the UI. The following is an example of how I am adding the

BigQueryLoadGoogleCloudStorage

task to the flow:

from prefect import Flow
from prefect.tasks.gcp.bigquery import BigQueryLoadGoogleCloudStorage
from google.cloud.bigquery import SchemaField

load_to_bigquery = BigQueryLoadGoogleCloudStorage(
    dataset_id="data_lake_dev",
    project= "data-platform"
)

with Flow("Move Data From DB to BQ") as flow:
    load_to_bigquery(
        uri="<https://storage.cloud.google.com/data-platform-dev/a-csv-file.csv>",
        table="website_data",
        schema=[
            SchemaField(name="id", field_type="INT64", mode="REQUIRED"),
            SchemaField(name="uuid", field_type="STRING"),
            SchemaField(name="site", field_type="STRING"),
            SchemaField(name="path", field_type="STRING"),
            SchemaField(name="type", field_type="STRING", mode="REQUIRED"),
            SchemaField(name="reported_at", field_type="TIMESTAMP", mode="REQUIRED"),
            SchemaField(name="helpful", field_type="BOOL", mode="REQUIRED"),
            SchemaField(name="message", field_type="STRING")
        ]
    )

Hello, I want to create dask cluster on Azure. Is their any API's available in Prefect which can create Dask Cluster in Azure .

Hey, I’m trying to use the

LOOP

and write the results on each iteration as documented here. However, it doesn’t seem to work. It only writes the last iteration.

@task()
def log_output(result):
    logger = prefect.context.get('logger')
    <http://logger.info|logger.info>(result)


@task(result=LocalResult(dir='./results', location='test-{task_loop_count}.prefect'))
def loop_test():
    loop_payload = prefect.context.get("task_loop_result", {})

    n = loop_payload.get("n", 1)
    print(n)

    if n > 5:
        return n

    raise LOOP(f'Iteration {n}', result=dict(n=n+1))


with Flow("Postgres -> BigQuery") as flow:
    x = loop_test()
    log_output(x)

See output logging and diagnostics in thread

Hi folks, does

@task

support both

name

and

task_run_name

at the same time? It seems that

name

takes precedence over

task_run_name

, am I right?

Hey everyone, I’ve got some

secrets

that I need to use in most of my tasks in a flow. I hate the visual of an upstream dependency to the same secret task from almost all my tasks. Is there a way for custom secrets to behave like, say

AWS_CREDENTIALS

, as per: https://docs.prefect.io/core/concepts/secrets.html#default-secrets Or a prefect backend compatible way of having my secrets reside in

prefect.context

, or a similar context like construct?

Hey folks, I'm thinking about using Prefect at a medium-sized company. I'm looking for guidelines on how to set up a flow that spans multiple teams / code repos. For example, node A in the dag may be owned by team A, runs in container A, outputs a large dataset to be consumed by the node B, owned by team B, in container B. Does anyone know of best practices for such a situation?

Hi I am running prefect on an on premises HPC with no docker (can be installed) and my flow in not really serialisable (I need to figure out which step). i was just wondering what is the best SOP for tracking the source of issues that can brake the flow for debugging. In the current setup the output to the terminal shows only the log message for start and end and none of the logs i put in the code. When i test the flows locally all the logs output are visible at the terminal. Thanks a lot!

Hello, community, our prefect flow run gets the error of no heartbeat from time to time. It seems retrying, but actually nothing happen, the normal 20 mins run gets stuck there for hours. Any idea how to fix that?

Dear all, while the flow was running perfectly before and still runs locally, I suddenly get an error that seems to be related to pickle when running it on cloud:

TypeError: can't pickle _thread.lock objects

See full error message below. Any ideas, what could be the reason?

Is there a way to use a simple function in a flow, for example when I am going to call the function many times. I can easily make it work by making my function a 'task', but I can't see why it should have all the scheduling overhead:

@task
def get_host_from_kv(opts):
    hid = Schema({
        'host': str,
        Optional('identity', default=0): int
    }).validate(opts.kvopts)
    return HostParam(identity=hid['identity'], host=hid['host'])


with Flow("Query a single host") as flow:
    opts = Parameter('opts')

    result = winrm_powershell(host=get_host_from_kv(opts),
                              command=load_script(opts, 'win_ps_version.ps1'),
                              opts=opts)