scala - scala spark UDF过滤器数组的结构
问题描述
我有一个带有架构的数据框
root
|-- x: Long (nullable = false)
|-- y: Long (nullable = false)
|-- features: array (nullable = true)
| |-- element: struct (containsNull = true)
| | |-- name: string (nullable = true)
| | |-- score: double (nullable = true)
例如,我有数据
+--------------------+--------------------+------------------------------------------+
| x | y | features |
+--------------------+--------------------+------------------------------------------+
|10 | 9 |[["f1", 5.9], ["ft2", 6.0], ["ft3", 10.9]]|
|11 | 0 |[["f4", 0.9], ["ft1", 4.0], ["ft2", 0.9] ]|
|20 | 9 |[["f5", 5.9], ["ft2", 6.4], ["ft3", 1.9] ]|
|18 | 8 |[["f1", 5.9], ["ft4", 8.1], ["ft2", 18.9]]|
+--------------------+--------------------+------------------------------------------+
我想过滤带有特定前缀的功能,比如“ft”,所以最终我想要结果:
+--------------------+--------------------+-----------------------------+
| x | y | features |
+--------------------+--------------------+-----------------------------+
|10 | 9 |[["ft2", 6.0], ["ft3", 10.9]]|
|11 | 0 |[["ft1", 4.0], ["ft2", 0.9] ]|
|20 | 9 |[["ft2", 6.4], ["ft3", 1.9] ]|
|18 | 8 |[["ft4", 8.1], ["ft2", 18.9]]|
+--------------------+--------------------+-----------------------------+
我没有使用 Spark 2.4+,所以我不能使用这里提供的解决方案:Spark (Scala) filter array of structs without explode
我尝试使用 UDF,但仍然无法正常工作。这是我的尝试。我定义了一个 UDF:
def filterFeature: UserDefinedFunction =
udf((features: Seq[Row]) =>
features.filter{
x.getString(0).startsWith("ft")
}
)
但是如果我应用这个 UDF
df.withColumn("filtered", filterFeature($"features"))
我得到错误Schema for type org.apache.spark.sql.Row is not supported
。我发现我无法Row
从 UDF 返回。然后我尝试了
def filterFeature: UserDefinedFunction =
udf((features: Seq[Row]) =>
features.filter{
x.getString(0).startsWith("ft")
}, (StringType, DoubleType)
)
然后我得到一个错误:
error: type mismatch;
found : (org.apache.spark.sql.types.StringType.type, org.apache.spark.sql.types.DoubleType.type)
required: org.apache.spark.sql.types.DataType
}, (StringType, DoubleType)
^
我还按照一些答案的建议尝试了一个案例类:
case class FilteredFeature(featureName: String, featureScore: Double)
def filterFeature: UserDefinedFunction =
udf((features: Seq[Row]) =>
features.filter{
x.getString(0).startsWith("ft")
}, FilteredFeature
)
但我得到了:
error: type mismatch;
found : FilteredFeature.type
required: org.apache.spark.sql.types.DataType
}, FilteredFeature
^
我试过了:
case class FilteredFeature(featureName: String, featureScore: Double)
def filterFeature: UserDefinedFunction =
udf((features: Seq[Row]) =>
features.filter{
x.getString(0).startsWith("ft")
}, Seq[FilteredFeature]
)
我有:
<console>:192: error: missing argument list for method apply in class GenericCompanion
Unapplied methods are only converted to functions when a function type is expected.
You can make this conversion explicit by writing `apply _` or `apply(_)` instead of `apply`.
}, Seq[FilteredFeature]
^
我试过了:
case class FilteredFeature(featureName: String, featureScore: Double)
def filterFeature: UserDefinedFunction =
udf((features: Seq[Row]) =>
features.filter{
x.getString(0).startsWith("ft")
}, Seq[FilteredFeature](_)
)
我有:
<console>:201: error: type mismatch;
found : Seq[FilteredFeature]
required: FilteredFeature
}, Seq[FilteredFeature](_)
^
在这种情况下我该怎么办?
解决方案
你有两个选项:
a) 为 UDF 提供一个模式,这让你返回Seq[Row]
b) 转换Seq[Row]
为Seq
ofTuple2
或 case 类,则无需提供模式(但如果使用元组,结构字段名称会丢失!)
对于您的情况,我更喜欢选项 a)(适用于具有许多字段的结构):
val schema = df.schema("features").dataType
val filterFeature = udf((features:Seq[Row]) => features.filter(_.getAs[String]("name").startsWith("ft")),schema)
推荐阅读
- php - 如何使用 css DomPDF 从 page.html 导出 PDF
- file - 注册数据写入文件的时间
- java - 如果未选择任何选项,则以角度禁用 nb-stepper 按钮
- html - 在 Dash 中将破折号组件向右移动
- sql-server - 努力将动态 SQL 语句链接在一起而没有错误
- rust - Rust crates 的发行说明在哪里发布
- kubernetes - 终止宽限期秒未按预期工作
- elasticsearch - 将搜索短语匹配到布尔字段值
- sql - 在 BigQuery 标准 SQL 中,如何获取行子集的中位数
- java - hive 2.3.9 无法使用 JDBC Uri 打开客户端传输:jdbc:hive2://localhost:10000/default: java.net.ConnectException: Connection denied