Darren Burton
03/16/2022, 3:24 PMZanie
03/16/2022, 3:28 PMDarren Burton
03/16/2022, 3:49 PMZanie
03/16/2022, 3:52 PMprefect orion start
from then run prefect orion start
againDarren Burton
03/16/2022, 3:54 PMZanie
03/16/2022, 4:17 PM--pre
is including pre-release dependencies as well and httpx happens to have a beta out that we are not compatible withDarren Burton
03/16/2022, 4:17 PMZanie
03/16/2022, 4:18 PMpip uninstall httpx
and pip install httpx
should get you back to a healthy stateDarren Burton
03/16/2022, 4:18 PMZanie
03/16/2022, 4:52 PM--pre
will probably get you the pre-release httpx again which is the issueDarren Burton
03/16/2022, 5:03 PMZanie
03/16/2022, 5:07 PM--pre
?Darren Burton
03/16/2022, 5:28 PM% 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.')
Zanie
03/16/2022, 5:44 PMPREFECT_DEBUG_MODE=1 python3 InfluxDB_Avg_Query-Prefect.py
Darren Burton
03/16/2022, 5:44 PMpip install -U "prefect==2.0b1"
Zanie
03/16/2022, 6:29 PMprefect.utilities.collections.quote
to get around this for nowreturn quote(my_object)
from your task then where you consume it do my_object = result.unquote()
Darren Burton
03/16/2022, 6:40 PMZanie
03/16/2022, 6:41 PMfrom prefect.utilities.collections import quote
{"value": my_dataframe}
Darren Burton
03/16/2022, 7:47 PMZanie
03/16/2022, 7:50 PMDarren Burton
03/16/2022, 7:53 PMZanie
03/16/2022, 7:54 PMprefect profile create cloud
then prefect profile use cloud
and prefect cloud login …
Darren Burton
03/16/2022, 10:36 PMZanie
03/16/2022, 10:39 PMDarren Burton
03/16/2022, 10:42 PMKevin Kho
03/16/2022, 10:48 PMfrom 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()
Darren Burton
03/16/2022, 10:53 PMKevin Kho
03/16/2022, 10:54 PMDarren Burton
03/16/2022, 11:22 PM18: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]:
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)
@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()
Kevin Kho
03/16/2022, 11:30 PMDarren Burton
03/16/2022, 11:51 PMKevin Kho
03/16/2022, 11:53 PMfrom 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()
Darren Burton
03/16/2022, 11:55 PMKevin Kho
03/17/2022, 12:09 AMDarren Burton
03/17/2022, 12:26 AMZanie
03/17/2022, 12:36 AMDarren Burton
03/18/2022, 1:25 AM