prefect-community #prefect-server

Docs

Join the conversation Join Slack

Channels

prefect-server

Marvin

09/12/2020, 5:54 PM

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

Sandeep Gainda

09/13/2020, 5:35 AM

@Marvin

Marvin

09/13/2020, 5:35 AM

I think you ought to know I’m feeling very depressed. Also, you're in the contest, @Sandeep Gainda.

Michael J Hall

09/13/2020, 2:20 PM

Hello. While trying to authenticate with the Prefect Cloud and configure my personal access token I am getting the following error message:

Michael J Hall

09/13/2020, 2:20 PM

[linuxserver02 ~]$ prefect auth login -t <copied token> Usage: prefect auth [OPTIONS] COMMAND [ARGS]... Error: Auth commands with server are not currently supported.

Michael J Hall

09/13/2020, 2:21 PM

I am attempting this from where Prefect server is configured, in this case our RHEL Linux server.

Michael J Hall

09/13/2020, 2:23 PM

Switching to the prefect backend cloud worked. Thanks.

Michael J Hall

09/13/2020, 3:03 PM

Walking through the tutorial on getting my first flow to run. Do I put the sample write flow code into a python script and run it using python3.6?

Michael J Hall

09/13/2020, 3:04 PM

https://docs.prefect.io/orchestration/tutorial/first.html#write-flow

Michael J Hall

09/13/2020, 4:09 PM

Getting this error while attempting to import prefect in Python3.6: "TomlDecodeError: invalid literal for int() with base 0: 'abcabcxyzxyz123123' (line 5 column 1 char 66)"

Michael J Hall

09/13/2020, 4:19 PM

I think it has something to do with the cloud agent auth token

Michael J Hall

09/13/2020, 8:31 PM

After I generated the cloud agent the second time, it appears have taken hold and I can import prefect in Python3.6. However when I run the sample code I am getting this error:

Michael J Hall

09/13/2020, 8:31 PM

[linuxserver02 ~]$ prefect backend cloud
Backend switched to cloud
[linuxserver02 ~]$ export PREFECT__CLOUD__AGENT__AUTH_TOKEN=ABC123ABC123123XYZ
[linuxserver02 ~]$ python3.6
Python 3.6.8 (default, Jun 11 2019, 15:15:01)
[GCC 4.8.5 20150623 (Red Hat 4.8.5-39)] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>> import prefect
>>> from prefect import task, Flow
>>>
>>> @task
... def hello_task():
...     logger = prefect.context.get("logger")
...     <http://logger.info|logger.info>("Hello, Cloud!")
...
>>> flow = Flow("hello-flow", tasks=[hello_task])
>>> flow.register(project_name="Hello, World!")
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/usr/local/lib/python3.6/site-packages/prefect/core/flow.py", line 1594, in register
    no_url=no_url,
  File "/usr/local/lib/python3.6/site-packages/prefect/client/client.py", line 730, in register
    project_name, project_name
ValueError: Project Hello, World! not found. Run `client.create_project("Hello, World!")` to create it.
>>>

Michael J Hall

09/13/2020, 8:32 PM

Can anyone suggest what may be happening here?

sark

09/14/2020, 3:52 AM

hi guys is my understanding correct that for

LocalStorage

it is local to the machine registering the flow, but for

LocalEnvironment

it is local to the the machine of the agent?

sark

09/14/2020, 3:56 AM

also i am registering my flows from a machine separate to the server and to whatever will run my flows, i hope this is not a anti-pattern

Hagai Arad

09/14/2020, 8:00 AM

Hi, I’m new to Prefect so maybe my question is dumb (hopes not 😅). How can I add to my flow an execution of .py files that are stored on Github or on my machine? The meaning is how to run code that is external to the flow I registered. I currently have everything running locally (I ran prefect backend server and prefect server start) and I successfully run simple flows. Thanks!

Juan

09/14/2020, 8:17 AM

Hi, my prefect server is receiving flows and the GUI shows them as scheduled although in vscode I see them as finished. Does somebody know something about this behavior ?

Rizky Eko Putra

09/14/2020, 9:11 AM

Hello, anyone know what should I do with this error

/usr/local/lib/python3.7/site-packages/prefect/core/flow.py:1594: UserWarning: No result handler was specified on your Flow. Cloud features such as input caching and resuming task runs from failure may not work properly.
  no_url=no_url,

Ian Fridge

09/15/2020, 6:45 PM

Hi. Is there a recommended git ops workflow best practice for teams which don't want developers to directly submit flows with an agent?

Charles Leung

09/15/2020, 8:59 PM

Hey prefect team, does prefect work on windows? how about local agent / docker agent?

Chandra Manginipalli

09/16/2020, 8:46 PM

Is it possible to start the prefect server at a port different from 8080? It appears the port 8080 is taken

sebastian.clanzett

09/17/2020, 4:25 PM

if i try to connect to a local prefect server with a prefect agent inside a docker container by using: prefect agent start --api http://192.168.1.1:4200/graphql, no flow gets deployed to the server. is there anything i am missing?

Nils Israel

09/17/2020, 5:09 PM

Hi, I am having a problem installing prefect server on Ubuntu 20.04. The UI can't connect to the backend. I tried the hello using curl and got a response. But the default connect to http://localhost:4200/graphql isn't working. Can someone help?

Chandra Manginipalli

09/17/2020, 6:12 PM

When I stop my prefect local server and restart, the flows I registered (and projects created) are gone. Are they not persisted in the backend like the cloud?

matt forbes

09/17/2020, 7:24 PM

