reading parquet file with array<map<string,string>> column

import pyspark
spark = pyspark.sql.SparkSession.builder.getOrCreate()
df = spark.read.load(path)
pdf = df.toPandas()

>>> pdf["user"][0]["sessions"][0]["views"]
   [Row(is_search = True, price = None, search_string = 'ABC', segment = 'listing', time = 1571250719.393951), Row(is_search = True, price = None, search_string = 'ZYX', segment = 'homepage', time = 1571250791.588197), Row(is_search = True, price = None, search_string = 'XYZ', segment = 'listing', time = 1571250824.106184)]

import spark.implicits._

val arrayData = Seq(
   Row("James", List("Java", "Scala"), Map("hair" - > "black", "eye" - > "brown")),
   Row("Michael", List("Spark", "Java", null), Map("hair" - > "brown", "eye" - > null)),
   Row("Robert", List("CSharp", ""), Map("hair" - > "red", "eye" - > "")),
   Row("Washington", null, null),
   Row("Jefferson", List(), Map())
)

val arraySchema = new StructType()
   .add("name", StringType)
   .add("knownLanguages", ArrayType(StringType))
   .add("properties", MapType(StringType, StringType))

val df = spark.createDataFrame(spark.sparkContext.parallelize(arrayData), arraySchema)
df.printSchema()
df.show(false)

df.select($ "name", explode($ "knownLanguages"))
   .show(false)

+ -- -- -- - + -- -- -- +
|
name | col |
   + -- -- -- - + -- -- -- +
   |
   James | Java |
   |
   James | Scala |
   |
   Michael | Spark |
   |
   Michael | Java |
   |
   Michael | null |
   |
   Robert | CSharp |
   |
   Robert | |
   + -- -- -- - + -- -- -- +

+ -- -- -- - + -- -- + -- -- - +
|
name | key | value |
   + -- -- -- - + -- -- + -- -- - +
   |
   James | hair | black |
   |
   James | eye | brown |
   |
   Michael | hair | brown |
   |
   Michael | eye | null |
   |
   Robert | hair | red |
   |
   Robert | eye | |
   + -- -- -- - + -- -- + -- -- - +

df.select($ "name", explode_outer($ "knownLanguages"))
   .show(false)

DESCRIBE struct_demo;
+-------------------+--------------------------+
| name | type |
+-------------------+--------------------------+
| id | bigint |
| name | string |
| employee_info | struct< | | | employer:string, | | | id:bigint, | | | address:string | | |> |
   | places_lived | array<struct< | | | street:string, | | | city:string, | | | country:string | | |>> |
      | memorable_moments | map<string,struct< | | | year:int, | | | place:string, | | | details:string | | |>> |
         | current_address | struct< | | | street_address:struct< | | | street_number:int, | | | street_name:string, | | | street_type:string | | |>, |
            | | country:string, |
            | | postal_code:string |
            | | > |

CREATE TABLE struct_demo
(
id BIGINT,
name STRING,

-- A STRUCT as a top-level column. Demonstrates how the table ID column
-- and the ID field within the STRUCT can coexist without a name conflict.
employee_info STRUCT < employer: STRING, id: BIGINT, address: STRING>,

   -- A STRUCT as the element type of an ARRAY.
   places_lived ARRAY < STRUCT <street: STRING, city: STRING, country: STRING>>,

      -- A STRUCT as the value portion of the key-value pairs in a MAP.
      memorable_moments MAP < STRING, STRUCT < year: INT, place: STRING, details: STRING>>,

         -- A STRUCT where one of the fields is another STRUCT.
         current_address STRUCT < street_address: STRUCT <street_number: INT, street_name: STRING, street_type: STRING>, country: STRING, postal_code: STRING >
            )
            STORED AS PARQUET; 

// Encoders for most common types are automatically provided by importing spark.implicits._
import spark.implicits._

val peopleDF = spark.read.json("examples/src/main/resources/people.json")

// DataFrames can be saved as Parquet files, maintaining the schema information
peopleDF.write.parquet("people.parquet")

// Read in the parquet file created above
// Parquet files are self-describing so the schema is preserved
// The result of loading a Parquet file is also a DataFrame
val parquetFileDF = spark.read.parquet("people.parquet")

// Parquet files can also be used to create a temporary view and then used in SQL statements
parquetFileDF.createOrReplaceTempView("parquetFile")
val namesDF = spark.sql("SELECT name FROM parquetFile WHERE age BETWEEN 13 AND 19")
namesDF.map(attributes => "Name: " + attributes(0)).show()
// +------------+
// |       value|
// +------------+
// |Name: Justin|
// +------------+

