prefect-community #prefect-community

Rob Fowler

09/15/2020, 10:08 AM

@Marvin why won't you enter me?

Marvin

09/15/2020, 10:08 AM

I’d give you advice, but you wouldn’t listen. No one ever does. Good luck in the contest @Rob Fowler.

Rob Fowler

09/15/2020, 10:09 AM

replaced all the broken bits of ansible with my pywinrm fork, vault and prefect

🚀 2

🤩 1

Nuno Silva

09/15/2020, 10:13 AM

Hi. I'm using

DaskExecutor

with LocalEnvironment. Agent runs locally, Dask cluster is running in azure aks. Randomly I get the error:

OSError: Timed out trying to connect to '<tcp://dask-user-44479c3f-5.pipelines:8786>' after 3 s

. I tried to increase the default timeout with

dask.config.set({"distributed.comm.timeouts.connect": 20})

but apparently to no avail, 3s is kept. Ideas how to increase the timeout? Thanks

Rob Fowler

09/15/2020, 10:28 AM

can you set it with kwargs on your DaskExecutor initialisation?

Rob Fowler

09/15/2020, 10:30 AM

timeout=XX ?

Rob Fowler

09/15/2020, 10:33 AM

if that is not handled it also passes a dict with cluster_kwargs = {..}

Rob Fowler

09/15/2020, 10:34 AM

https://github.com/PrefectHQ/prefect/blob/65155e9b066a092c22c202def31293e9dd580f5f/src/prefect/engine/executors/dask.py#L79

Julian

09/15/2020, 11:51 AM

Hey all, how would you write a (complex) schedule to trigger a flow run on both the first of the week and the first of the month, but only once if

first_of_the_week == first_of_the_month

? Is this possible, just using schedule and filters or do I need to check this condition within the flow run and stop one of the schedule runs? It's crucial the flow run is not executed twice, just because both conditions are met

kursat.t

09/15/2020, 2:26 PM

Hi there, We are looking for azkaban alternative, evaluating airflow and prefect. At our daily batch computation flow, we have 800+jobs. I wonder that what is the max num of tasks that prefect can handle in single flow?

Ben Fogelson

09/15/2020, 3:17 PM

Is it possible to use

LocalResult

to save results from a flow running in a docker container to the host filesystem? Here’s a toy example of what I’m trying to do:

Copy code

from prefect.environments.storage import Docker, Local
from prefect.engine.results import LocalResult
from prefect import Flow, task

@task
def foo():
    return "foo"

with Flow('run_local', storage=Local(), result=LocalResult()) as run_local_flow:
    foo()
run_local_flow.register("proj")

with Flow('run_docker', storage=Docker(), result=LocalResult()) as run_docker_flow:
    foo()
run_docker_flow.register("proj")

When I kick off a run of

run_local_flow

, the

foo

task produces a result on my host filesystem (e.g.

/Users/ben.fogelson/.prefect/results/prefect-result-2020-09-15t15-11-21-771486-00-00

) as expected. When I kick off a run of

run_docker_flow

, the UI says that there is a result saved to an analogous location, but there isn’t anything on my host filesystem (I’m guessing it is getting saved to the container filesystem). Being able to run a flow with docker storage but inspect the results locally would be super useful for development.

Kyle Pierce

09/15/2020, 5:12 PM

Are there any documentation or blog articles on how to register a flow on docker? I have the container setup and running but I cant register a flow. Getting an error with the graphql url

requests.exceptions.HTTPError: 400 Client Error: Bad Request for url: http://{name}.<http://us-east-1.elb.amazonaws.com:4200/graphql|us-east-1.elb.amazonaws.com:4200/graphql>

Theres a env variable with my host name in the .env file.

Samuel Koshy

09/15/2020, 5:54 PM

@Marvin

Marvin

09/15/2020, 5:54 PM

It gives me a headache just trying to think down to your level. I'll still enter you in the contest, @Samuel Koshy.

👍 1

Chandra Manginipalli

09/15/2020, 6:37 PM

Anyone know how to pass GraphQL variables in Interactive API. I tried the following with no luck for state. Syntax Error: Expected Name, found \"{\

