在下面的片段中,我尝试将一个数据流(从Kafka接收)转换成一个Dataframe。
def main_process(time, dStream):
print("========= %s =========" % str(time))
try:
# Get the singleton instance of SparkSession
spark = getSparkSessionInstance(dStream.context.getConf())
# Convert RDD[String] to RDD[Row] to DataFrame
rowRdd = dStream.map(lambda t: Row(Temperatures=t))
df = spark.createDataFrame(rowRdd)
df.show()
print("The mean is: %m" % df.mean())
因此,平均值永远不会被计算,我想这是因为“df”不是一个Dataframe(?)。
我试过用 df = spark.createDataFrame(df.toPandas())
根据相关文档,但是编译器不识别“topandas()”,并且转换从未发生。
我是否在正确的道路上,如果是,我应该如何应用转换?
或者我的方法是错误的,我必须用另一种方式处理数据流?
提前谢谢!
暂无答案!
目前还没有任何答案,快来回答吧!