spark dataframes : reading json having duplicate column names but different datatypes

user_3881782 0
Viewed 1628

I have json data like below where version field is the differentiator -

file_1 = {"version": 1, "stats": {"hits":20}}

file_2 = {"version": 2, "stats": [{"hour":1,"hits":10},{"hour":2,"hits":12}]}

In the new format, stats column is now Arraytype(StructType).

Earlier only file_1 was needed so I was using

spark.read.schema(schema_def_v1).json(path)

Now I need to read both these type of multiple json files which come together. I cannot define stats as string in schema_def as that would affect the corruptrecord feature(for stats column) which checks malformed json and schema compliance of all the fields.

Example df output required in 1 read only -

version | hour | hits
1       | null | 20
2       | 1    | 10
2       | 2    | 12

I have tried to read with mergeSchema option but that makes stats field String type.

Also, I have tried making two dataframes by filtering on the version field, and applying spark.read.schema(schema_def_v1).json(df_v1.toJSON). Here also stats column becomes String type.

I was thinking if while reading, I could parse the df column headers as stats_v1 and stats_v2 on basis of data-types can solve the problem. Please help with any possible solutions.

apache-spark-sql apache-spark jsonschema jackson json
3 Answers

UDF to check string or array, if it is string it will convert string to an array.

import org.apache.spark.sql.functions.udf
import org.json4s.{DefaultFormats, JObject}
import org.json4s.jackson.JsonMethods.parse
import org.json4s.jackson.Serialization.write
import scala.util.{Failure, Success, Try}

object Parse {
    implicit val formats = DefaultFormats
    def toArray(data:String) = {
      val json_data = (parse(data))
      if(json_data.isInstanceOf[JObject]) write(List(json_data)) else data
    }
}

val toJsonArray = udf(Parse.toArray _)


scala> "ls -ltr /tmp/data".!
total 16
-rw-r--r--  1 srinivas  root  37 Jun 26 17:49 file_1.json
-rw-r--r--  1 srinivas  root  69 Jun 26 17:49 file_2.json
res4: Int = 0

scala> val df = spark.read.json("/tmp/data").select("stats","version")
df: org.apache.spark.sql.DataFrame = [stats: string, version: bigint]

scala> df.printSchema
root
 |-- stats: string (nullable = true)
 |-- version: long (nullable = true)


scala> df.show(false)
+-------+-------------------------------------------+
|version|stats                                      |
+-------+-------------------------------------------+
|1      |{"hits":20}                                |
|2      |[{"hour":1,"hits":10},{"hour":2,"hits":12}]|
+-------+-------------------------------------------+

Output

scala> 

import org.apache.spark.sql.types._
val schema = ArrayType(MapType(StringType,IntegerType))

df
.withColumn("json_stats",explode(from_json(toJsonArray($"stats"),schema)))
.select(
    $"version",
    $"stats",
    $"json_stats".getItem("hour").as("hour"),
    $"json_stats".getItem("hits").as("hits")
).show(false)

+-------+-------------------------------------------+----+----+
|version|stats                                      |hour|hits|
+-------+-------------------------------------------+----+----+
|1      |{"hits":20}                                |null|20  |
|2      |[{"hour":1,"hits":10},{"hour":2,"hits":12}]|1   |10  |
|2      |[{"hour":1,"hits":10},{"hour":2,"hits":12}]|2   |12  |
+-------+-------------------------------------------+----+----+

Without UDF

scala> val schema = ArrayType(MapType(StringType,IntegerType))

scala> val expr = when(!$"stats".contains("[{"),concat(lit("["),$"stats",lit("]"))).otherwise($"stats")

df
.withColumn("stats",expr)
.withColumn("stats",explode(from_json($"stats",schema)))
.select(
    $"version",
    $"stats",
    $"stats".getItem("hour").as("hour"),
    $"stats".getItem("hits").as("hits")
)
.show(false)

+-------+-----------------------+----+----+
|version|stats                  |hour|hits|
+-------+-----------------------+----+----+
|1      |[hits -> 20]           |null|20  |
|2      |[hour -> 1, hits -> 10]|1   |10  |
|2      |[hour -> 2, hits -> 12]|2   |12  |
+-------+-----------------------+----+----+
user_85934140

Read the second file first, explode stats, use schema to read first file.


from pyspark.sql import SparkSession
from pyspark.sql.functions import  explode

spark = SparkSession.builder.getOrCreate()
sc = spark.sparkContext

file_1 = {"version": 1, "stats": {"hits": 20}}

file_2 = {"version": 2, "stats": [{"hour": 1, "hits": 10}, {"hour": 2, "hits": 12}]}

df1 = spark.read.json(sc.parallelize([file_2])).withColumn('stats', explode('stats'))
schema = df1.schema

spark.read.schema(schema).json(sc.parallelize([file_1])).printSchema()

output >> root
 |-- stats: struct (nullable = true)
 |    |-- hits: long (nullable = true)
 |    |-- hour: long (nullable = true)
 |-- version: long (nullable = true)
user_80711240

IIUC, you can read the JSON files using spark.read.text and then parse the value with json_tuple, from_json. notice for stats field we use coalesce to parse fields based on two or more schema. (add wholetext=True as an argument of spark.read.text if each file contains a single JSON document cross multiple lines)

from pyspark.sql.functions import json_tuple, coalesce, from_json, array

df = spark.read.text("/path/to/all/jsons/")

schema_1 = "array<struct<hour:int,hits:int>>"
schema_2 = "struct<hour:int,hits:int>"

df.select(json_tuple('value', 'version', 'stats').alias('version', 'stats')) \
    .withColumn('status', coalesce(from_json('stats', schema_1), array(from_json('stats', schema_2)))) \
    .selectExpr('version', 'inline_outer(status)') \
    .show()
+-------+----+----+
|version|hour|hits|
+-------+----+----+
|      2|   1|  10|
|      2|   2|  12|
|      1|null|  20|
+-------+----+----+
user_95107290
Related