Pyspark：检查元组是否包含在元组列表中答案

【问题标题】：Pyspark: check if a tuple is contained in a list of tuplesPyspark：检查元组是否包含在元组列表中
【发布时间】：2019-12-01 10:57:45
【问题描述】：

我正在尝试分析来自 2 个不同来源（A 和 B）的数据的可靠性。由于字段范围相当不平等，因此我将重点放在常见字段上并进行比较。

在这里我选择了价格和数量，并希望确保元组 [priceA, quantityA] 包含在我的元组列表中 [[price1B, quantity1B], [price2B, quantity2B], .. ] 来自源 B。

我试图创建一个 udf 来查看其他引用，但我刚刚开始使用 Pyspark，我并不真正了解如何定义我的 udf 和在给定情况下指定的适当 DataType。

我的 2 个独立来源有 2 个数据框

我为每个 df 附加了一个新列“combined”：StructField(combined_a,ArrayType(IntegerType,true),false)))

df_a = df_a.withColumn("combined_a", array("Quantity", "PRICE"))

并创建了一个唯一元组列表：

list_a = list(df_a.select("combined_a").distinct().toPandas()["combined_a"])

输出列表_a

list_a = [ [81.0, 100.0], [56.0, 6.0], [10000.0, 45.32], [42.0, 6.0] ...]

我找不到任何可以满足我要求的内置函数：我想添加一个布尔类型的新列“combinaison_in_b”。试过了：

df_a = df_a.withColumn('combinaison_in_b_found' , col('combined_a').isin(list_b))

返回以下错误

An error occurred while calling z:org.apache.spark.sql.functions.lit.
: java.lang.RuntimeException: Unsupported literal type class java.util.ArrayList [50, 51]

继续使用 udf。试过了：

def IsInDataframe(combined_a , list_b):
    found = TRUE
    for c in combined_a
        if c not in list_b:
            found = False
        if found:
            return True
        else:
            return False

def udf_append(list_b):
    return udf(lambda combined_a : IsInDataframe(combined_a , list_b))

df_a.withColumn("combinaison_in_b_found", udf_append(list_b)(col("combined_a"))).cast('boolean')

(udf 语法取自pyspark how do we check if a column value is contained in a list

如果有人能解释它说 return udf 的部分，我将不胜感激

我想作为输出我的 df 附加列“combinaison_in_b_found”真/假。

_______________________________________________
id |    combined_a    | combinaison_in_b_found
1  |  [81.0, 100.0]   |    false
2  |  [56.0, 6.0]     |    true
...

【问题讨论】：

标签： arrays dataframe pyspark user-defined-functions contains

【解决方案1】：

试试这个：

df_a = spark.createDataFrame([(1,[81.0, 100.0]), (1, [56.0, 6.0]),(3,[77.0, 88.0]), (4,[42., 8.])], ('id', 'combined_a') )
df_a.show()
list_b = [ [81.0, 100.0], [56.0, 6.0], [10000.0, 45.32], [42.0, 6.0]]
print('list_b: {}'.format(list_b))
my_udf = udf(lambda pair: 'true' if pair in list_b else 'false', StringType())
df_a = df_a.withColumn('combinaison_in_b_found', my_udf(df_a['combined_a']))
df_a.show()

这是输出：

+---+-------------+
| id|   combined_a|
+---+-------------+
|  1|[81.0, 100.0]|
|  1|  [56.0, 6.0]|
|  3| [77.0, 88.0]|
|  4|  [42.0, 8.0]|
+---+-------------+

list_b: [[81.0, 100.0], [56.0, 6.0], [10000.0, 45.32], [42.0, 6.0]]
+---+-------------+----------------------+
| id|   combined_a|combinaison_in_b_found|
+---+-------------+----------------------+
|  1|[81.0, 100.0]|                  true|
|  1|  [56.0, 6.0]|                  true|
|  3| [77.0, 88.0]|                 false|
|  4|  [42.0, 8.0]|                 false|
+---+-------------+----------------------+

【讨论】：