在 Spark 中过滤有效和无效记录

Posted 2023-04-15

【中文标题】在 Spark 中过滤有效和无效记录

【英文标题】：Filter valid and invalid records in Spark

【发布时间】：2019-07-11 11:32:32

【问题描述】：

我有 pyspark 数据框，它有 'n' 行，每行有一列 result

结果列的内容是JSON

"crawlDate": "2019-07-03 20:03:44", "Code": "200", "c1": "XYZ", "desc": "desc",  "attributes": "abc":123, "def":456
"crawlDate": "2019-07-04 20:03:44", "Code": "200", "c1": "ABC", "desc": "desc1"
"crawlDate": "2019-07-04 20:03:44", "Code": "200", "c1": "ABC", "desc": "desc1", "attributes": "abc":456, "def":123

df.show()：

现在我想检查有多少条记录（ROWS）有 attributes 元素，有多少条记录没有。

我尝试在spark中使用array_contains、filter和explode函数，但没有得到结果。

有什么建议吗？

【参考方案1】：

import org.apache.spark.sql.functions._

df.select(get_json_object($"result", "$.attributes").alias("attributes")) .filter(col("attributes").isNotNull).count()

通过这个逻辑，我们可以得到属性现有记录计数的计数

供您参考，请阅读此内容 https://docs.databricks.com/spark/latest/dataframes-datasets/complex-nested-data.html

如果您的输入是 JSON 格式，则另一种解决方案，然后

val df = spark.read.json("path of json file")
df.filter(col("attributes").isNotNull).count()

我们可以在 python 中获得类似的 API。

【讨论】：

用简单的逻辑（包含），我得到了解决方案。发布相同。谢谢。

【参考方案2】：

经过一番努力，下面的简单逻辑奏效了

total_count = old_df.count()
new_df = old_df.filter(old_df.result.contains("attributes"))
success_count = new_df.count()
failure_count = total_count - success_count