使用monotonically_increasing_id()将行号分配给pyspark数据帧

问题描述 投票:12回答:3

我使用monotonically_increasing_id()使用以下语法将行号分配给pyspark数据帧:

df1 = df1.withColumn("idx", monotonically_increasing_id())

现在df1有26,572,528条记录。所以我期待idx值从0-26,572,527。

但是当我选择max(idx)时,它的值非常大:335,008,054,165。

这个功能发生了什么?使用此函数与具有相似记录数的其他数据集合并是否可靠?

我有大约300个数据帧,我想将它们组合成一个数据帧。因此,一个数据帧包含ID,而其他数据帧包含与行对应的不同记录

indexing merge pyspark
3个回答
20
投票

来自documentation

生成单调递增的64位整数的列。

生成的ID保证单调增加且唯一,但不是连续的。当前实现将分区ID放在高31位中,将每个分区中的记录号放在低33位中。假设数据框的分区少于10亿,每个分区的记录少于80亿。

因此,它不像RDB中的自动增量ID,并且它不可靠用于合并。

如果您需要像RDB中那样的自动增量行为并且您的数据是可排序的,那么您可以使用row_number

df.createOrReplaceTempView('df')
spark.sql('select row_number() over (order by "some_column") as num, * from df')
+---+-----------+
|num|some_column|
+---+-----------+
|  1|   ....... |
|  2|   ....... |
|  3| ..........|
+---+-----------+

如果您的数据不可排序,并且您不介意使用rdds创建索引然后回退到数据框架,则可以使用rdd.zipWithIndex()

一个例子可以找到here

简而言之:

# since you have a dataframe, use the rdd interface to create indexes with zipWithIndex()
df = df.rdd.zipWithIndex()
# return back to dataframe
df = df.toDF()

df.show()

# your data           | indexes
+---------------------+---+
|         _1          | _2| 
+-----------=---------+---+
|[data col1,data col2]|  0|
|[data col1,data col2]|  1|
|[data col1,data col2]|  2|
+---------------------+---+

在此之后,您可能需要进行一些更改,以使您的数据框符合您的需要。注意:不是一个非常高效的解决方案。

希望这可以帮助。祝好运!

编辑:来考虑一下,你可以结合使用monotonically_increasing_id来使用row_number

# create a monotonically increasing id 
df = df.withColumn("idx", monotonically_increasing_id())

# then since the id is increasing but not consecutive, it means you can sort by it, so you can use the `row_number`
df.createOrReplaceTempView('df')
new_df = spark.sql('select row_number() over (order by "idx") as num, * from df')

虽然不确定性能。


19
投票

使用api函数,您可以简单地执行以下操作

from pyspark.sql.window import Window as W
from pyspark.sql import functions as F
df1 = df1.withColumn("idx", F.monotonically_increasing_id())
windowSpec = W.orderBy("idx")
df1.withColumn("idx", F.row_number().over(windowSpec)).show()

我希望答案是有帮助的


2
投票

我发现@mkaran的解决方案很有用,但对我来说,在使用窗口函数时没有排序列。我想维护数据帧行的顺序作为它们的索引(你会在pandas数据帧中看到)。因此编辑部分的解决方案很有用。由于这是一个很好的解决方案(如果性能不是一个问题),我想作为一个单独的答案分享它。

# Add a increasing data column 
df_index = df.withColumn("idx", monotonically_increasing_id())

# Create the window specification
w = Window.orderBy("idx")

# Use row number with the window specification
df_index = df_index.withColumn("index", F.row_number().over(w))

# Drop the created increasing data column
df2_index = df2_index.drop("idx")

df是您的原始数据框架,df_index是新的数据框架。

© www.soinside.com 2019 - 2024. All rights reserved.