PySpark DataFrame - 过滤嵌套列答案

【问题标题】：PySpark DataFrame - Filter nested columnPySpark DataFrame - 过滤嵌套列
【发布时间】：2021-03-16 18:16:43
【问题描述】：

我知道那里有很多类似的问题，但我没有找到任何与我的场景完全匹配的问题，所以请不要对重复标志太感到高兴。我正在使用 Spark 3.0.1 在 Azure Databricks 中使用 Python 3 笔记本。

我有以下数据框

+---+---------+--------+
|ID |FirstName|LastName|
+---+---------+--------+
|1  |John     |Doe     |
|2  |Michael  |        |
|3  |Angela   |Merkel  |
+---+---------+--------+

可以用这段代码创建

from pyspark.sql.types import StructType,StructField, StringType, IntegerType
import pyspark.sql.functions as F

data2 = [(1,"John","Doe"),
    (2,"Michael",""),
    (3,"Angela","Merkel")
  ]

schema = StructType([ \
    StructField("ID",IntegerType(),True), \
    StructField("FirstName",StringType(),True), \
    StructField("LastName",StringType(),True), \
  ])
 
df1 = spark.createDataFrame(data=data2,schema=schema)
df1.printSchema()
df1.show(truncate=False)

我把它转换成这个DataFrame

+---+-----------------------------------------+
|ID |Names                                    |
+---+-----------------------------------------+
|1  |[[FirstName, John], [LastName, Doe]]     |
|2  |[[FirstName, Michael], [LastName, ]]     |
|3  |[[FirstName, Angela], [LastName, Merkel]]|
+---+-----------------------------------------+

使用此代码

df2 = df1.select(
            'ID', 
            F.array(
                F.struct(
                    F.lit('FirstName').alias('NameType'), 
                    F.col('FirstName').alias('Name')
                ), 
                F.struct(
                    F.lit('LastName').alias('NameType'), 
                    F.col('LastName').alias('Name')
                )
            ).alias('Names')
        )

df2.printSchema()
df2.show(truncate=False)

现在，我正在尝试过滤掉Names，其中LastName 为空或为空字符串。我的总体目标是拥有一个可以在 json 中序列化的对象，其中不包括 Names 和空的 Name 值。

像这样

[
    {
        "ID": 1,
        "Names": [
            {
                "NameType": "FirstName",
                "Name": "John"
            },
            {
                "NameType": "LastName",
                "Name": "Doe"
            }
        ]
    },
    {
        "ID": 2,
        "Names": [
            {
                "NameType": "FirstName",
                "Name": "Michael"
            }
        ]
    },
    {
        "ID": 3,
        "Names": [
            {
                "NameType": "FirstName",
                "Name": "Angela"
            },
            {
                "NameType": "LastName",
                "Name": "Merkel"
            }
        ]
    }
]

我试过了

df2 = df1.select(
            'ID', 
            F.array(
                F.struct(
                    F.lit('FirstName').alias('NameType'), 
                    F.col('FirstName').alias('Name')
                ), 
                F.struct(
                    F.lit('LastName').alias('NameType'), 
                    F.col('LastName').alias('Name')
                )
            ).filter(lambda x: x.col('LastName').isNotNull()).alias('Names')
        )

但我收到错误 'Column' object is not callable。

我也尝试过df2 = df2.filter(F.col('Names')['LastName']) > 0)，但这给了我一个invalid syntax 错误。

我试过了

df2 = df2.filter(lambda x: (len(x)>0), F.col('Names')['LastName'])

但这会给出错误TypeError: filter() takes 2 positional arguments but 3 were given。

有人可以告诉我如何让它工作吗？

【问题讨论】：

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

【解决方案1】：

可以使用高阶函数filter：

import pyspark.sql.functions as F

df3 = df2.withColumn(
    'Names', 
    F.expr("filter(Names, x -> case when x.NameType = 'LastName' and length(x.Name) = 0 then false else true end)")
)

df3.show(truncate=False)
+---+-----------------------------------------+
|ID |Names                                    |
+---+-----------------------------------------+
|1  |[[FirstName, John], [LastName, Doe]]     |
|2  |[[FirstName, Michael]]                   |
|3  |[[FirstName, Angela], [LastName, Merkel]]|
+---+-----------------------------------------+

【讨论】：