Pyspark根据另一列和先前数据逐步添加ID答案

【问题标题】：PySpark incrementally add id based on another column and previous dataPyspark根据另一列和先前数据逐步添加ID
【发布时间】：2023-01-24 02:26:03
【问题描述】：

从 name 列增量派生 ID，如果有新值添加到该 name 列，则在下一次加载时分配尚未分配给先前数据的需要 ID

示例 - 第一次加载：

Name
a
b
b
a

结果

ID	Name
1	a
2	b
2	b
1	a

下一个负载：

Name
a
b
b
a
c
d
c

结果：

ID	Name
1	a
2	b
2	b
1	a
3	c
4	d
3	c

如问题所述，在 PySpark 中寻找解决方案

【问题讨论】：

如果下一次加载有重复的名称怎么办，例如又是“a”、“b”？
由于 'a' 和 'b' 已经分配了 ID，因此下一次加载时它也会获得相同的 ID

标签： pyspark apache-spark-sql user-defined-functions

【解决方案1】：

您可以创建额外的数据框df_map，您可以在其中存储负载之间的 ID。如果需要，您可以从磁盘保存和恢复此数据帧。

df1 = spark.createDataFrame(
    data=[['a'], ['b'], ['b'], ['a']],
    schema=["name"]
)
df2 = spark.createDataFrame(
    data=[['a'], ['b'], ['b'], ['a'], ['c'], ['d'], ['c'], ['0']],
    schema=["name"]
)

w = Window.orderBy('name')

# create empty map
df_map = spark.createDataFrame([], schema='name string, id int')
df_map.show()

# get additional name->id map for df1
n = df_map.select(F.count('id').alias('n')).collect()[0].n
df_map = df1.subtract(df_map.select('name')).withColumn('id', F.row_number().over(w) + F.lit(n)).union(df_map)
df_map.show()

# map can be saved to disk between runs

# get additional name->id map for df2
n = df_map.select(F.count('id').alias('n')).collect()[0].n
df_map = df2.subtract(df_map.select('name')).withColumn('id', F.row_number().over(w) + F.lit(n)).union(df_map)
df_map.show()

# join to get the final dataframe
df2.join(df_map, on='name').show()

【讨论】：

该操作应该是动态的，因为有一个计划的触发器每天自动触发行插入/更新，因此寻找在那个时间点在内存上运行的逻辑来派生 Id

【解决方案2】：

您可以使用 window 和 dense_rank。下面的代码将使数据框按“名称”列排序，并为每个唯一名称提供一个递增的唯一 ID。

from pyspark.sql import functions as F
from pyspark.sql import types as T
from pyspark.sql import Window as W

window = W.orderBy('name')
(
    df
    .withColumn('id', F.dense_rank().over(window))
).show() 

+----+---+
|name| id|
+----+---+
|   a|  1|
|   a|  1|
|   b|  2|
|   b|  2|
|   c|  3|
|   c|  3|
|   d|  4|
+----+---+

【讨论】：

这将根据名称的顺序重新分配所有 ID。尝试插入“0”。