prefect-community #prefect-community

Leon Kozlowski

01/31/2022, 9:10 PM

What will happen if I deploy a new version of a flow while a previous version is in a running state?

Daniel Kornhauser

01/31/2022, 10:26 PM

Can anybody point to a relatively large project that uses prefect in github to checkout their directory tree structure ? Wondering how they group and name the python files of their flows and tasks when they have a lot of them.

Yorrick Mendes

01/31/2022, 10:34 PM

Hey prefectors, I’ve got a rather basic question. Is it possible to add 2 flows as tasks/dependencies to a main flow? It doesn’t look like it’s possible besides unpacking each flow and adding all the tasks and dependencies individually. The other way to go about it would be to add two tasks that each run a flow locally but that doesn’t seem like the cleanest solution and I’d probably lose a lot of the UI sugar with that approach

Suresh R

02/01/2022, 7:40 AM

Hi! Each mapped task is getting executed twice in my flow, What might be the reason?

Copy code

INFO
CloudTaskRunner
Task 'collate-quarterly-crdlotcookie[5]': Starting task run...
INFO
CloudTaskRunner
Task 'collate-quarterly-crdlotcookie[5]': Starting task run...

Vinoth Kumar

02/01/2022, 8:03 AM

Hi unable to install prefect Orion . I am getting ImportError: cannot import name 'ThreadedChildWatcher' from 'asyncio'

Tony Waddle

02/01/2022, 12:03 PM

Sorry if I'm missing something obvious - how do you cancel a running agent? Can't see it in docs, -h or ui

Muddassir Shaikh

02/01/2022, 12:31 PM

The slack alerts sent by state handlers

handler = slack_notifier(only_states=[Failed])

do not contain details of the flow name and the taskname, how can we add that?

Konstantin

02/01/2022, 12:41 PM

Copy code

Hello everyone, I can't figure out how to use native libraries in Prefect, for example "prefect.tasks.gsheets"
Give an example of the code for using this class

Muddassir Shaikh

02/01/2022, 12:54 PM

The search bar is not able to find the taskname when the task name are created from a function. Example: PREFECT_SPLIT_FILES..... is a taskname but when searched doesn't appear. Does it have something to do with task_name and task_run_name ?

Madhup Sukoon

02/01/2022, 1:57 PM

Hello! I am extremely new to Prefect, so apologies if this is a rookie question or has already been answered. I have a repository containing a bunch of scripts, folders and modules. I want to import a subset of the functions defined here and build a Flow in a new

flow.py

file and then run it via a Docker Agent. I also have some external (pip-installable) dependencies that would need to be present for the flow to run. Is it possible to do this? Does someone have an example Flow that uses Github Storage and Docker Run?

Vladimir Bolshakov

02/01/2022, 2:09 PM

Hi! Is it possible to customise base URL of Orion UI in static builds? I have started standalone Orion server (without agent) and UI is not refreshed, because of CORS errors. UI app try to fetch data from

localhost:4200

. I think the reason is here: https://github.com/PrefectHQ/prefect/blob/orion/orion-ui/src/plugins/api.ts#L198

Madhup Sukoon

02/01/2022, 5:24 PM

Hi! Just wondering if it's possible (maybe using

env

host_config

) to have the Docker Agent install packages in the docker container at runtime? (Looking for something similar to Docker Storage's

python_dependencies

, but for runtime)

Constantino Schillebeeckx

02/01/2022, 5:44 PM

I've got a question about the ECS Agent documentation, in this section, there is a "Agent role policy" example, but there is no mention (I think) of where that gets used. The documentation only goes through the "execution role" and the "task role", where does "agent role" come into play?

Madhup Sukoon

02/01/2022, 6:16 PM

I am using Prefect Cloud. My task Goes to

Scheduled

Submitted

and then nothing happens. The only entry in logs is:

Copy code

Submitted for execution: Container ID: 143cabe06c721.....

My flow uses GitHub Storage and DockerRun, and uses some secrets. Any way I could get more verbose logs on what might be happening? This is what my

flow.py

looks like:

Copy code

from main import planner, creator, publisher

planner = task(planner)
creator = task(creator)
publisher = task(publisher)

foo = Secret('foo')
bar = Secret('bar')

with Flow(
    "myflow",
    run_config = DockerRun(
        env={"EXTRA_PIP_PACKAGES": "requests pyyaml"}
    ),
    storage = GitHub(
        repo="me/myrepo",
        path="flow.py",
        access_token_secret="GITHUB_ACCESS_TOKEN"
    )
    
) as flow:
    plan = planner()
    media = creator(foo, bar, plan)
    post = publisher(foo, bar, media)

main

is an adjacent python file (

main.py

)

Davoud Hejazi

02/01/2022, 9:22 PM

I want to add slack notifications in case there is an error while running the flow. I have the lines below as part of my code:

Copy code

from prefect.utilities.notifications import slack_notifier
from prefect.engine.state import Failed
SLACK_WEBHOOK_URL = "the slack secret webhook url"
# os.environ["PREFECT__CONTEXT__SECRETS__SLACK_WEBHOOK_URL"] = "<https://hooks.slack.com/services/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>"
handler = slack_notifier(only_states=[Failed])
prefect.context.setdefault("secrets", {})
prefect.context.secrets["SLACK_WEBHOOK_URL"] = "the slack secret webhook url"

I have even created config.toml file in .predict directory with the content below:

Copy code

[context.secrets]
SLACK_WEBHOOK_URL = "the slack secret webhook url"

My task stats with:

Copy code

@task(nout=2, max_retries=1, retry_delay=timedelta(seconds=1), checkpoint=False, state_handlers=[handler])

I also tried to add prefect app to slack. But when running my flows I am seeing this error (part of the error): see thread To make it clear, when I remove the

state_handlers=[handler]

from my task definition the flow runs fine. When I visit the prefect app in slack I see this too, which I am not sure if it connected to my problem or not. Any comments on how I can solve the slack notification problem?

Kevin Mullins

02/01/2022, 10:33 PM

Any advice on how to troubleshoot a job being scheduled but not submitted to Kubernetes? I have Prefect Core 0.15.11 with the agent installed in k8s. I ran a first flow which is running still, and then triggered a separate flow that got scheduled in the prefect UI but not submitted to K8S. I don’t say any logs from the agent saying it is deploying the flow run with that ID.

Shaoyi Zhang

02/01/2022, 11:49 PM

Is there a timeout for submitted flow runs with Kubernetes agent when there’s not enough resources in the k8s namespace?

Horatiu Bota

02/02/2022, 8:37 AM

hi community & thanks for the help! i've got a parameters json file that i want to pass to a prefect flow using

flow.run(parameters=json.loads(open(file).read()))

-- the parameters file looks something like this:

Copy code

{
    "other_params": {},
    "company_params": {
        "company_id": "ID1",
        "company_name": "Company",
        "datasets": [
            {
                "dataset_label": "label",
                "dataset_url": "URL"
            },
        ]
    }
}

is there any way to access nested values in the actual flow? something like:

Copy code

with Flow("flow") as flow:
        company_params = Parameter("company_params")
        datasets = company_params.get("datasets")

Suresh R

02/02/2022, 10:45 AM

Hi!, I have a mapped task with 21 childs, also set task concurrency of 5, But only 2 child tasks are scheduled at a time, What might be the cause?

Chris Arderne

02/02/2022, 12:09 PM

@task

decorator has

log_stdout=True

but is there any way to add

stderr

to logging?

shijas km

02/02/2022, 2:50 PM

Hi I am facing an issue I need to add a shell task before my python tasks in a flow but its not working , throwing some error like flow outside cant be called but when i comment out all task except the shell task it works can I combine both shell task and python task in a single flow? any example??

