prefect-community #prefect-community

Channels

prefect-community

Adam Brusselback

02/10/2021, 3:33 PM

Had a question on just the general approach to automate tasks that need to run on specific machines in the environment? Like I would like to get my database backups scheduled through Prefect, so should I have a Prefect agent installed on my backup server so I am able to kick off that task with shell? This is in addition to my other jobs which are more client-specific ETL tasks able to run from just about anywhere behind my firewall without issue

1 reply · 2 participants

Samuel Hinton

02/10/2021, 4:08 PM

Having ye old issue of “Flows are submitted and an agent sees them, but they get stuck in ‘submitted for execution’“. I am wondering if anyone knows how to debug this. I know that: • the labels on the task and the agent match • the docker agent has pulled the image, started a container and “completed flow run submission” from the logs • Using either the Local executor or LocalDaskExecutor does not change the fact that: • Flows never progress from the “submitted for execution” state Is there a way of trying to look inside an agent to figure out what is not working?

7 replies · 3 participants

Mitchell Bregman

02/10/2021, 5:10 PM

I have had this issue with 2 flows now, getting a flow state of

Failed

on the UI, while every task and the flow itself is working as expected.. what can be causing this?

[10 February 2021 12:06pm]: Unexpected error: TypeError('Could not serialize object of type Success.\nTraceback (most recent call last):\n File "/usr/local/lib/python3.7/site-packages/distributed/protocol/pickle.py", line 49, in dumps\n result = pickle.dumps(x, **dump_kwargs)...

25 replies · 2 participants

Rafał Wojdyła

02/10/2021, 5:16 PM

👋 I have a question about Result/Target usage, I would like the target template to be based on the flow parameters - how would I do that? ``````

18 replies · 2 participants

Alex Furrier

02/10/2021, 5:19 PM

Is there anything akin to a lambda/anonymous task in Prefect? E.g. I have an iterable and want to transform it into a dictionary. I could define a task to do so but in this case something that simple could be done with an anonymous function

2 replies · 2 participants

Mitchell Bregman

02/10/2021, 7:31 PM

Is there a way to have 2 different schedule patterns attached to a single flow. For example, I want this flow to run on a 5 minute interval during business hours only, and on an hourly interval off business hours.

👍 1

13 replies · 3 participants

Verun Rahimtoola

02/10/2021, 7:40 PM

hi, if i have a task that can take in a very large number of parameters, is it possible to pass them all in as a collection to the task? eg:

with Flow(...) as f:
   param_1 = ...
   param_2 = ...
   ...
   param_n = ...
   x = my_task([param_1, param_2, ..., param_n])

4 replies · 2 participants

Verun Rahimtoola

02/10/2021, 7:42 PM

and then have prefect detect those params as upstream of the task?

Vincent

02/10/2021, 7:51 PM

Hi - I was wondering if anyone has a workaround this issue? https://github.com/PrefectHQ/prefect/issues/4085

2 replies · 2 participants

Jonathan Lowery

02/10/2021, 8:16 PM

Is there a (locking?) mechanism to make mapped tasks mutually exclusive at runtime? Use case is pulling data for an ETL from multiple sources and wanting to limit one task at a time attempting to insert to a target table. When the tasks run over each other trying to insert to the same table, it causes contention Architecture right now is: gather sources to poll for updates -> map(poll for updates) -> map(insert to db) I suppose one alternate approach would be feeding all the gathered tasks through a grouping task that sets them as upstream/downstream from one another based on target table?

6 replies · 4 participants

James Phoenix

02/10/2021, 9:33 PM

Hey everyone!

James Phoenix

02/10/2021, 9:33 PM

I'm trying to use a prefect.client() to contact my prefect cloud in flask.

James Phoenix

02/10/2021, 9:34 PM

from clean_energy.client_wrappers.prefect_flows.flows import test #THIS IS A FLOW
from flask import (Blueprint)

prefect_client_routes = Blueprint('prefect_client_routes', __name__, url_prefix="/api/prefect")

import prefect
from config.settings import Config

client = prefect.Client(api_token=Config.PREFECT_USER_TOKEN)

@prefect_client_routes.route('/')
def testing():
    return test.register(project_name='clean_energy')

James Phoenix

02/10/2021, 9:34 PM

Currently i'm getting:

James Phoenix

02/10/2021, 9:34 PM

prefect.utilities.exceptions.ClientError: [{'path': ['project'], 'message': 'field "project" not found in type: \'query_root\'', 'extensions': {'path': '$.selectionSet.project', 'code': 'validation-failed', 'exception': {'message': 'field "project" not found in type: \'query_root\''

James Phoenix

02/10/2021, 9:34 PM

I know:

James Phoenix

02/10/2021, 9:34 PM

• The project exists on Prefect Cloud

James Phoenix

02/10/2021, 9:35 PM

• My API_token is a USER token

James Phoenix

02/10/2021, 9:35 PM

What am I doing wrong?

3 replies · 2 participants

Shawn Marhanka

02/10/2021, 10:18 PM

Hi Prefect! Is there a way to create new read only (or restricted) users in Prefect Cloud? My account is saying I have reached the limit and cannot add anyone else. I have a team member that just needs to view the flow schematic, but doesn’t need access to perform any actions. Thank you

2 replies · 2 participants

Danny Vilela

02/11/2021, 3:27 AM

Hi Prefect! Quick question about flow parameters: if I wanted to know what parameters a flow “accepts”, I’m aware I can do

flow.parameters() -> Set[Parameter]

. But it seems like the implementation is “shallow,” in that it just checks for top-level tasks that are parameters (as opposed to recursively pulling parameters from tasks):

def parameters(self) -> Set[Parameter]:
    """
    Returns any parameters of the flow.

    Returns:
        - set: a set of any Parameters in this flow
    """
    return {p for p in self.tasks if isinstance(p, Parameter)}

Is there a reasonable way to extend the Flow class to search for all parameters declared within the

Flow(..) as flow

context manager?

5 replies · 2 participants

sark

02/11/2021, 7:02 AM

hello everyone, i was wondering if for a CI setup if it is good, to not only make it register flows to ensure that the deployment reflects what is in code that way, but also when a flow is removed from the repo that it gets automatically deleted at prefect as well? or does this interact badly with how prefect is designed and the intended development workflow?

3 replies · 2 participants

Karolína Bzdušek

02/11/2021, 8:44 AM

Not sure where to post this, few of my notes regarding docs • at page about Secrets, one want to click to redirect to the link about AWS/GitHub/Twitter Credentials, and got only 404 error. (https://docs.prefect.io/orchestration/concepts/secrets.html) • As I am trying to find out what GitHub token I should store (GitHub is a storage under company account in private repo), I found out that one of the option of the storage is "ref", which I somewhere (I forgot where in docs) found out is set to default "master". I think it would be nice to have this mentioned, the was one of the reasons why it couldn't find the repo (our default name for the branch is "main", doesn't occur to me that could be a problem).

4 replies · 4 participants

Laura Vaida

02/11/2021, 9:39 AM

hi folks, what ist the actual equivalent to this?: flow.environment = DaskKubernetesEnvironment()

2 replies · 2 participants

Jan Marais

02/11/2021, 9:48 AM

I see in the docs that the default message MIME type for the

EmailTask

is text/html but it seems that this is hard coded. Would it make sense to have it as a parameter to the task in the future? I just have some plain text to be sent and gmail is not parsing my newlines properly.

3 replies · 2 participants

Ruben

02/11/2021, 11:22 AM

Hi there, I was send here by support. How can I register a flow to prefect Core and then let it run once? I don’t need any scheduled reruns or anything. Just a single run on an agent polling for this flow. I know it can be done using the UI, but I’d rather do this from python.

2 replies · 2 participants

Gareth Dwyer

02/11/2021, 11:56 AM

Hey all, noob question - I’m trying to run through the hello world tutorial but using Docker on a VPS instead of locally and I get

404 Client Error for <http+docker://localhost/v1.40/containers/create>: Not Found ("No such image: prefecthq/prefect:0.14.7")

Did I do something wrong or is this related to the new 14.7 release that I see on GitHub?

4 replies · 2 participants

Adam

02/11/2021, 12:42 PM

Hi @Jim Crist-Harif and team, read some of your earlier replies, but wanted to ask whether by any chance you have had time to publish a best practice blog post or repo that shows an opinionated way of having a project that: • Uses a base docker image that contains prefect, all the custom deps and shared code • Stores the flows on Github / s3 / etc • Can organise individual flows into groups/folders locally to keep things a bit clean (and play well with pythonpath etc) and ideally have some convention in individual flow files to indicate which project they should be registered with. • Some type of CI/CD process that does the build of the image (when needed) and stores the flows • Uses Kubernetes (with or without Dask) @Chris White @Jeremiah I’ve been using Prefect for about 6 months now and I have many of the above aspects done but it’s all becoming quite a big mess. We’re looking for an opinionated / best practice way to do it now. Based on a lot of the questions above, I think a cookie cutter with sane defaults would really benefit the community. WDYT? P.S. Would be more than happy to show you the mess we have and get your advice on how things can be improved

🎯 3

➕ 2

25 replies · 9 participants

Nikul

02/11/2021, 2:16 PM

Hi All, I'm trying to run a flow using LocalDaskExecutor (using LocalDaskExecutor()), but I get an error: 'signal only works in main thread'. I was wondering if anyone knows what the problem could be. I get similar issues with DaskExecutor also. I'm running prefect==0.14.5, dask==2021.1.1, distributed==2020.12.0 Thank you

👀 1

5 replies · 2 participants

Michael Hadorn

02/11/2021, 4:26 PM

I use a docker run with gitlab storage. I got this error then:

prefect.utilities.exceptions.ClientError: 400 Client Error: Bad Request for url: <http://host.docker.internal:4200/graphql>

The following error messages were provided by the GraphQL server:

    INTERNAL_SERVER_ERROR: Variable "$input" got invalid value null at
        "input.states[0].task_run_id"; Expected non-nullable type UUID! not to be null.

The GraphQL query was:

    mutation($input: set_task_run_states_input!) {
            set_task_run_states(input: $input) {
                states {
                    status
                    message
                    id
            }
        }
    }

The passed variables were:

    {"input": {"states": [{"state": {"message": "Starting task run.", "cached_inputs": {}, "context": {"tags": []}, "_result": {"__version__": "0.14.6", "type": "NoResultType"}, "__version__": "0.14.6", "type": "Running"}, "task_run_id":
 null, "version": null}]}}

locally and also with the local agent on the ui server it's working.

12 replies · 2 participants

Title

Michael Hadorn

02/11/2021, 4:26 PM

I use a docker run with gitlab storage. I got this error then:

prefect.utilities.exceptions.ClientError: 400 Client Error: Bad Request for url: <http://host.docker.internal:4200/graphql>

The following error messages were provided by the GraphQL server:

    INTERNAL_SERVER_ERROR: Variable "$input" got invalid value null at
        "input.states[0].task_run_id"; Expected non-nullable type UUID! not to be null.

The GraphQL query was:

    mutation($input: set_task_run_states_input!) {
            set_task_run_states(input: $input) {
                states {
                    status
                    message
                    id
            }
        }
    }

The passed variables were:

    {"input": {"states": [{"state": {"message": "Starting task run.", "cached_inputs": {}, "context": {"tags": []}, "_result": {"__version__": "0.14.6", "type": "NoResultType"}, "__version__": "0.14.6", "type": "Running"}, "task_run_id":
 null, "version": null}]}}

locally and also with the local agent on the ui server it's working.

I saw, that this error also occures with others. Maybe a recreate of the tenant will help...

Michael Adkins

02/11/2021, 5:04 PM

Hi @Michael Hadorn, is this error occurring reproducibly/consistently?

Michael Hadorn

02/11/2021, 5:08 PM

yes. but only in the docker enviroment

Michael Adkins

02/11/2021, 5:09 PM

Could you provide us with the flow/code that's causing this?

Michael Hadorn

02/11/2021, 5:10 PM

and I run several times the same kind of task. but only the 4 last one crashes.

hm. maybe i should create a minimal example. it's quite big...

Michael Adkins

02/11/2021, 5:11 PM

That'd be really helpful for us 🙂

Michael Hadorn

02/11/2021, 5:13 PM

also the flow is not stopping.

Michael Adkins

02/11/2021, 5:14 PM

What docker image are you running the flow in?

Michael Hadorn

02/11/2021, 5:15 PM

prefecthq/prefect:0.14.6-python3.8

Michael Adkins

02/11/2021, 5:15 PM

👍 makes sense -- just wanted to check if there was something weird

Michael Hadorn

02/12/2021, 3:34 PM

ok, after resetting the ui (delete the volume) this error is not reappearing. i have no idea. But is solved for now for me. thx for looking at it.

✅ 1

View count: 1