scala - Spark(Scala)过滤结构数组而不爆炸
问题描述
我有一个带有键的数据框和一个带有数据框列中的结构数组的列。每行包含一列,如下所示:
[
{"id" : 1, "someProperty" : "xxx", "someOtherProperty" : "1", "propertyToFilterOn" : 1},
{"id" : 2, "someProperty" : "yyy", "someOtherProperty" : "223", "propertyToFilterOn" : 0},
{"id" : 3, "someProperty" : "zzz", "someOtherProperty" : "345", "propertyToFilterOn" : 1}
]
现在我想做两件事:
- 过滤“propertyToFilterOn”= 1
- 对其他属性应用一些逻辑 - 例如连接
所以结果是:
[
{"id" : 1, "newProperty" : "xxx_1"},
{"id" : 3, "newProperty" : "zzz_345"}
]
我知道如何使用爆炸来做到这一点,但是在将其重新组合在一起时,爆炸还需要按键上的 groupBy。但由于这是一个流数据帧,我还必须在其上添加水印,这是我试图避免的。
有没有其他方法可以在不使用爆炸的情况下实现这一目标?我确信有一些 Scala 魔法可以实现这一点!
谢谢!
解决方案
随着 spark 2.4+ 出现了许多用于数组的高阶函数。(见https://docs.databricks.com/spark/2.x/spark-sql/language-manual/functions.html)
val dataframe = Seq(
("a", 1, "xxx", "1", 1),
("a", 2, "yyy", "223", 0),
("a", 3, "zzz", "345", 1)
).toDF( "grouping_key", "id" , "someProperty" , "someOtherProperty", "propertyToFilterOn" )
.groupBy("grouping_key")
.agg(collect_list(struct("id" , "someProperty" , "someOtherProperty", "propertyToFilterOn")).as("your_array"))
dataframe.select("your_array").show(false)
+----------------------------------------------------+
|your_array |
+----------------------------------------------------+
|[[1, xxx, 1, 1], [2, yyy, 223, 0], [3, zzz, 345, 1]]|
+----------------------------------------------------+
您可以使用数组过滤器高阶函数过滤数组中的元素,如下所示:
val filteredDataframe = dataframe.select(expr("filter(your_array, your_struct -> your_struct.propertyToFilterOn == 1)").as("filtered_arrays"))
filteredDataframe.show(false)
+----------------------------------+
|filtered_arrays |
+----------------------------------+
|[[1, xxx, 1, 1], [3, zzz, 345, 1]]|
+----------------------------------+
对于您谈论的“其他逻辑”,您应该能够像这样使用转换高阶数组函数:
val tranformedDataframe = filteredDataframe
.select(expr("transform(filtered_arrays, your_struct -> struct(concat(your_struct.someProperty, '_', your_struct.someOtherProperty))"))
但是如本文所述,从转换函数返回结构存在问题:
所以你最好使用 dataset api 进行转换,如下所示:
case class YourStruct(id:String, someProperty: String, someOtherProperty: String)
case class YourArray(filtered_arrays: Seq[YourStruct])
case class YourNewStruct(id:String, newProperty: String)
val transformedDataset = filteredDataframe.as[YourArray].map(_.filtered_arrays.map(ys => YourNewStruct(ys.id, ys.someProperty + "_" + ys.someOtherProperty)))
val transformedDataset.show(false)
+--------------------------+
|value |
+--------------------------+
|[[1, xxx_1], [3, zzz_345]]|
+--------------------------+
推荐阅读
- user-interface - Xamarin Forms Frame 不随内容扩展
- android - 应用程序未在通知访问设置 Android 中列出
- ios - 未使用 AVAssetWrite 录制音频
- react-native - react-native 搜索栏和 flatList 问题
- javascript - 悬停图像大小问题
- cordova - 离子 2 SSl 钉扎
- autohotkey - 在 AutoHotKey 中重用 GUI 输入
- mysql - 将数据从 aws s3 导入 mysql 或任何非 aws 数据库
- c# - 我应该选择哪种 C# 项目类型?需要 UWP 设计功能,但需要 WPF 功能
- c++ - 用户输入大数字时的无限循环