prefect-community #prefect-community

Prateek Gupta

09/13/2021, 5:39 AM

Can I get some resources for ML deployment?

Prateek Gupta

09/13/2021, 5:39 AM

It is batch prediction

Prateek Gupta

09/13/2021, 5:41 AM

One thing my team mate claimed is that task name had issues when making multiple commits

Prateek Gupta

09/13/2021, 5:43 AM

So ended up renaming the flow name and version

Anh Nguyen

09/13/2021, 7:38 AM

Hi everybody! I created docker file to build image for etl flow(prefect). That only run 1 time. I wanna to run schedule and image status not exited

haf

09/13/2021, 12:23 PM

Can I return a python list from a Task implementation? I'm getting `TypeError: Task is not iterable. If your task returns multiple results, pass

nout

to the task decorator/constructor, or provide a

Tuple

return-type annotation to your task.` but I can't use

nout

because I don't know the result size and it is not a good fit for Tuple (since it's a list of tuple)

haf

09/13/2021, 1:01 PM

It would seem dask.distributed has had a few releases https://github.com/dask/distributed/releases since March 4th when https://github.com/dask/distributed/pull/4460 was merged for Python 3.9 support? Time to support in Prefect? (https://github.com/PrefectHQ/prefect/issues/3453)

Sam Cook

09/13/2021, 2:08 PM

Does anyone know if there is a simple function call or graphql mutation to update a flow's README? I was hoping there was a field I could pass a README (as a string) through in the registration process, but it doesn't look like it.

Brian Phillips

09/13/2021, 7:01 PM

Does S3Result use the default

AWS_CREDENTIALS

secret? I'm having trouble setting credentials so that tasks are able to write with S3Result. This pattern does not seem to work either

Copy code

aws_credentials = PrefectSecret('AWS_CREDENTIALS')
with prefect.context(secrets={'AWS_CREDENTIALS': aws_credentials}):
    <task>

Ishavpreet Singh

09/13/2021, 7:06 PM

HI, I am working on using DBTShellTask as part of my Poc with prefect. I see my Dbt shell Task fails with exit code 2... but apart from this I don't see any details on error.. Any ideas on where can I find additional logs ?

Max Ivanchenko

09/13/2021, 7:14 PM

Hi everybody! I'm researching the possibility to use prefect instead of airflow. Everything looks good but has prefect any alternative to airflow variables? (Parameters that can be configured from UI, not before each runs but once when needed)

Owen McMahon

09/13/2021, 7:47 PM

Hi! I have a Mapped Task that I have checkpointing setup for (with the

map_index

in the filename so it properly writes out each mapped task to an individual result) within a flow running against Prefect Cloud. I just came across a weird scenario where the Flow did run the Mapped Task fully through (100 Mapped Tasks in total), but noticed afterwards that 7 of them had a status of 'Cached'. This caught my eye - as it should not have loaded any of them from the Cache. When I looked closer at the logs of one of the 'Cached' Mapped Tasks - it looks like it finished successfully, and then restarted ~7 mins later and loaded from Cache. It appears that all data is still there as I expected - but behavior seemed a bit odd. Wondering if anyone else has seen this before? Thanks!

Ben Muller

09/13/2021, 8:10 PM

Hi Team, do you guys have any examples of an implementation of a "flow builder" Class that someone has built? I have about 10 flows that are identical besides a few minor details. I do not want to use different parameters because I feel like it lacks the proper visibility in the UI and each flow should have its own name. I'm just unsure of how to implement something like this so that registration, running etc of the flow will still work. Can I simply have a class that is invoked with some arguments and then a method which when called runs the flow context manager with the flow inside of it?

Kyle McChesney

09/13/2021, 8:28 PM

is there a good way to define a flow with

set_dependencies

, but also get its result and pass along to other flows. I.E. mixing and matching imperative vs functional?

Kyle McChesney

09/13/2021, 8:28 PM

I have a startup task which I want to run at the start of every flow, which just logs some info, etc

Kyle McChesney

09/13/2021, 8:28 PM

I have not found a good way to force it to be at the start of the DAG

Kyle McChesney

09/13/2021, 8:29 PM

I am trying to use

set_dependencies

on the actual first tasks, with my startup_task as an upstream

data = flow.set_dependencies(generate_data(data_csv_url), upstream_tasks=[startup()])

Kyle McChesney

09/13/2021, 8:30 PM

(data_csv_url is a parameter in this case)

Kyle McChesney

09/13/2021, 8:30 PM

data =

should be the result of

generate_data

Kyle McChesney

09/13/2021, 8:31 PM

seems like a flow state handler might be the better approach 😐

Brian Phillips

09/13/2021, 8:41 PM

How can I clean up a DaskExecutor after a flow has finished? I'd like to call

Cluster.stop()

Constantino Schillebeeckx

09/13/2021, 9:09 PM

Is there anyway to get a locally run flow to return a non zero exit code if it fails? Notice in the screenshot shown below, the last exit-code was 0.

Constantino Schillebeeckx

09/13/2021, 10:14 PM

Where can I find a list of all API urls which I can access with prefect.client.client.Client.get
? For reference, I'm trying to hit he following without success:

Copy code

[2021-01-14 16:00:00.000] ERROR    --- 400 Client Error: Bad Request for url: <https://api.prefect.io/flows?name=dbt_test>
Traceback (most recent call last):
  File "deploy/register_flows.py", line 491, in <module>
    main()
  File "deploy/register_flows.py", line 485, in main
    create_proj_and_register_flows(flows, args)
  File "deploy/register_flows.py", line 278, in create_proj_and_register_flows
    register_flow(flow, flow_file, args)
  File "deploy/register_flows.py", line 314, in register_flow
    flow_already_registered(flow.name)
  File "deploy/register_flows.py", line 319, in flow_already_registered
    resp = CLIENT.get('flows', params={"name": flow_name})
  File "/Users/constantino.schillebeekx/.pyenv/versions/dwh/lib/python3.8/site-packages/prefect/client/client.py", line 406, in get
    response = self._request(
  File "/Users/constantino.schillebeekx/.pyenv/versions/dwh/lib/python3.8/site-packages/prefect/client/client.py", line 710, in _request
    response = self._send_request(
  File "/Users/constantino.schillebeekx/.pyenv/versions/dwh/lib/python3.8/site-packages/prefect/client/client.py", line 620, in _send_request
    response.raise_for_status()
  File "/Users/constantino.schillebeekx/.pyenv/versions/dwh/lib/python3.8/site-packages/requests/models.py", line 943, in raise_for_status
    raise HTTPError(http_error_msg, response=self)
requests.exceptions.HTTPError: 400 Client Error: Bad Request for url: <https://api.prefect.io/flows?name=dbt_test>

Brad

09/14/2021, 2:55 AM

Hi team - following on from https://github.com/PrefectHQ/prefect/issues/3619, I'm seeing a looping task exit early with

Cached

despite not being complete. I wonder if this is the expected result or a bug. Example in thead

Gaylord Cherencey

09/14/2021, 5:11 AM

Hey team dummy question, if I have 3 flows

and

I want flow

and

to be triggered each time flow

is run successfully but I want to define this dependency in flow

and

because the "parent" will be owned by another team. Is there a way to do this in Prefect or if we would have to go for an eventing solution?

Jacob Blanco

09/14/2021, 7:05 AM

Edit: Figured it out, it should be

with case(...)

not

if case

. Having a strange issue where when running a flow it warns that a Parameter was declared but not added the Flow, and when I try to run the flow with the parameters established it errors out saying that the parameters are unexpected. This is the structure of the flow:

Copy code

with Flow(name="My Flow") as flow:
    if case(Parameter("do_something", default=False), True):
        result_of_thing = run_thing()
    else:
        result_of_thing = run_another_thing()

Copy code

from flow_definition import flow
flow.run(parameters={"do_something": True})

✅ 2

Hugo Polloli

09/14/2021, 7:35 AM

Hi, I'm building a state handler and I'm wondering if it's possible to attach information myself to the task, or state that will be available inside of the handler. Ideally being able to do something like this

Copy code

@task(state_handlers=[my_handler])
def my_task(my_input):
    self.message = f"Input {my_input} wasn't ok"
    assert input_is_ok(my_input)

    self.message = f"Failed to do_some_things for {my_input}"
    res = do_some_things(my_input)

    self.message = f"Failed to do_a_few_more for {my_input}"
    res2 = do_a_few_more(res)

    return res2

I understand I could use signals, using the value property of a signal to store what I want to print, but that would mean wrapping each row I'm performing an action in inside a try/except block, then raising the signal inside the except block, I've done that and it gets really cluttered I maybe missed something obvious and am going in the wrong direction though...

Issam Assafi

09/14/2021, 7:43 AM

Hello, What if in the same flow, i need multiple versions of Python for each task, or let's say different versions of some packages for each task... is this something Prefect can handle and how?

shekhar koirala

09/14/2021, 8:01 AM

my code is simple as this

Copy code

@task(max_retries=3, retry_delay=datetime.timedelta(minutes=10))
def fetch_data(count,check):
    data = []
    logger = prefect.context.get("logger")
    <http://logger.info|logger.info>(data)
    return data
    

def run():
    with Flow("fetch-data") as local_flow:
        count = Parameter("count", default=100)
        check = Parameter("limit", default=10)
        fetch_data(count,check)

    local_flow.register(project_name="test")
    id = local_flow.run()
    print(id)
        
if __name__ == "__main__":
    run()

Lucas Beck

09/14/2021, 8:20 AM

Hi Everyone, My colleagues and I recently started migrating our internal tool to Prefect, in order to see if it could support our use cases. So far we have had a generally good experience, specially with the simpler use cases. Recently though, we reached a few walls that have been hard to overcome. We have tried to go through the documentation and search this slack channel without much luck, so that is why I am now writing here to see if we can get some help. Let me quickly describe our overall setup followed by the challenges/blockers to see if someone has any suggestions. Please do share your thoughts also in the way organize things, perhaps a different way of organization could help. Our setup and use cases We have various different

ETL

like jobs to run, some on schedule and some on demand. We usually break down big computing tasks in smaller parts, each contained in its own

docker

container. These tasks, when combined, form a computation DAG. This DAG can mostly be parallelized (mapped). We then leverage an Azure Kubernetes Service (AKS) cluster to horizontally scale the compute jobs into hundrends, sometimes thousands of containers. So far we have been using our own internal tool to orchestrate tasks and manage dependencies, and that is what we are currently investigating if we can replace with prefect. Our prefect setup has a self hosted server instead being the cloud solution. Challenge 1 The first and most important challenge we are facing regards scaling jobs. I will use the flow we tried to migrate to prefect as the example here. I am attaching the DAG schematic as a picture. In this flow, each task use prefect's

RunNamespacedJob

to spin up a new job that has exactly one pod, which will perform a given computation. We then use other functions from the prefect SDK to read the logs after completion and delete the kubernetes jobs. The whole DAG can be run in parallel (mapped) for multiple inputs. Usually the job works fine for 3-5 inputs, but as soon as we try to scale to as little as 30 inputs we start seeing a ton of heartbeat errors or no pod statuses (also attaching an image here) . Once running with our internal tool, we already scaled up to around 500 inputs in parallel in the same cluster setup. Anyone else has experienced this? By the way, we are using the

DaskExecutor

for that with default parameters Challenge 2 We currently have a single repo where we want to keep all prefect flows under version control. In order to reuse code, we also made it a package that can be imported around. The structure looks like this:

Copy code

prefect-flows/
  setup.py
  requirements.txt
  prefectflows/
    flows/
      flow1.py
      flow2.py
      ...
    core/
      kubernetes.py
      ...

The idea is that the actual flows sit under the

flow

folder and can import boiler plate/ utils from the

core

folder. For instance, under

kubernetes.py

are all functions that use prefect's SDK to interact with kubernetes. An example function is

run_job

which will run a job, read logs and eventually delete that job. This all works fine if running locally, but fails once registering a flow and trying to run it through the server. The error is :

Copy code

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

It is important to mention that I have dockerized the package using the prefect base image, and I am telling prefect to run the flow with that image. Something like the following:

Copy code

Flow(
        flow_name,
        storage=Azure(
            container=container,
            stored_as_script=False,
            connection_string=azure_connection_string,
        ),
        run_config=KubernetesRun(
            image="imagewithourpackageinstalled:latest",
...

Maybe I misunderstood where the unpickling is happening, but would expect that if the flow uses an image that has all packages used when registering that flow, then that should work. I think that summarizes it, but please let us know if you need further description or to share more of the code. Thanks in advance for your help!