Pyspark - 将结构列合并到数组中

Question

我想将多个结构列合并到一个数组中。

我尝试了 array(col1, col2) from .. 但它最终导致数据类型不匹配，即使所有 struct 列都是相同的类型。

查询 ->

select array(struct(f_name_add, True as is_data_found),struct(l_name_add, True as is_data_found)) as tag from (select array(map('value',f_name),map('value',f_add)) as f_name_add, array(map('value',l_name),map('value',l_add)) as l_name_add from (select distinct f_name, f_add, l_name, l_add from db.tabl1 where id = 'aaa')

输入

Sample Table

 id   f_name  f_add  l_name  l_add

 aaa  tom    in     nats    in

预期输出：

"tag":
 [
       {
            "f_name_add": [
                {
                    "value":"tom"
                },
                {
                    "value": "in"
                }
            ],
            "is_data_found": true
        },
        {
         "l_name_add": [
                {
                    "value":"nats"
                },
                {
                    "value": "in"
                }
            ],
            "is_data_found": true
        }
]

错误：

cannot resolve 'array(named_struct('f_name_add', __auto_generated_subquery_name.f_name_add, 'is_data_found', true), named_struct('l_name_add', __auto_generated_subquery_name.l_name_add, 'is_data_found', true))' due to data type mismatch: input to function array should all be the same type, but it's [struct<f_name_add:array<map<string,string>>,is_data_found:boolean>, struct<l_name_add:array<map<string,string>>,is_data_found:boolean>]

Answer 1

看起来错误消息说有两个structs： struct<f_name_add:array<map<string,string>> 与 struct<l_name_add:array<map<string,string>>。

由于f_name_add 和l_name_add，它们之间的比较并不相同。

在这里，我将尝试创建一个 json 字符串数组，但我不确定这是否正是您想要的 :(

1. 设置一个简单的例子

a = [('aaa', 'tom', 'in', 'nats', 'in'),('bbb', 'tom1', 'on', 'nats1', 'on'),]
df = spark.createDataFrame(a, ['id', 'f_name',  'f_add',  'l_name', 'l_add'])
df.show()


+---+------+-----+------+-----+
| id|f_name|f_add|l_name|l_add|
+---+------+-----+------+-----+
|aaa|   tom|   in|  nats|   in|
|bbb|  tom1|   on| nats1|   on|
+---+------+-----+------+-----+

1. 算法如下：

df.registerTempTable("tabl1")
df = spark.sql("select array(to_json(struct(f_name_add, True as is_data_found)), \
                             to_json(struct(l_name_add, True as is_data_found))) as tag \
                from (select array(map('value',f_name),map('value',f_add)) as f_name_add, \
                       array(map('value',l_name),map('value',l_add)) as l_name_add \
                from (select distinct f_name, f_add, l_name, l_add from tabl1 where id = 'aaa'))")

df.show(truncate=False)

+------------------------+
|tag                                                                                                                                          |
+------------------------+
|[{"f_name_add":[{"value":"tom"},{"value":"in"}],"is_data_found":true}, {"l_name_add":[{"value":"nats"},{"value":"in"}],"is_data_found":true}]|
+------------------------+

如果你运行df.dtypes，我们就会有，你可以看到我们这里实际上有一个 json 字符串数组。

[('tag', 'array<string>')]