pyspark - 在 Spark 会话中获得一致的随机值

我想将一列随机值添加到数据帧（每行都有一个 id）对于我正在测试的东西。我是努力在 Spark 会话中获得可重复的结果 - 每个行 id 的随机值相同。我能够通过使用重现结果

from pyspark.sql.functions import rand

new_df = my_df.withColumn("rand_index", rand(seed = 7))

但只有当我在同一个 Spark 会话中运行它时它才有效。重新启动 Spark 并运行脚本后，我没有得到相同的结果。

我还尝试定义一个 udf，测试是否可以在一个间隔内生成随机值（整数）并使用random从Python与随机种子 set

import random
random.seed(7)
spark.udf.register("getRandVals", lambda x, y: random.randint(x, y), LongType())

但无济于事。

有没有办法确保 Spark 会话中可重复生成随机数使得行 id 获得相同的随机值？我真的很感激一些指导:) 谢谢您的帮助！

我怀疑您获得的种子的通用值相同，但根据分区的顺序不同，分区的顺序受从磁盘读取时数据分布的影响，并且每次可能有更多或更少的数据。但实际上我并不了解你的代码。

rand 函数生成相同的随机数据（否则种子的意义是什么），并且分区以某种方式获得它的一部分。如果你看的话你应该猜出图案！

这是 2 个不同基数数据帧的示例。您可以看到种子给出了相同或超集的结果。因此，在我看来，排序和分区很重要。

from pyspark.sql.functions import *
from pyspark.sql.types import *
from pyspark.sql.functions import col
df1 = spark.range(1, 5).select(col("id").cast("double"))
df1 = df1.withColumn("rand_index", rand(seed = 7))                                   
df1.show()

df1.rdd.getNumPartitions()
print('Partitioning distribution: '+ str(df1.rdd.glom().map(len).collect()))

returns:

+---+-------------------+
| id|         rand_index|
+---+-------------------+
|1.0|0.06498948189958098|
|2.0|0.41371264720975787|
|3.0|0.12030715258495939|
|4.0| 0.2731073068483362|
+---+-------------------+

8 partitions & Partitioning distribution: [0, 1, 0, 1, 0, 1, 0, 1]

更多数据也是如此：

...
df1 = spark.range(1, 10).select(col("id").cast("double"))
...

returns:

+---+-------------------+
| id|         rand_index|
+---+-------------------+
|1.0| 0.9147159860432812|
|2.0|0.06498948189958098|
|3.0| 0.7069655052310547|
|4.0|0.41371264720975787|
|5.0| 0.1982919638208397|
|6.0|0.12030715258495939|
|7.0|0.44292918521277047|
|8.0| 0.2731073068483362|
|9.0| 0.7784518091224375|
+---+-------------------+

8 partitions & Partitioning distribution: [1, 1, 1, 1, 1, 1, 1, 2]

您可以在 Spark 会话内或会话外看到 4 个常见的随机值。