Pyspark - 过滤包含空值的 ArrayType 行答案

【问题标题】：Pyspark -- Filter ArrayType rows which contain null valuePyspark - 过滤包含空值的 ArrayType 行
【发布时间】：2021-12-24 13:55:09
【问题描述】：

我是 PySpark 的初学者。假设我有一个像这样的 Spark 数据框：

test_df = spark.createDataFrame(pd.DataFrame({"a":[[1,2,3], [None,2,3], [None, None, None]]}))

现在我希望过滤数组不包含 None 值的行（在我的情况下只保留第一行）。

我试过用：

test_df.filter(array_contains(test_df.a, None))

但是它不起作用并抛出错误：

AnalysisException: “无法解析 'array_contains(a, NULL)' 由于数据类型不匹配：空类型值不能用作 arguments;;\n'Filter array_contains(a#166, null)\n+- LogicalRDD [a#166]，假\n

我应该如何以正确的方式过滤？非常感谢！

【问题讨论】：

标签： apache-spark pyspark apache-spark-sql

【解决方案1】：

你需要使用forall函数。

df = test_df.filter(F.expr('forall(a, x -> x is not null)'))
df.show(truncate=False)

【讨论】：

感谢您的回复。但是我的 spark 版本是 2.4.0 并且没有 forall 功能。有没有不升级spark版本的替代方法？
正如blackbishop所说，2.4版本可以使用exists

【解决方案2】：

您可以使用aggregate 高阶函数来计算空值的数量并过滤count = 0 的行。这将使您能够删除数组中至少有1 个None 的所有行。

data_ls = [
    (1, ["A", "B"]),
    (2, [None, "D"]),
    (3, [None, None])
]

data_sdf = spark.sparkContext.parallelize(data_ls).toDF(['a', 'b'])

data_sdf.show()

+---+------+
|  a|     b|
+---+------+
|  1|[A, B]|
|  2| [, D]|
|  3|   [,]|
+---+------+

# count the number of nulls within an array
data_sdf. \
    withColumn('c', func.expr('aggregate(b, 0, (x, y) -> x + int(y is null))')). \
    show()

+---+------+---+
|  a|     b|  c|
+---+------+---+
|  1|[A, B]|  0|
|  2| [, D]|  1|
|  3|   [,]|  2|
+---+------+---+

创建列后，您可以将过滤器应用为filter(func.col('c')==0)。

【讨论】：

【解决方案3】：

你可以使用exists函数：

test_df.filter("!exists(a, x -> x is null)").show() 

#+---------+
#|        a|
#+---------+
#|[1, 2, 3]|
#+---------+

【讨论】：