https://prefect.io
Join Slack
I upgraded to My Orion instance last night to 2.0beta with the server running. It looked like everyt...
# prefect-server
d
I upgraded to My Orion instance last night to 2.0beta with the server running. It looked like everything worked from a UI perspective. However, when I run a flow they all fail with "Crash detected! Execution was interrupted by an unexpected exception. I am thinking I should probably stop all of the Prefect process and restart the Server. I don't see a command line or an option in the UI to stop or restart the Orion Server.
z
Where is your server runner?
What error are your flows reporting when crashing?
d
The Server is running locally on a MacBook Pro
z
To restart your server, you can just Ctrl-C in the terminal that you ran 
prefect orion start
from then run 
prefect orion start
again
d
I am doing some more testing and will let you know. I killed all of the prefect processes including the API, Sever and Agent. I have restarted everything. That is exactly what I did. I am still getting "Crash detected! Execution was interrupted by an unexpected exception." I am not sure what the issue is just yet. I have a Flow that has six queries to a InfluxDB.
I shut all Prefect processes down, and re-executed pip install --pre -U prefect to see if I can get my Python script to run from the command line. I am wondering if I need to reset the database.
z
I’ve just found out that including the 
--pre
is including pre-release dependencies as well and httpx happens to have a beta out that we are not compatible with
That may be the source of your errors
d
It still fails even running from the command line, each query fails with Crash detected! Execution was interrupted by an unexpected exception.
z
pip uninstall httpx
and 
pip install httpx
should get you back to a healthy state
d
Thanks let me try that
Should I try re-installing with pip install --pre -U prefect after re-installing httpx?
The script still failed to complete
I have reset the database and stopped the server and the script still fails to complete.
z
Reinstalling with 
--pre
will probably get you the pre-release httpx again which is the issue
You’re going to have to share a crash traceback for me to help more though
d
Sure, how do I create that?
z
It should be in the logs for the failing flow run
Or in the console
Did you reinstall with 
--pre
?
d
I have now ran pip install -U prefect and running the query again and it still crashes.
Copy code
% python3 InfluxDB_Avg_Query-Prefect.py
12:21:52.691 | INFO    | prefect.engine - Created flow run 'cocky-marmoset' for flow 'bin-plot-avg-main'
12:21:52.692 | INFO    | Flow run 'cocky-marmoset' - Using task runner 'ConcurrentTaskRunner'
12:21:52.863 | INFO    | Flow run 'cocky-marmoset' - Created subflow run 'black-manatee' for flow 'influx-db-queries'
Executing InfuxDB queries to generate a dataframe for each application
======================================================================

Starting CrowdSrike query...
12:21:52.907 | INFO    | Flow run 'black-manatee' - Created task run 'query_influxdb_csavg-27d58d41-0' for task 'query_influxdb_csavg'
completed, Starting nessues...
12:21:52.951 | INFO    | Flow run 'black-manatee' - Created task run 'query_influxdb_nessusdavg-8341d2a1-0' for task 'query_influxdb_nessusdavg'
completed, Starting Jamf...
12:21:52.999 | INFO    | Flow run 'black-manatee' - Created task run 'query_influxdb_jamfavg-180dbbc1-0' for task 'query_influxdb_jamfavg'
completed, Starting Java...
12:21:53.045 | INFO    | Flow run 'black-manatee' - Created task run 'query_influxdb_javaavg-3c69037c-0' for task 'query_influxdb_javaavg'
completed, Starting Tanium...
12:21:53.093 | INFO    | Flow run 'black-manatee' - Created task run 'query_influxdb_taniumavg-e4411b29-0' for task 'query_influxdb_taniumavg'
completed, Starting kerneld...
12:21:53.142 | INFO    | Flow run 'black-manatee' - Created task run 'query_influxdb_kerneldavg-779bda23-0' for task 'query_influxdb_kerneldavg'
Completed. Returning results to Flow
java query time is 2022-03-16 12:21:58.730028
tanium query time is 2022-03-16 12:23:25.526345
CrowdStrike query time is 2022-03-16 12:23:30.659676
jamf query time is 2022-03-16 12:23:31.996037
nessusd query time is 2022-03-16 12:23:35.418833
kerneld query time is 2022-03-16 12:23:38.463353
12:23:38.524 | INFO    | Task run 'query_influxdb_csavg-27d58d41-0' - Crash detected! Execution was interrupted by an unexpected exception.
12:23:38.546 | INFO    | Task run 'query_influxdb_nessusdavg-8341d2a1-0' - Crash detected! Execution was interrupted by an unexpected exception.
12:23:38.568 | INFO    | Task run 'query_influxdb_jamfavg-180dbbc1-0' - Crash detected! Execution was interrupted by an unexpected exception.
12:23:38.589 | INFO    | Task run 'query_influxdb_javaavg-3c69037c-0' - Crash detected! Execution was interrupted by an unexpected exception.
12:23:38.609 | INFO    | Task run 'query_influxdb_taniumavg-e4411b29-0' - Crash detected! Execution was interrupted by an unexpected exception.
12:23:38.632 | INFO    | Task run 'query_influxdb_kerneldavg-779bda23-0' - Crash detected! Execution was interrupted by an unexpected exception.
/Users/darren.burton/opt/anaconda3/envs/Standard/lib/python3.7/site-packages/prefect/client.py:1190: UserWarning:

