Spark Sql 查询嵌套记录。我想先过滤嵌套的记录数组，然后爆炸（将它们展开成行）

Posted 2023-04-12

【中文标题】Spark Sql 查询嵌套记录。我想先过滤嵌套的记录数组，然后爆炸（将它们展开成行）

【英文标题】：Spark Sql Query On Nested Records. I want to do Filter nested array of records first and then explode(expand them into rows)

【发布时间】：2016-05-23 05:01:32

【问题描述】：

我想首先过滤具有 Max 的行，然后我想只分解嵌套列中具有 Max 的行。

我的 Avro 记录：

"name": "Parent",
"type":"record",
"fields":[
    "name": "firstname", "type": "string",

    
        "name":"children",
        "type":
            "type": "array",
            "items":
                        "name":"child",
                        "type":"record",
                        "fields":[

                       "name":"name", "type":"string"                                                                    
                       "name":"price","type":["long", "null"]

                        ]
                    
            
    
]

我正在使用 Spark SQL 上下文来查询读取的数据帧。所以如果输入是

Row no   Firstname Children.name
    1    John       [[Max, 20],[Pg, 22]]
    2    Bru        [[huna, 10], [aman, 12]]

我首先通过分解内表进行查询。所以嵌套列分成 2 行。

Row no   Firstname Children.name    children.price
        1    John       Max               20
        1    John       Pg                22
        2    Bru        huna              10
        2    Bru        aman              12

q1) 我想首先过滤包含 Max 的行，然后我想只分解包含 Max 的行。在当前情况下，如果我在一列中有数百万个值，那么它首先生成数百万行，然后检查 Max 是否存在。

q2) 我想首先过滤价格 > 12 的行，然后我想只分解价格 > 12 的行。在当前情况下，如果我在一列中有数百万个值，则它首先生成数百万行，然后检查是否存在 price > 12。

类似这样的： val results = sqlc.sql("SELECT firstname, child.name FROM parent LATERAL VIEW explode(children) childTable AS child where child.price > 12")

【问题讨论】：

过滤包含 max 的子名称以创建新的数据框，然后爆炸。你试过吗？

以下查询不起作用 val results = sqlc.sql("SELECT firstname, children.name FROM parent where children.name = 'Max'")

我说包含不等于。您可以在 spark scala 文档中找到所有可以使用的 sql 函数

***.com/a/35628252/1560062

另外请查看您的其他问题！他们暂时悬而未决。

【参考方案1】：

以下是两个问题的答案： ans1) 如果要查找嵌套记录数组中是否存在“字符串”：

var results = sqlc.sql("SELECT firstname, children.name  FROM parent where array_contains(children['name'], 'pg') ")

ans2) 如果你想对嵌套记录数组应用条件。使用 UDF

sqlc.udf.register("myPriceFilter", (price: mutable.WrappedArray[String]) => (price exists (a =>  (a.toLong < 67735) )))

var results = sqlc.sql("SELECT firstname, explode(children.price)  FROM parent where myPriceFilter(children['price']) ")