与 get_json_object 一起使用时，json 路径无法按预期工作

Question

太长了；与 一起使用时，以下 JSON 路径对我不起作用pyspark.sql.functions.get_json_object。

$.Blocks[?(@.Type=='LINE')].Confidence

长版...

我想按单行内的数组进行分组

例如，对于下面的结构

root
|--id: string
|--payload: string

的值payload是一个表示 json 块的字符串，其结构如下所示

{
        "Blocks": [
            {
                "Type": "LINE",
                "Confidence": 90
            },
            {
                "Type": "LINE",
                "Confidence": 98
            },
            {
                "Type": "WORD",
                "Confidence": 99
            },
            {
                "Type": "PAGE",
                "Confidence": 97
            },
            {
                "Type": "PAGE",
                "Confidence": 89
            },
            {
                "Type": "WORD",
                "Confidence": 99
            }
        ]
    }

我想按类型汇总所有置信度，以便我们得到以下新列......

{
    "id": 12345,
    "payload": "..."
    "confidence": [
        {
            "Type": "WORD",
            "Confidence": [
                99,
                99
            ]
        },
        {
            "Type": "PAGE",
            "Confidence": [
                97,
                89
            ]
        },
        {
            "Type": "LINE",
            "Confidence": [
                90,
                98
            ]
        }
    ]
}

为此，我计划使用get_json_object(...)提取每种类型块的置信度。

例如...

get_json_object(col("payload"), "$.Blocks[?(@.Type=='LINE')].Confidence")

但$.Blocks[?(@.Type=='LINE')].Confidence不断返回null。这是为什么？

我通过在https://jsonpath.curiousconcept.com/#上针对上面的示例json进行测试来验证 json 路径是否有效payload，并得到以下结果...

[
   90,
   98
]

如果使用上面的路径不是一种选择，那么如何聚合它？

以下是完整的代码示例。我希望第一个在置信栏中.show()打印出来。[90, 98]

from pyspark.sql import SparkSession
from pyspark.sql.types import StructField, StringType, StructType, IntegerType
from pyspark.sql.functions import get_json_object, col


def main():
    spark = SparkSession.builder.appName('test_session').getOrCreate()
    df = spark.createDataFrame([
        (
            12345,  # id
            """
{
        "Blocks": [
            {
                "Type": "LINE",
                "Confidence": 90
            },
            {
                "Type": "LINE",
                "Confidence": 98
            },
            {
                "Type": "WORD",
                "Confidence": 99
            },
            {
                "Type": "PAGE",
                "Confidence": 97
            },
            {
                "Type": "PAGE",
                "Confidence": 89
            },
            {
                "Type": "WORD",
                "Confidence": 99
            }
        ]
    }

            """  # payload
        )
    ],
        StructType(
            [
                StructField("id", IntegerType(), True),
                StructField("payload", StringType(), True)
            ])
    )
    
    # this prints out null (why?)
    df.withColumn("confidence", get_json_object(col("payload"), "$.Blocks[?(@.Type=='LINE')].Confidence")).show()
    
    # this prints out the correct values, [90,98,99,97,89,99]
    df.withColumn("confidence", get_json_object(col("payload"), "$.Blocks[*].Confidence")).show()


if __name__ == "__main__":
    main()

Answer 1

Spark如何解析JSON路径没有官方文档，但根据其源代码，看起来它不支持@作为当前对象。事实上它支持非常有限的语法：

// parse `[*]` and `[123]` subscripts
// parse `.name` or `['name']` child expressions
// child wildcards: `..`, `.*` or `['*']`

因此，如果您愿意采用另一种方法，这里使用预定义的架构和函数，例如from_json, explode, collect_list：

// parse `[*]` and `[123]` subscripts
// parse `.name` or `['name']` child expressions
// child wildcards: `..`, `.*` or `['*']`