在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出生，第三排动物已经死亡）。

我遇到的问题是，收集数据是为了在数据库中重新使用动物ID，因此在动物死亡后，其ID可能会再次出现。

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

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

我一直在尝试几种不同的方法，但似乎都没有成功。数据库非常大（几GB），所以需要一些工作效率很高的东西。

非常感谢您的帮助。

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]|
// +---+-----+---+--------+---------+