将每个 spark 数据框元素与相同数据框的所有其余部分进行比较

Question

我正在寻找一种有效的方法来将一些映射函数应用于数据帧中的每对元素。例如

records = spark.createDataFrame(
    [(1, 'a'), (2, 'b'), (3, 'c'), (4, 'd')], \
    ['id', 'val'])
records.show()

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

我想取值 a、b、c、d 并将它们与其他所有值进行比较：

a -> b
a -> c
a -> d
b -> c
b -> d
c -> d

相比之下，我的意思是自定义函数，它采用这两个值并计算它们之间的一些相似性指数。 假设输入数据框可能包含一千万个元素，您能否建议执行此计算的有效方法？

Spark 版本 2.4.6 (AWS emr-5.31.0)，使用带有 pyspark 的 EMR 笔记本

Answer 1

将val 列值收集到lookup 列中。然后将lookup 数组中的每个值与val 列进行比较。

检查下面的代码。

>>> records\ 
.select(F.collect_list(F.struct(F.col("id"),F.col("val"))).alias("data"),F.collect_list(F.col("val")).alias("lookup"))\ 
.withColumn("data",F.explode(F.col("data"))) \
.select("data.*",F.expr("filter(lookup,v -> v != data.val)").alias("lookup")) \
#.withColumn("compare",expr("transform(lookup, v -> val [.....] )")) # May be you can add your logic in this -> [.....] 
.show()

+---+---+---------+
| id|val|   lookup|
+---+---+---------+
|  1|  a|[b, c, d]|
|  2|  b|[a, c, d]|
|  3|  c|[a, b, d]|
|  4|  d|[a, b, c]|
+---+---+---------+

Answer 2

这是一个带有 collect_list 聚合的交叉连接操作。如果您希望 a 的匹配列表仅包含 [b,c,d]，则应在执行 collect_list 之前应用该过滤器。

records.alias("lhs")
.crossJoin(episodes.alias("rhs"))
.filter("lhs.val!=rhs.val")
.groupBy("lhs")
.agg(functions.collect_list("rhs.val").alias("lookup"))
.selectExpr("lhs.*", "lookup");