Constantino Schillebeeckx

02/02/2022, 3:36 PM

is there some sort of API endpoint that we can hit which serves as a health check? I'm wondering how I can programmatically check that the agent is healthy

David Wang

02/02/2022, 4:02 PM

Hi everyone, I am currently trying to setup an ECS agent with aws. However, the task seems to stop and exit with the error

essential container in task exited

. The prefect ecs agent also does not appear in the UI. I have been able to start the ECS agent locally, but am unable to start it through aws. Another thing to mention is that when I try to add the logConfiguration to try to see what could be going wrong with the service it will give me an error of

ResourceInitializationError: failed to validate logger args: : signal: killed

. I’ve double checked with devOps that the IAM roles and network configurations should be correct too. Any ideas on how to debug this or why this is happening?

E Li

02/02/2022, 4:08 PM

Hi, my task is mapped to some children tasks. if some of the children tasks fail in a run, is there a way to skip the successful ones and only re-run the failed? Can this be achieved by state handlers? thanks

Jason Motley

02/02/2022, 5:26 PM

If I set up a "flow of flows" orchestration and schedule it, and the flow of flows all runs before the scheduled runs for the individual flows, will the flows just run again (unnecessarily)?

An Hoang

02/02/2022, 5:38 PM

I have a

flow_A

with

task1, task2... task10

. Now I want to construct

flow_B

with

task1, task2.... task 20

and

task11_result = task11(task10_result)

. How do I add the tasks of

flow_A

as upstream tasks for

flow_B

. I know I can do

create_flow_run

, but I do not want monitoring for

task1...task10

flow_B

Kind of like:

Copy code

with Flow("flow_A") as flow:
    flow_A_param1 = Parameter("flow_A_param1")
    ...
    flow_A_param_n = Parameter(...)
    ...
    task1 = ....
    ...
    task10_result = task10(...)

with Flow("flow_B") as flow:
    flow_A_param1 = Parameter(...)
    ...
    flow_A_param_n = Parameter(...)
    
    flow_A_result = flow_A.run(parameters = {"flow_A_param1": flow_A_param1,...., "flow_A_param_n": flow_A_param_n}) #dont want monitoring for these tasks
    task10_result = flow_A_result.result[flow_A.get_task(task10)]
    task11_result = task11(task10_result)
    ...
    task20_result = task20(...)

where `flow_B`'s parameters are just `flow_A`'s parameters. But in the example above I always have to repeat the parameters of

flow_A

flow_B

and keep them both up to date with each other. Is there a better way to do this?

Chris Reuter

02/02/2022, 6:00 PM

👋 Hi everyone! We're working with Gradient Flow to collect some data on workflow orchestration use cases and features. We want to know what's important to our community as we move forward. Please consider taking this 10 question survey. All responses are anonymous, of course.

upvote 2

Devin Flake

02/02/2022, 7:51 PM

Is there any chance that Prefect could release a arm64 image?

Nelson Griffiths

02/02/2022, 9:58 PM

I am running into some unexpected behavior with extra loggers in prefect cloud. When I add my extra loggers to the config and run my flow with a local agent everything works as expected. When then switching to the DockerRun config and Bitbucket storage my extra loggers no longer get passed to the Cloud Logs. Is there something extra I need to be doing when moving from a local agent to a docker agent to get my logs to be passed through?

jspeis

02/03/2022, 1:43 AM

I'm using pylint and it seems like when I use the

upstream_tasks

kwarg (e.g. as in example below) I get an

Unexpected keyword argument 'upstream_tasks' in function call (unexpected-keyword-arg)

... is there a way I can get pylint to recognize this is ok? (other than disabling?)

Copy code

@task
def add_one(val):
    return val + 1

with Flow("Add One") as flow:
    p = Parameter("param")
    first = add_one(p)
    second = add_one(p, upstream_tasks=[first])