PySpark：更新给定行数的 DataFrame 的列值答案

【问题标题】：PySpark: Update column values for a given number of rows of a DataFramePySpark：更新给定行数的 DataFrame 的列值
【发布时间】：2018-07-11 23:45:19
【问题描述】：

我有一个有 10 行和 2 列的 DataFrame：一个带有随机标识符值的 ID 列和一个用 None 填充的 VAL 列。

vals = [
        Row(ID=1,VAL=None),
        Row(ID=2,VAL=None),
        Row(ID=3,VAL=None),
        Row(ID=4,VAL=None),
        Row(ID=5,VAL=None),
        Row(ID=6,VAL=None),
        Row(ID=7,VAL=None),
        Row(ID=8,VAL=None),
        Row(ID=9,VAL=None),
        Row(ID=10,VAL=None)
]
df = spark.createDataFrame(vals)

现在假设我要更新 VAL 列，其中 3 行的值为“lets”、3 行的值为“bucket”，4 行的值为“这个”。

在 PySpark 中是否有一种直接的方法？

注意：ID值不一定是连续的，桶分布不一定是均匀的

【问题讨论】：

是 - 选择哪些行获得哪些新值的标准是什么？
用户定义的标准不基于其他列的值。整个数据集被视为平等。
您必须接受近似值，或者在一般情况下不能很好地扩展的解决方案。很大程度上取决于唯一标签的数量和数据量。
更新有什么标准吗？
@user8371915 唯一标签不会很多，但行数在几百万的数量级

标签： apache-spark pyspark

【解决方案1】：

我会尝试用一些伪代码来解释一个想法，然后你会映射到你的解决方案。

在一个分区上使用窗口函数，我们可以为数据帧中的每一行生成row_number() 序列号，并将其存储在row_num 列中。接下来，您的“规则”可以表示为另一个小数据框：[min_row_num, max_row_num, label]。

您只需要在行号上加入这两个数据集，添加新列：

df1.join(df2, 
   on=col('df1.row_num').between(col('min_row_num'), col('max_row_num'))
  )
.select('df1.*', 'df2.label')

【讨论】：