how to do nested collect_list in spark sql?

user_7773107 0
Viewed 405

I'm new to data bricks spark SQL. I'm looking to nested collect_list and tried to find out. Below is my spark actual sql query

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

I modified as below

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

after modifying the query, I got the error as below

It is not allowed to use an aggregate function in the argument of another aggregate function. Please use the inner aggregate function in a sub-query.;

Is there any alternative to approach below Json from table columns.

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

input to sql queries:
policy table which contains columns(id,number, type, transactionid, productid, productname)
output of sql queries:
[Row(Number='POL-CP-155', Type='Applicatn', Id=14924102, product=[Row(ProductId=2526, productname='Commercial Property')])]

here is my code for running spark sql query

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')

please find the image of policy.csv policy

Thanks in advance!

pyspark apache-spark-sql apache-spark
1 Answers

I have not tested as I don't have sample data.

Try below queries.

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

Or

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