如何在组中找到第一个非空值? (使用数据集api进行二次排序)
How to find first non-null values in groups? (secondary sorting using dataset api)
我正在处理一个表示事件流的数据集(比如作为跟踪事件从网站触发)。所有事件都有时间戳。我们经常遇到的一个用例是尝试为给定字段找到第一个非空值。因此,例如,像这样的东西最能帮助我们到达那里:
val eventsDf = spark.read.json(jsonEventsPath)
case class ProjectedFields(visitId: String, userId: Int, timestamp: Long ... )
val projectedEventsDs = eventsDf.select(
eventsDf("message.visit.id").alias("visitId"),
eventsDf("message.property.user_id").alias("userId"),
eventsDf("message.property.timestamp"),
...
).as[ProjectedFields]
projectedEventsDs.groupBy($"visitId").agg(first($"userId", true))
上述代码的问题在于无法保证输入 first
聚合函数的数据顺序。我希望它按 timestamp
排序,以确保它是按时间戳记的第一个非空 userId,而不是任何随机的非空 userId。
有没有办法定义分组内的排序?
使用 Spark 2.10
顺便说一句, 中为 Spark 2.10 建议的方法是在分组之前进行排序——这是行不通的。例如下面的代码:
case class OrderedKeyValue(key: String, value: String, ordering: Int)
val ds = Seq(
OrderedKeyValue("a", null, 1),
OrderedKeyValue("a", null, 2),
OrderedKeyValue("a", "x", 3),
OrderedKeyValue("a", "y", 4),
OrderedKeyValue("a", null, 5)
).toDS()
ds.orderBy("ordering").groupBy("key").agg(first("value", true)).collect()
有时会 return Array([a,y])
有时 Array([a,x])
使用我的挚爱windows(...体验一下您的生活变得多么简单!)
import org.apache.spark.sql.expressions.Window
val byKeyOrderByOrdering = Window
.partitionBy("key")
.orderBy("ordering")
.rangeBetween(Window.unboundedPreceding, Window.unboundedFollowing)
import org.apache.spark.sql.functions.first
val firsts = ds.withColumn("first",
first("value", ignoreNulls = true) over byKeyOrderByOrdering)
scala> firsts.show
+---+-----+--------+-----+
|key|value|ordering|first|
+---+-----+--------+-----+
| a| null| 1| x|
| a| null| 2| x|
| a| x| 3| x|
| a| y| 4| x|
| a| null| 5| x|
+---+-----+--------+-----+
注意:不知何故,Spark 2.2.0-SNAPSHOT(今天构建)无法给我正确的答案,没有 rangeBetween
我认为应该是默认的无限范围。
我正在处理一个表示事件流的数据集(比如作为跟踪事件从网站触发)。所有事件都有时间戳。我们经常遇到的一个用例是尝试为给定字段找到第一个非空值。因此,例如,像这样的东西最能帮助我们到达那里:
val eventsDf = spark.read.json(jsonEventsPath)
case class ProjectedFields(visitId: String, userId: Int, timestamp: Long ... )
val projectedEventsDs = eventsDf.select(
eventsDf("message.visit.id").alias("visitId"),
eventsDf("message.property.user_id").alias("userId"),
eventsDf("message.property.timestamp"),
...
).as[ProjectedFields]
projectedEventsDs.groupBy($"visitId").agg(first($"userId", true))
上述代码的问题在于无法保证输入 first
聚合函数的数据顺序。我希望它按 timestamp
排序,以确保它是按时间戳记的第一个非空 userId,而不是任何随机的非空 userId。
有没有办法定义分组内的排序?
使用 Spark 2.10
顺便说一句,
case class OrderedKeyValue(key: String, value: String, ordering: Int)
val ds = Seq(
OrderedKeyValue("a", null, 1),
OrderedKeyValue("a", null, 2),
OrderedKeyValue("a", "x", 3),
OrderedKeyValue("a", "y", 4),
OrderedKeyValue("a", null, 5)
).toDS()
ds.orderBy("ordering").groupBy("key").agg(first("value", true)).collect()
有时会 return Array([a,y])
有时 Array([a,x])
使用我的挚爱windows(...体验一下您的生活变得多么简单!)
import org.apache.spark.sql.expressions.Window
val byKeyOrderByOrdering = Window
.partitionBy("key")
.orderBy("ordering")
.rangeBetween(Window.unboundedPreceding, Window.unboundedFollowing)
import org.apache.spark.sql.functions.first
val firsts = ds.withColumn("first",
first("value", ignoreNulls = true) over byKeyOrderByOrdering)
scala> firsts.show
+---+-----+--------+-----+
|key|value|ordering|first|
+---+-----+--------+-----+
| a| null| 1| x|
| a| null| 2| x|
| a| x| 3| x|
| a| y| 4| x|
| a| null| 5| x|
+---+-----+--------+-----+
注意:不知何故,Spark 2.2.0-SNAPSHOT(今天构建)无法给我正确的答案,没有 rangeBetween
我认为应该是默认的无限范围。