apache-spark - 如何将正在运行的 Id 新列添加到 Spark Dataframe ( pyspark)

Question

我有以下数据框:

timestamp	      sum
31/01/2017 09:00	0
31/01/2017 10:00	0
31/01/2017 11:00	0
31/01/2017 12:00	2
31/01/2017 13:00	2
31/01/2017 14:00	2
31/01/2017 15:00	11

并且想添加一个新的 Id 列 - 只是一个像这样的连续数字:

+----------------+---+---------+
|       timestamp|sum|running_id|
+----------------+---+---------+
|2017-01-31 09:00|  0|        0|
|2017-01-31 10:00|  0|        1|
|2017-01-31 11:00|  0|        2|
|2017-01-31 12:00|  2|        3|
|2017-01-31 13:00|  2|        4|
|2017-01-31 14:00|  2|        5|
|2017-01-31 15:00| 11|        6|

我是这样做的:

sub_data_spark =  sub_data_spark.rdd.zipWithIndex().map(lambda x: (x[0][0],x[0][1],x[1])).toDF(sub_data_spark.columns+["running_id"])

有人可以建议一种“更干净”的方式吗？

谢谢，鲍里斯

Answer 1

没有 zipWithIndex 或 zipWithUniqueId 的唯一方法，您应该使用函数 monotonically_increasing_id

这个函数的工作原理如下:

A column that generates monotonically increasing 64-bit integers.

The generated ID is guaranteed to be monotonically increasing and unique, but not consecutive. The current implementation puts the partition ID in the upper 31 bits, and the record number within each partition in the lower 33 bits. The assumption is that the data frame has less than 1 billion partitions, and each partition has less than 8 billion records.

因此，对于您的情况，您可以这样使用:

sub_data_spark.withColumn('Id', monotonically_increasing_id()).show()

这将返回给您模型的唯一 ID。但不会从0开始，也不会是顺序的