首先，加載 python 庫以使用 Pyspark 并使用 bokeh 庫：
%spark.pyspark
import bkzep
import numpy as np
from bokeh.io import output_notebook, show
from bokeh.plotting import figure
from bokeh.models import ColumnDataSource
from bokeh.layouts import gridplot
from pyspark.sql.functions import col, coalesce, lit, monotonically_increasing_id
from pyspark.sql import DataFrame
from pyspark.sql.functions import *
output_notebook(notebook_type='zeppelin')
然后準備df：
%pyspark
from pyspark.sql.functions import *
def plot_summaries(sensor, dfName):
df = DataFrame(z.get(dfName), sqlContext)
pdf = df.toPandas()
#.select("*") \
#.orderBy("index") \
#.limit(1000) \
#.withColumn("id", col("index")) \
#sample = pdf.sample(50)
source = ColumnDataSource(pdf)
#print(pdf)
TOOLTIPS = [
("month", "@month"),
("day", "@day"),
("hour", "@hour"),
("min", "@{min}{0.3f}"),
("avg", "@{avg}{0.3f}"),
("max", "@{max}{0.3f}"),
("median", "@{median}{0.3f}"),
("stddev", "@{stddev}{0.3f}"),
]
TOOLTIPS2 = [
("month", "@month"),
("day", "@day"),
("count", "@{count}{0.3f}"),
]
fig = figure(title="Hourly summaries of '{}'".format(sensor), tooltips=TOOLTIPS)
#fig.line(x='id', y='avg', source=source, color="orange")
#fig.line(x='id', y='min', source=source, color="green")
#fig.line(x='id', y='max', source=source, color="red")
fig.line(x='id', y='median', source=source, color="blue")
#fig.line(x='id', y='stddev', source=source, color="aquamarine")
#fig2 = figure(title="Hourly summaries of '{}' counters".format(sensor), tooltips=TOOLTIPS2)
#fig2.line(x='id', y='count', source=source, color="orange")
show(gridplot([fig], ncols=1, plot_width=1000, plot_height=400))
#show(fig)
sensors = [
"Water_Level_Sensor_stddev",
"Water_Level_Sensor_mean"
]
然后呼叫該函式以獲得bokeh繪圖：
%pyspark
from pyspark.sql.functions import *
keyCol = "month_day_hour"
#for sensor in sensors:
plot_summaries("Water_Level_Sensor_stddev", "pivoted")
然后得到以下例外：
---------1------------------------------------------------------------------
AttributeError Traceback (most recent call last)
<ipython-input-118-bda5385b9d44> in <module>
4
5 #for sensor in sensors:
----> 6 plot_summaries("Water_Level_Sensor_stddev", "resultIndexed")
<ipython-input-106-d6669aca8991> in plot_summaries(sensor, dfName)
3 def plot_summaries(sensor, dfName):
4 df = DataFrame(z.get(dfName), sqlContext)
----> 5 pdf = df.toPandas()
6 #.select("*") \
7 #.orderBy("index") \
/spark/python/lib/pyspark.zip/pyspark/sql/pandas/conversion.py in toPandas(self)
136
137 # Below is toPandas without Arrow optimization.
--> 138 pdf = pd.DataFrame.from_records(self.collect(), columns=self.columns)
139 column_counter = Counter(self.columns)
140
/spark/python/lib/pyspark.zip/pyspark/sql/dataframe.py in collect(self)
594 """
595 with SCCallSiteSync(self._sc) as css:
--> 596 sock_info = self._jdf.collectToPython()
597 return list(_load_from_socket(sock_info, BatchedSerializer(PickleSerializer())))
598
AttributeError: 'NoneType' object has no attribute 'collectToPython'
uj5u.com熱心網友回復：
原因是“pivoted”是DataFrame上面段落中提出的scala 。我錯了，可以在不序列化 Scala 資料幀的情況下使用它并將其作為 Pyspark 資料幀讀取。但是，該物件似乎已被拾取，但 pyspark 無法處理它的錯誤和意外型別。
轉載請註明出處，本文鏈接：https://www.uj5u.com/qiye/393591.html
標籤：Python 斯卡拉 阿帕奇火花 火花 阿帕奇齐柏林飞艇