我们正在尝试编写一个scala udf函数,并从pyspark中的map函数调用它 . dateframe模式非常复杂,我们要传递给此函数的列是StructType数组 .
trip_force_speeds = trip_details.groupby("vehicle_id","driver_id", "StartDtLocal", "EndDtLocal")\ .agg(collect_list(struct(col("event_start_dt_local"), col("force"), col("speed"), col("sec_from_start"), col("sec_from_end"), col("StartDtLocal"), col("EndDtLocal"), col("verisk_vehicle_id"), col("trip_duration_sec")))\ .alias("trip_details"))
在我们的map函数中,我们需要做一些计算 .
def calculateVariables(rec: Row):HashMap[String,Float] = {
val trips = rec.getAs[List]("trips")
val base_variables = new HashMap[String, Float]()
val entropy_variables = new HashMap[String, Float]()
val week_day_list = List("monday", "tuesday", "wednesday", "thursday", "friday")
for (trip <- trips)
{
if (trip("start_dt_local") >= trip("StartDtLocal") && trip("start_dt_local") <= trip("EndDtLocal"))
{
base_variables("trip_summary_count") += 1
if (trip("duration_sec").toFloat >= 300 && trip("duration_sec").toFloat <= 1800) {
base_variables ("bounded_trip") += 1
base_variables("bounded_trip_duration") = trip("duration_sec") + base_variables("bounded_trip_duration")
base_variables("total_bin_1") += 30
base_variables("total_bin_2") += 30
base_variables("total_bin_3") += 60
base_variables("total_bin_5") += 60
base_variables("total_bin_6") += 30
base_variables("total_bin_7") += 30
}
if (trip("duration_sec") > 120 && trip("duration_sec") < 21600 )
{
base_variables("trip_count") += 1
}
base_variables("trip_distance") += trip("distance_km")
base_variables("trip_duration") = trip("duration_sec") + base_variables("trip_duration")
base_variables("speed_event_distance") = trip("speed_event_distance_km") + base_variables("speed_event_distance")
base_variables("speed_event_duration") = trip("speed_event_duration_sec") + base_variables("speed_event_duration")
base_variables("speed_event_distance_ratio") = trip("speed_distance_ratio") + base_variables("speed_event_distance_ratio")
base_variables("speed_event_duration_ratio") = trip("speed_duration_ratio") + base_variables("speed_event_duration_ratio")
}
}
return base_variables
}
当我们尝试编译scala代码时,我们得到了错误
我尝试使用Row但得到了这个错误
“错误:类型参数(List)的种类不符合预期的类型参数类型(类型T).List的类型参数与类型T的预期参数不匹配:类型List有一个类型参数,但类型T有没有 - ”
在我的情况下,旅行是一个行列表 . 这是架构
StructType(List(StructField(verisk_vehicle_id,StringType,true),StructField(verisk_driver_id,StringType,false),StructField(StartDtLocal,TimestampType,true),StructField(EndDtLocal,TimestampType,true),StructField(trips,ArrayType(StructType(List(StructField(week_start_dt_local,TimestampType,true),StructField(week_end_dt_local,TimestampType,true),StructField(start_dt_local,TimestampType,true),StructField(end_dt_local,TimestampType,true),StructField(StartDtLocal,TimestampType,true),StructField(EndDtLocal,TimestampType,true),StructField(verisk_vehicle_id,StringType,true),StructField(duration_sec,FloatType,true),StructField(distance_km,FloatType,true),StructField(speed_distance_ratio,FloatType,true),StructField(speed_duration_ratio,FloatType,true),StructField(speed_event_distance_km,FloatType,true),StructField(speed_event_duration_sec,FloatType,true))),true),true),StructField(trip_details,ArrayType(StructType(List(StructField(event_start_dt_local,TimestampType,true),StructField(force,FloatType,true),StructField(speed,FloatType,true),StructField(sec_from_start,FloatType,true),StructField(sec_from_end,FloatType,true),StructField(StartDtLocal,TimestampType,true),StructField(EndDtLocal,TimestampType,true),StructField(verisk_vehicle_id,StringType,true),StructField(trip_duration_sec,FloatType,true))),true),true)))
我们试图覆盖spark结构类型的函数签名的方式有什么不对,但这对我不起作用 .
我来自python背景,我在python作业中面临一些性能问题,这就是为什么我决定在Scala中编写这个map函数的原因 .
1 回答
您必须使用Udf中的Row类型而不是StructType . StructType表示模式本身而不是数据 . 您可以使用Scala中的一个小例子:
编辑 . 为了更好地理解:
考虑模式描述时出错:StructType(List(StructField))作为字段的类型 . 您的DataFrame中没有List类型 .
如果将calculateVariables视为udf,则不需要for循环 . 我的意思是:
正如我在示例中所示,您可以直接在udf中返回更新后的Hash