pyspark 搜索数据框并随机选择要添加到新数据框的值答案

【问题标题】：pyspark search dataframe and randomly select value to add to new dataframepyspark 搜索数据框并随机选择要添加到新数据框的值
【发布时间】：2022-01-25 17:05:43
【问题描述】：

假设我有两个数据框：df, grp_df 是 df.groupby(["region"]).set_list()

 user        item        region
 james        I1         Canada
  amy         I5         Germany
 chris        I33        U.S.

grp_df

  region          Item_lst
  Canada         [I1, I2,... In]
  Germany        [ I3, I5, ... In]
  U.S.           [I33, I22, I11]
  ...             ...

对于每个用户，我想在同一区域内选择以前未购买的新商品并将其添加到新的 pyspark 数据框中。

new_df

user      item        region
james       I2        Canada
amy         I3        Germany
chris       I22        U.S.

在 pyspark 中最有效的方法是什么？

我的方法：

df = df.join(grp_df, ["region"], "left")

def get_neg_sample(item, item_lst):
    return np.random.choice(item_lst.remove(item))

get_neg_sample_udf = udf(get_neg_sample, IntegerType())

df.withColumn("neg_item", get_neg_sample_udf("item", "item_lst))

【问题讨论】：

标签： python apache-spark pyspark

【解决方案1】：

你要找的函数是 array_contains ，你可以用这个函数和 join 来得到你想要的结果

val newDf = df.join(gdf,df.col("country")===gdf.col("country") && !array_contains(gdf.col("item_list"),df.col("item")))

【讨论】：

感叹号是什么意思？
抱歉，gdf 应该有 "item_lst" 列
我更新了答案以反映正确的列名，感叹号是非运算符。它基本上说数组不包含这个值
hm.. 不是很理解这个。什么是===？尝试转换为 pyspark 但继续失败..
在python中它只是==而不是===

【解决方案2】：

对于 spark 2.4+，您可以使用 shuffle 和 array_remove。

new_df = df.join(grp_df, 'region').select('region', 'user', F.expr('shuffle(array_remove(Item_lst, item))[0]').alias('item'))
new_df.show(truncate=False)

【讨论】：