No default storage has been set on the server. Using temporary local storage for results.

12:23:38.699 | ERROR   | Flow run 'black-manatee' - Finished in state Failed('6/6 states failed.')
Converting Query Results to bins and plotting in plotly - bin_plots function
12:23:38.723 | INFO    | Flow run 'cocky-marmoset' - Created task run 'bin_plot_function-906e9ce7-0' for task 'bin_plot_function'
Converting Combined Datframe Results info a Facet to plot with plotly - facet_plots function
12:23:38.813 | INFO    | Flow run 'cocky-marmoset' - Created task run 'facet_plot_function-f75bc435-0' for task 'facet_plot_function'
12:23:38.844 | ERROR   | Task run 'facet_plot_function-f75bc435-0' - Finished in state NotReady("Upstream task run 'None' did not reach a 'COMPLETED' state.", type=PENDING)
12:23:38.844 | INFO    | Task run 'bin_plot_function-906e9ce7-0' - Crash detected! Execution was interrupted by an unexpected exception.
/Users/darren.burton/opt/anaconda3/envs/Standard/lib/python3.7/site-packages/prefect/client.py:1190: UserWarning:

No default storage has been set on the server. Using temporary local storage for results.

12:23:38.914 | ERROR   | Flow run 'cocky-marmoset' - Finished in state Failed('2/3 states failed.')
z
Can you run this as 
PREFECT_DEBUG_MODE=1 python3 InfluxDB_Avg_Query-Prefect.py
d
Yes let me try that in a few min. I ended up uninstalling Prefect and using the specific version released yesterday
Copy code
pip install -U "prefect==2.0b1"
That still failed, so I am running your command for additional debug info
It doesn't like something. The Python code has not changed and has been running fine on the previous version.
z
Ah this is some silly behavior where they’ve overridden the equality operator
You can use 
prefect.utilities.collections.quote
to get around this for now
i.e. 
return quote(my_object)
from your task then where you consume it do 
my_object = result.unquote()
This will be fixed in the next release, but shame on numexpr for breaking the equality operator
d
Thank you. I will try that. Do I need to import additional Prefect libraries to use it, if so do you have a reference?
z
You can just 
from prefect.utilities.collections import quote
You can also just return your object wrapped in any other object, like 
{"value": my_dataframe}
d
Thanks @Zanie I was able to make the appropriate modifications to my Python code and get a run to complete. BTW would you happen to have a link to how to modify the code to use Prefect Cloud 2.0 beta? I wasn't finding it earlier.
please disregard, I just found it: https://orion-docs.prefect.io/ui/cloud/
z
Yep
I’d recommend creating a separate profile for cloud
d
at the Python virtual environment level?
z
Ah no we have profile support e.g. 
prefect profile create cloud
then 
prefect profile use cloud
and 
prefect cloud login …
You can easily switch back to local runs then
👍 1
d
I thought I had solved my problem, but apparently I didn't. I am still trying to figure out the proper syntax to quote and unquote the dataframe to get the flow to work. Would have a more detailed example?
z
I'd just wrap it in a list or dict
You just need to hide the dataframe return value inside another object
d
I am not familiar with that process so doing some searching
k
Could you show me the code you have Darren?
Michael is saying do something like this:
Copy code
from prefect import flow, task 
import pandas as pd
from typing import Dict

