本文向您展示如何使用Apache火花函数来生成惟一增加一列的数值。

我们审查三个不同的方法使用。你应该选择最有效的方法与你的用例。

使用zipWithIndex ()在弹性分布式数据集(抽样)

的zipWithIndex ()函数只能在抽样。你不能直接使用它DataFrame。

转换您的DataFrame抽样,适用zipWithIndex ()你的数据,然后将抽样回DataFrame。

我们将使用以下示例代码添加惟一的id数字基本表有两个条目。

% python df =火花。createDataFrame([(“爱丽丝”,“10”)(“苏珊”,“12”)],[“名称”,“年龄”])df1 = df.rdd.zipWithIndex () .toDF () df2 = df1.select(坳(“_1。*”)(“_2”).alias上校(increasing_id)) df2.show ()

运行示例代码,我们得到以下结果:

+ - - - + - - - + - - - - - - - - - - - - - | + |名字年龄| increasing_id | +——+ - - - + - - - - - - - - - - - - -爱丽丝+ | | 10 | 0 | |苏珊| 12 | 1 | +——+ - - - + - - - - - - - - - - - - - +

使用monotonically_increasing_id ()独特的,但不是连续的数字

的monotonically_increasing_id ()生成函数单调递增64位整数。

生成的id数字是保证增加和独特的,但是这不能保证连续。

我们将使用以下示例代码将单调递增数字id添加到一个基本表与两个条目。

%从pyspark.sql python。功能导入* df_with_increasing_id = df。与Column("monotonically_increasing_id", monotonically_increasing_id()) df_with_increasing_id.show()

运行示例代码,我们得到以下结果:

+ - - - + - - - + - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - | + |名字年龄| monotonically_increasing_id | +——+ - - - + - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - 10爱丽丝+ | | | 8589934592 | |苏珊12 | 25769803776 | | +——+ - - - + - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - +

结合monotonically_increasing_id ()与row_number ()两列

的row_number ()函数生成连续的数字。

结合monotonically_increasing_id ()生成两列数据,可用于识别数据条目。

我们将使用以下示例代码添加单调递增id号和行号与两个条目一个基本表。

%从pyspark.sql python。函数从pyspark.sql进口*。窗口导入*窗口= Window.orderBy(坳(monotonically_increasing_id)) df_with_consecutive_increasing_id = df_with_increasing_id。withColumn (increasing_id, row_number () .over(窗口))df_with_consecutive_increasing_id.show ()

运行示例代码,我们得到以下结果:

+ - - - + - - - + - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - + - - - - - - - - - - - - - | + |名字年龄| monotonically_increasing_id | increasing_id | +——+ - - - + - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - + - - - - - - - - - - - - -爱丽丝+ | | 10 | 8589934592 | 1 | |苏珊| 12 | 25769803776 | 2 | +——+ - - - + - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - + - - - - - - - - - - - - - +

如果你需要增加基于上次更新最大值,您可以定义一个先前的最大值,然后从那里开始计数。

我们要构建的示例代码,我们就跑。

首先,我们需要定义的值previous_max_value。为此,您通常会从现有的输出表获取价值。对于这个示例,我们将它定义为1000。

python previous_max_value = 1000 df_with_consecutive_increasing_id %。与Column("cnsecutiv_increase", col("increasing_id") + lit(previous_max_value)).show()

当这是结合前面的示例代码和运行,我们得到以下结果:

+——+ - - - + - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - + - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - | + |名字年龄| monotonically_increasing_id | increasing_id | cnsecutiv_increase | +——+ - - - + - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - + - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -爱丽丝+ | | 1001 | | 8589934592 | 1 | |苏珊| 1002 | 25769803776 | 2 | | +——+ - - - + - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - + - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - +