在spark/scala中提供重复值唯一标识符

Question

我希望有人能知道一个简单的解决这个问题的方法，使用spark和scala。

我有一些动物运动的网络数据，其格式如下(目前为spark中的数据)：

id  start end   date
12  0     10    20091017
12  10    20    20091201
12  20    0     20091215
12  0     15    20100220
12  15    0     20100320

id是动物的id，开始和结束是运动的位置(即第二行是从位置id 10到位置id 20的移动)。如果开始或结束为0，这意味着该动物已经出生或已经死亡(即第一排动物12出生，第3行动物死亡)。

我遇到的问题是，这些数据被收集起来，以便在数据库中重新使用动物的身份，所以在动物死亡后，它的身份可能会再次发生。

我想要做的是将一个独特的标签应用到所有被重复使用的动作中。所以你会得到一个数据库

id  start end   date
12a 0     10    20091017
12a 10    20    20091201
12a 20    0     20091215
12b 0     15    20100220
12b 15    0     20100320

我尝试了几种不同的方法，但似乎找不到有效的方法。数据库非常大(几千兆字节)，因此需要一些工作效率很高的东西。

任何帮助都是非常感谢的。

Answer 1

唯一可以在DataFrames上直接工作的解决方案是使用窗口函数，但在这里我仍然不期望获得特别高的性能：

import org.apache.spark.sql.expressions.Window

val df = Seq(
  (12,  0, 10, 20091017), (12,  10, 20, 20091201),
  (12,  20, 0, 20091215), (12,  0, 15, 20100220),
  (12,  15, 0, 20100320)
).toDF("id", "start", "end", "date")

val w = Window.partitionBy($"id").orderBy($"date")
val uniqueId = struct(
  $"id", sum(when($"start" === 0, 1).otherwise(0)).over(w))

df.withColumn("unique_id", uniqueId).show

// +---+-----+---+--------+---------+
// | id|start|end|    date|unique_id|
// +---+-----+---+--------+---------+
// | 12|    0| 10|20091017|   [12,1]|
// | 12|   10| 20|20091201|   [12,1]|
// | 12|   20|  0|20091215|   [12,1]|
// | 12|    0| 15|20100220|   [12,2]|
// | 12|   15|  0|20100320|   [12,2]|
// +---+-----+---+--------+---------+