python - Pyspark 对列中列表中的值进行联接和操作

Question

我有两个数据框，即

df1 = sc.parallelize([
  ['u1', 'type1', ['a', 'b']],
  ['u2', 'type1', ['a', 'c', 'd']],
  ['u1', 'type2', ['d']]
]).toDF(('person', 'type', 'keywords'))

df2 = sc.parallelize([
  ['a', 2],
  ['b', 1],
  ['c', 0],
  ['d', 1],
  ['e', 3],
]).toDF(('keyword', 'score'))

我需要计算每个人和每个类型其关键字的平均分数。因此，person 'u1' 在 type 'type1' 上的平均值将为 1.5，因为它的关键字 'a' 和 'b' 贡献为 2+1/2 =1.5

我尝试了一种包含连接的方法:

df = df1.join(df2) \
  .select('person', 'type', 'keywords', 'keyword', 'score') \
  .groupBy('person', 'type') \
  .agg(avg('score'))

但问题是，它是计算每个可能的关键字的平均值，而不仅仅是计算用户和类型所具有的关键字，因此我到处都得到 1.4，这是所有关键字的所有分数除以其数量的总和。我只需总结每个用户和类型的 keywords 列表中这些关键字的分数。

Answer 1

您必须首先分解关键字:

from pyspark.sql.functions import explode, avg, col

(df1.select("person", "type", explode("keywords").alias("keyword"))
    .join(df2, "keyword")
    .groupBy("person", "type")
    .agg(avg("score")))

虽然可以做这样的事情

from pyspark.sql.functions import expr

(df1.join(df2, expr("array_contains(keywords, keyword)"))
    .groupBy("person", "type")
    .agg(avg("score")))

为了达到相同的结果，您在实践中要避免这种情况，以避免扩展到笛卡尔积。