在 Spark 中执行聚合函数时出错：ArrayType 无法转换为 org.apache.spark.sql.types.StructType

Posted 2023-03-11

【中文标题】在 Spark 中执行聚合函数时出错：ArrayType 无法转换为 org.apache.spark.sql.types.StructType

【英文标题】：Error while performing aggregate functions in Spark: ArrayType cannot be cast to org.apache.spark.sql.types.StructType

【发布时间】：2022-01-22 08:20:34

【问题描述】：

我正在从包含 gps 数据的 json 创建 Spark DF。当我尝试计算列的平均值时，出现以下错误：

Py4JJavaError: An error occurred while calling o470.collectToPython.
: java.lang.ClassCastException: org.apache.spark.sql.types.ArrayType cannot be cast to org.apache.spark.sql.types.StructType

我不明白这个错误，因为我没有 ArrayType。这是我的架构：

root
 |-- LastUpdateData: string (nullable = true)
 |-- DataGenerated: string (nullable = true)
 |-- Delay: long (nullable = true)
 |-- GPSQuality: long (nullable = true)
 |-- Lat: double (nullable = true)
 |-- Line: string (nullable = true)
 |-- Lon: double (nullable = true)
 |-- Route: string (nullable = true)
 |-- Speed: long (nullable = true)
 |-- VehicleCode: string (nullable = true)
 |-- VehicleId: long (nullable = true)
 |-- VehicleService: string (nullable = true)

StructType(List(StructField(LastUpdateData,StringType,true),StructField(DataGenerated,StringType,true),StructField(Delay,LongType,true),StructField(GPSQuality,LongType,true),StructField(Lat,DoubleType,true),StructField(Line,StringType,true),StructField(Lon,DoubleType,true),StructField(Route,StringType,true),StructField(Speed,LongType,true),StructField(VehicleCode,StringType,true),StructField(VehicleId,LongType,true),StructField(VehicleService,StringType,true)))

这是我的代码：

df.agg("Delay": "avg").collect()

【问题讨论】：

Agg 接受列或列的名称，而不是像 python 中的 dict 一样的结构。您需要做的是 df.groupby('your columns...').agg(F.avg('delay')) 。因此，例如，如果我们想要按车辆 ID 分组并获得平均速度，只需按车辆 ID 分组，按平均速度获得 agg

【参考方案1】：

试试下面的。

from pyspark.sql import functions

#returns the average value in the Delay Column
delay_df = df.agg(functions.avg("Delay"))

#view the output
delay_df.show()

【讨论】：

非常感谢！如何显示输出？当我执行此代码时，它不会给我一个错误，但它也不会显示输出。当我尝试使用 show 时，它给了我一个奇怪的输出：

更改了上面的解决方案。确保在 .show 之后添加 () 像这样delay_df.show()