@task
def get_data():
    df = pd.DataFrame({"a": [1,2,3], "b": [2,3,4]})
    return {"df": df}

@task
def add_col(df: Dict):
    df = df["df"]
    df['c'] = 1
    return {"df": df}

@flow
def myflow():
    df = get_data()
    add_col(df)

myflow()
d
Thanks. Iet me try that. So it is at the task level where the modification is required, correct?
k
Yes
d
Still getting errors - unknown type object now
Copy code
18:12:10.714 | INFO    | prefect.engine - Created flow run 'boisterous-millipede' for flow 'myflow'
18:12:10.715 | INFO    | Flow run 'boisterous-millipede' - Using task runner 'ConcurrentTaskRunner'
18:12:10.767 | INFO    | Flow run 'boisterous-millipede' - Created task run 'query_influxdb_javamaxtest-6df64fa8-0' for task 'query_influxdb_javamaxtest'
18:12:15.673 | INFO    | Task run 'query_influxdb_javamaxtest-6df64fa8-0' - Crash detected! Execution was interrupted by an unexpected exception.
/Users/darren.burton/opt/anaconda3/envs/Standard/lib/python3.7/site-packages/prefect/client.py:1190: UserWarning:

No default storage has been set on the server. Using temporary local storage for results.

18:12:15.750 | ERROR   | Flow run 'boisterous-millipede' - Finished in state Failed('1/1 states failed.')
Out[16]:
Copy code
Failed(message='1/1 states failed.', type=FAILED, result=[Crashed(message='Execution was interrupted by an unexpected exception.', type=FAILED, result=ValueError('unknown type object'), task_run_id=b0c42fa9-2724-4353-94e0-e0458f145b2b)], flow_run_id=23cad0de-f1a0-4f87-83cf-10c062407b90)
Copy code
@task
def query_influxdb_javamaxtest():
    
#Query: using Pandas DataFrame - Java Task

    javamax_df = query_api.query_data_frame('from(bucket: "bt-test")'
      '|> range(start: -31h, stop: now())'
      '|> filter(fn: (r) => r["_measurement"] == "bt_process_cpu") '                                     
      '|> filter(fn: (r) => r["_field"] == "Java_Max")'
      '|> drop(columns: ["_measurement"])'
      '|> yield(name: "last")' )

    """
    Close client
    """
    client.close()
    
    return {"df": javamax_df}


@flow
def myflow():
    df = query_influxdb_javamaxtest()
    

myflow()
k
Ah ok we’ll need to look more into this behavior on our end. Sorry about that.
d
Thanks you so much and let me know if you come up with anything to try including upgrading to an pre-release version to test.
k
Did you try the quote like Michael suggested though?
Copy code
from lib2to3.pgen2.pgen import DFAState
from prefect import flow, task 
import pandas as pd
from typing import Dict
from prefect.utilities.collections import quote

@task
def get_data():
    df = pd.DataFrame({"a": [1,2,3], "b": [2,3,4]})
    return quote(df)

@task
def add_col(df):
    df = df.unquote()
    df['c'] = 1
    return quote(df)

@flow
def myflow():
    df = get_data()
    add_col(df)

myflow()
d
I had tried, but I will try again as I have reduced my code to test with.
It doesn't seem to work for me. I replaced the return statement and added lib2to3 import statement.
k
Ah ok yeah I think this is something we’ll have to work on next release. Michael may be able to respond with more info tomorrow
d
Thank you
z
We've already got a fix for the next release, not sure why quoting didn't work but I'd just roll back a version until next week
The release is out now
d
Thanks Michael. I rolled back today, but look forward to doing the upgrade and testing either tomorrow or over the weekend.
2 Views
Open in Slack
PreviousNext
Powered by

Prefect Community

Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.

Powered by