我试图在scala中编写一个udf函数,并在我的pyspark工作中使用它。我的数据帧架构是
root
|-- vehicle_id: string
|-- driver_id: string
|-- StartDtLocal: timestamp
|-- EndDtLocal: timestamp
|-- trips: array
| |-- element: struct
| | |-- week_start_dt_local: timestamp
| | |-- week_end_dt_local: timestamp
| | |-- start_dt_local: timestamp
| | |-- end_dt_local: timestamp
| | |-- StartDtLocal: timestamp
| | |-- EndDtLocal: timestamp
| | |-- vehicle_id: string
| | |-- duration_sec: float
| | |-- distance_km: float
| | |-- speed_distance_ratio: float
| | |-- speed_duration_ratio: float
| | |-- speed_event_distance_km: float
| | |-- speed_event_duration_sec: float
|-- trip_details: array
| |-- element: struct
| | |-- event_start_dt_local: timestamp
| | |-- force: float
| | |-- speed: float
| | |-- sec_from_start: float
| | |-- sec_from_end: float
| | |-- StartDtLocal: timestamp
| | |-- EndDtLocal: timestamp
| | |-- vehicle_id: string
| | |-- trip_duration_sec: float
我正在尝试编写一个udf函数
def calculateVariables(row: Row):HashMap[String, Float] = {
case class myRow(week_start_dt_local: Timestamp, week_end_dt_local: Timestamp, start_dt_local: Timestamp, end_dt_local :Timestamp, StartDtLocal:Timestamp,EndDtLocal:Timestamp,vehicle_id:String,duration_sec:Int,distance_km:Int,speed_distance_ratio:Float,speed_duration_ratio:Float,speed_event_distance_km:Float,speed_event_duration_sec:Float)
val trips = row.getAs[WrappedArray[myRow]](4)
在这个map函数中我试图将行转换为case类但不能。我得到这个错误。
java.lang.ClassCastException:org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema无法强制转换为VariableCalculation.VariableCalculation $ myRow $ 3
问题是,.as在Row不只是一个类型转换而已。内部类型trips实际上是Row
所以row.getAs[WrappedArray[Row]]("trips")会工作。然后你可以map覆盖它并myRow从中构建Row。
您可以使用Sparks以某种方式自动执行此操作,Encoder但它们更适合应用于整个数据集。
您是否考虑过为整个模式制定案例类,然后才这样做dataframe.as[MyCaseClass]?这将使您可以正确访问整个嵌套结构
版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。