mutation($state: JSON!) {

set_flow_run_state(input: {flow_run_id: "83b4e6be-b1ac-416c-833c-236c23f5e1ca", version: 5, state: $state}) {

id

state: {

type: "Success"

message: "It worked!"

👀 1

Chandra Manginipalli

09/15/2020, 6:38 PM

Also, I was able to run a flow successfully. Happen to know how to issue pause/resume queries in GraphQL for a flow run?

kiran

09/15/2020, 6:52 PM

@Marvin

Marvin

09/15/2020, 6:52 PM

Don't feel you have to take any notice of me, please. I'll just enter you in the contest @kiran.

kevin

09/15/2020, 7:37 PM

When calling

Client().create_flow_run()

what is the best way to inject some custom context? Would it be something like this at a high level:

Copy code

client = Client()
  with context(run_id=some_run_id):
    client.create_flow_run()

As a follow up, the use case for the above question is that we want to inject a custom run_id into the env variable

PREFECT__LOGGING__LOG_ATTRIBUTES="['run_id']"

so that our logger can collect that attribute. Would this be the correct approach?

Brett Naul

09/15/2020, 10:39 PM

doesn't feel like it's worth a github issue but this is already done, right? https://docs.prefect.io/orchestration/agents/kubernetes.html

Resources

The current default resource usage of a prefect-job has a request and limit for CPU of
100m
and the agent limits itself to
128Mi
for memory and
100m
for CPU. Make sure your cluster has enough resources that it does not start to get clogged up with all of your flow runs. A more customizable Kubernetes environment is on the roadmap!

Vitaly Shulgin

09/15/2020, 10:40 PM

Hello team

Vitaly Shulgin

09/15/2020, 10:40 PM

Copy code

Attempting to call `flow.register` during execution of flow file will lead to unexpected results.

Vitaly Shulgin

09/15/2020, 10:41 PM

I tried to generate, and register flow from within running flow, and got the above error message? Is it possible to register new flow from running flow?

Vitaly Shulgin

09/15/2020, 10:43 PM

Or, what will be solution to implement such functionality?

sark

09/16/2020, 3:01 AM

hi guys from the docs i know how to pass a parameter to a new task i define myself, but how do i pass parameters to an existing task like

CreateContainer

sebastian.clanzett

09/16/2020, 6:30 AM

Hi guys. Is it possible to run tasks from a flow on different specialised agents?

Julian

09/16/2020, 6:45 AM

Hey, I already mentioned my problem in the thread https://prefect-community.slack.com/archives/CL09KU1K7/p1600170685303800 Summarized, If I use a custom filter for my schedule and register the flow (with this schedule attached to it), it throws an error

Klemen Strojan

09/16/2020, 7:18 AM

Hey all, We have a master flow that looks like this:

Copy code

# schedule
daily_schedule = CronSchedule("0 7 * * *", start_date=pendulum.now(tz="Europe/Vienna"))

# result
RESULT = AzureResult('prefect', connection_string_secret='my_azure_secret')

with Flow("master_flow", schedule=daily_schedule, result=RESULT) as flow:
    flow_1_param_1 = FlowRunTask(
        flow_name="flow_1",
        project_name="project_1",
        wait=True,
        parameters={'param': '1'})
    flow_1_param_2 = FlowRunTask(
        flow_name="flow_1",
        project_name="project_1",
        wait=True,
        parameters={'param': '2'})
    flow_1_param_3 = FlowRunTask(
        flow_name="flow_1",
        project_name="project_1",
        wait=True,
        parameters={'param': '3'})
    flow_2 = FlowRunTask(
        flow_name="flow_2",
        project_name="project_1",
        wait=True)
    flow_3 = FlowRunTask( 
        flow_name="flow_3",
        project_name="project_1",
        wait=True)
    flow_1_param_1.set_downstream(flow_1_param_2)
    flow_1_param_2.set_downstream(flow_1_param_3)
    flow_1_param_3.set_downstream(flow_2)
    flow_2.set_downstream(flow_3)

###############################################################################
# Settings

flow.storage = GitHub(
    repo="some_repo",
    path="some_path",
    secrets=["GITHUB_ACCESS_TOKEN"]
)

# Labels for identifying agents
flow.environment = LocalEnvironment(labels=["label_1", "label_2"])

We are using Cloud and have two agents with local Dask clusters. Labels from the master flow points towards one agent and labels from all other flows point towards the second agent. When we run the master flow, everything runs normally, no errors or warnings in the log. But in reality, only

flow_1_param_1

flow_2

, and

flow_3

ran. What are we missing?

Robin

09/16/2020, 8:22 AM

Hey folks, we are finally scaling up our dask kubernetes on EKS and having quite some fun with it! ☺️ While investigating on some (probably external) performance issues, I was wondering whether there is a way to illustrate task duration histograms from the prefect cloud UI? I found the dask dashboard super helpful to understand, where bottlenecks are. Give me a handshake 👋 if you would be interested in such a feature. 🙂 Right now, something like a simple histogram of number of tasks vs. task durations (maybe with a binning parameter) combined with a task filter would be already super helpful. Both to see, which children within a mapped task take longer than others as well as to see which mapped tasks take longer than others... Anyway, we are already appreciating the UI quite a bit, also because the mobile version allows to check on flow state pretty seemlessly! 🙏

❤️ 1

✔️ 1

👋 5

Dinu Gherman

09/16/2020, 11:59 AM

Hi @Marvin!