import org.apache.spark.api.java.function.MapFunction;
import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;

Dataset<Row> peopleDF = spark.read().json("examples/src/main/resources/people.json");

   // DataFrames can be saved as Parquet files, maintaining the schema information
   peopleDF.write().parquet("people.parquet");

   // Read in the Parquet file created above.
   // Parquet files are self-describing so the schema is preserved
   // The result of loading a parquet file is also a DataFrame
   Dataset<Row> parquetFileDF = spark.read().parquet("people.parquet");

      // Parquet files can also be used to create a temporary view and then used in SQL statements
      parquetFileDF.createOrReplaceTempView("parquetFile");
      Dataset<Row> namesDF = spark.sql("SELECT name FROM parquetFile WHERE age BETWEEN 13 AND 19");
         Dataset<String> namesDS = namesDF.map(
            (MapFunction<Row, String>) row -> "Name: " + row.getString(0),
               Encoders.STRING());
               namesDS.show();
               // +------------+
               // | value|
               // +------------+
               // |Name: Justin|
               // +------------+

peopleDF = spark.read.json("examples/src/main/resources/people.json")

# DataFrames can be saved as Parquet files, maintaining the schema information.
peopleDF.write.parquet("people.parquet")

# Read in the Parquet file created above.
# Parquet files are self - describing so the schema is preserved.
# The result of loading a parquet file is also a DataFrame.
parquetFile = spark.read.parquet("people.parquet")

# Parquet files can also be used to create a temporary view and then used in SQL statements.
parquetFile.createOrReplaceTempView("parquetFile")
teenagers = spark.sql("SELECT name FROM parquetFile WHERE age >= 13 AND age <= 19")
teenagers.show()
# + -- -- -- +
   # | name |
   # + -- -- -- +
   # | Justin |
   # + -- -- -- +

df < -read.df("examples/src/main/resources/people.json", "json")

# SparkDataFrame can be saved as Parquet files, maintaining the schema information.
write.parquet(df, "people.parquet")

# Read in the Parquet file created above.Parquet files are self - describing so the schema is preserved.
# The result of loading a parquet file is also a DataFrame.
parquetFile < -read.parquet("people.parquet")

# Parquet files can also be used to create a temporary view and then used in SQL statements.
createOrReplaceTempView(parquetFile, "parquetFile")
teenagers < -sql("SELECT name FROM parquetFile WHERE age >= 13 AND age <= 19")
head(teenagers)
# # name
# # 1 Justin

# We can also run custom R - UDFs on Spark DataFrames.Here we prefix all the names with "Name:"
schema < -structType(structField("name", "string"))
teenNames < -dapply(df, function(p) {
   cbind(paste("Name:", p$name))
}, schema)
for (teenName in collect(teenNames) $name) {
   cat(teenName, "\n")
}
# # Name: Michael
# # Name: Andy
# # Name: Justin

CREATE TEMPORARY VIEW parquetTable
USING org.apache.spark.sql.parquet
OPTIONS(
   path "examples/src/main/resources/people.parquet"
)

SELECT * FROM parquetTable

path└── to└── table├── gender = male│├──...│││├──country = US││└── data.parquet│├── country = CN││└── data.parquet│└──...└──gender = female├──...│├──country = US│└── data.parquet├── country = CN│└── data.parquet└──...

import org.apache.spark.sql.SparkSession
assert(spark.isInstanceOf[SparkSession])

import org.apache.spark.sql.DataFrameReader
val reader = spark.read
assert(reader.isInstanceOf[DataFrameReader])

csv(csvDataset: Dataset[String]): DataFrame
csv(path: String): DataFrame
csv(paths: String * ): DataFrame

format(source: String): DataFrameReader

jdbc(
   url: String,
   table: String,
   predicates: Array[String],
   connectionProperties: Properties): DataFrame
jdbc(
   url: String,
   table: String,
   properties: Properties): DataFrame
jdbc(
   url: String,
   table: String,
   columnName: String,
   lowerBound: Long,
   upperBound: Long,
   numPartitions: Int,
   connectionProperties: Properties): DataFrame

json(jsonDataset: Dataset[String]): DataFrame
json(path: String): DataFrame
json(paths: String * ): DataFrame

load(): DataFrame
load(path: String): DataFrame
load(paths: String * ): DataFrame