has anyone had success running prefect server on kubernetes directly (rather than docker-compose)?

Chandra Manginipalli

09/17/2020, 7:46 PM

Is there a way to run the prefect agent in detached mode like this. Just so if shell is closed the agent continues to run? That way I can run the flows on a local server.

👀 1

Dave

09/18/2020, 10:39 AM

Hi, Is it correct that we aren't able to set an interval lower than one minute if you're running Prefect Core? I just tried to register a flow and got the following:

"Interval can not be less than one minute when deploying to Prefect Cloud."

I'm running the latest version of Prefect.

Vipul

09/18/2020, 11:33 AM

I am having issue while running with Flow within Flow which has Parameter that are determined at run time.

with Flow("MasterFlow") as master_flow:
    cobdate = Parameter("cobdate")
    staging_area = FlowRunTask(flow_name='staging_area',
                               parameters=dict(cobdate=cobdate),
                               project_name="SubFlow")
    business_logic_layer = FlowRunTask(flow_name='business_logic_layer',
                                       project_name="SubFlow")
    data_mart = FlowRunTask(flow_name='data_mart',
                            project_name="SubFlow",
                            wait=True)
    staging_area.set_downstream(data_mart)
    business_logic_layer.set_downstream(data_mart)

Vipul

09/18/2020, 11:35 AM

Sorry, pressed enter too early. The issue I am seeing is that • From UI, when I run MasterFlow, it does not give me an option to populate cobdate

Title

Vipul

09/18/2020, 11:35 AM

Sorry, pressed enter too early. The issue I am seeing is that • From UI, when I run MasterFlow, it does not give me an option to populate cobdate

Matt Wong-Kemp

09/18/2020, 12:59 PM

You need to pass the parameter as a runtime option, I think - at the minute you're creating a dict with one field,

cobdate

, which container the Parameter object, not the value

:upvote: 1

@task
def prepare_parameters(cobdate):
    return dict(cobdate=cobdate)

with Flow("MasterFlow") as master_flow:
    cobdate = Parameter("cobdate")
    staging_parameters = prepare_parameters(cobdate)
    staging_area = FlowRunTask(flow_name='staging_area',
                               parameters=staging_parameters,
                               project_name="SubFlow")

Vipul

09/18/2020, 1:15 PM

Thanks for quick reply.You mean I need to create a new task to pass this parameter to FlowRunTask? Just wondering why can't it directly get it from Parameter as I thought Parameter value can be updated runtime?

@Matt Wong-Kemp Also, if I make change as suggested by you, I am getting the following error: File "d:\1. vipul\1. tech\python\prefect\venv\lib\site-packages\prefect\client\client.py", line 965, in create_flow_run res = self.graphql(create_mutation, variables=dict(input=inputs)) File "d:\1. vipul\1. tech\python\prefect\venv\lib\site-packages\prefect\client\client.py", line 280, in graphql params=dict(query=parse_graphql(query), variables=json.dumps(variables)), File "C:\Program Files (x86)\Python\lib\json\__init__.py", line 231, in dumps return _default_encoder.encode(obj) File "C:\Program Files (x86)\Python\lib\json\encoder.py", line 199, in encode chunks = self.iterencode(o, _one_shot=True) File "C:\Program Files (x86)\Python\lib\json\encoder.py", line 257, in iterencode return _iterencode(o, 0) File "C:\Program Files (x86)\Python\lib\json\encoder.py", line 179, in default raise TypeError(f'Object of type {o.class.name} ' TypeError: Object of type FunctionTask is not JSON serializable

-> res = self.graphql(create_mutation, variables=dict(input=inputs))

(Pdb) inputs

{'flow_id': 'bc156d0a-b2dc-4425-85a6-10e2a9a5ef1f', 'parameters': <Task: prepare_parameters>, 'idempotency_key': 'db6be705-45eb-4a99-adea-86c4506227d9'}

I think the JSON error is due to fact that 'parameters' has value of <Task: prepare_parameters> and JSON does not know how to serialiaze it... @nicholas

nicholas

09/18/2020, 5:22 PM

@Vipul - when running the flow from the UI, you'll need to navigate to the

Run

tab here:

And then in your original flow you can make

cobdate

required or give it a default:

with Flow("MasterFlow") as master_flow:
    cobdate = Parameter("cobdate", required=True)
    # OR      Parameter("cobdate", default="some default")

    # Then pass the parameters to the runtime context of FlowRunTask
    # instead of the Class instantiation
    staging_area = FlowRunTask(flow_name='staging_area',
                               project_name="SubFlow")(parameters={"cobdate": cobdate})

    business_logic_layer = FlowRunTask(flow_name='business_logic_layer',
                                       project_name="SubFlow")
    data_mart = FlowRunTask(flow_name='data_mart',
                            project_name="SubFlow",
                            wait=True)
    staging_area.set_downstream(data_mart)
    business_logic_layer.set_downstream(data_mart)

Vipul

09/18/2020, 5:59 PM

Wonderful @nicholas that worked, thanks for your help and @Matt Wong-Kemp

🚀 2

nicholas

09/18/2020, 6:02 PM

Great !

Vipul

09/27/2020, 10:49 AM

Hey @nicholas it worked based on your approach but what I have seen from the schematic is that adding the line

staging_area = FlowRunTask(flow_name='staging_area',
                               project_name="SubFlow")(parameters={"cobdate": cobdate})

has cause two extra tasks to be added as List and Dict before calling Flow "staging_area", is this expected behaviour? Thanks

View count: 2