如何在spark SQL中实现嵌套collect_list？

Question

我对数据砖spark SQL是新手。我正在寻找嵌套的collect_list并试图找出答案。下面是我的spark实际sql查询

select
            policy.CustomerId,
            collect_list(struct(Number, Type, Id, Product.product))as policydetail 
         from
            policy 
            Left Join
               (
                  SELECT
                     policy.CustomerId,
                     Collect_list(struct(ProductId, productname)) as Product 
                  FROM
                     policy 
                  group by 
                     CustomerId
               )
               product 
               on product.CustomerId = policy.CustomerId
               group by 
               policy.CustomerId

我修改如下

select
                policy.id,
                collect_list(struct(Number, Type, Id, Collect_list(struct(ProductId, productname))))as policydetail 
            from
                policy             
            group by 
               policy.CustomerPartyId

修改查询后，我得到如下错误

"Policy":[
            {
               "Number":"123456",
               "Type":"new",
               "Id":"34355656",
               "Product":[
                  {
                     "ProductId":"2526",
                     "ProductName":"abc"
                  }
               ]
            }
         ]

from pyspark import SparkConf
from pyspark import SparkContext
from pyspark.sql.session import SparkSession
from pyspark.sql import SQLContext

sc = sc = SparkContext.getOrCreate()
spark = SparkSession(sc)
sc_sql = SQLContext(sc)

#ExtractConfig
policy = spark.read.format('csv').options(header='true', inferSchema='true').load("D:/policy.csv")
#RawPath


policy.createOrReplaceTempView("policy")

policydetails = sc_sql.sql("select policy.CustomerId, collect_list(struct(Number, Type, Id, Collect_list(struct(ProductId, productname))))as policydetail from policy group by policy.CustomerId")

pandasDF = policydetails.toPandas()
pandasDF = policydetails.toPandas().to_csv('data1.csv')

Answer 1

我没有测试，因为我没有样本数据。

尝试下面的查询。

select 
    policy.CustomerId, 
    collect_list(
       struct(
          Number, 
          Type, 
          Id, 
          array(
              struct(
                  ProductId, 
                  productname)
              )
       )
) as policydetail 
from policy 
group by policy.CustomerId

或

select 
    collect_list(
           struct(
               Number, 
               Type, 
               Id, 
               array(
                  struct(
                      ProductId, 
                      productname
                  )
               )
           )
) as policydetail 
from policy