How to change Spark Dataframe column data type in an array

Question

I am having two small problems regarding my one bigger problem: I want to read in JSON data once a day and save it as Parquet for later data-related work. Working with parquet is so much faster. But the thing where I am stuck at is the fact that when reading that parquet, Spark always tries to get the schema from the schema file or just takes the schema from the first parquet file and presumes the schema is same for all files. But there are cases when we don't have any data for some days in some columns.

So let's say I have a JSON file with data with the following schema:

root
 |-- Id: long (nullable = true)    
 |-- People: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- Name: string (nullable = true)
 |    |    |-- Amount: double (nullable = true)

And then I have another JSON file where there is no data for the "People" column. And therefore the schema is the following:

root
 |-- Id: long (nullable = true)    
 |-- People: array (nullable = true)
 |    |-- element: string (containsNull = true)

When I would read them in together with read.json, Spark goes through all the files and infers the merged schema from these, more specifically from the first one and just leaves the rows from the second file empty, but the schema is correct.

But when I read these separately and write to parquet separately, then I cannot read them together because for Parquet, the schema doesn't match and I get an error.

My first idea was to read in the file with missing data and change its schema manually by casting column types to match the first schema, but this manual conversion is faulty, it can be out of sync and I don't even know how to cast this string type to array or struct type.

And another problem is when the "Amount" field has only full integers, then Spark reads them in as longs but not doubles, as is necessary. But if I use:

val df2 = df.withColumn("People.Amount", col("People.Amount").cast(org.apache.spark.sql.types.ArrayType(org.apache.spark.sql.types.DoubleType,true)))

Then it doesn't change the type of the original column, but adds a new column named People.Amount

flyingmeatball flyingmeatball · Accepted Answer · 2017-01-26T16:14:01

I think you may be able to jigger something up with schema merging (see docs here). If the first parquet you have has the correct schema, could you then do something like this to apply that schema to the new parquets?

// Read the partitioned table
val mergedDF = spark.read.option("mergeSchema", "true").parquet("data/test_table")
mergedDF.printSchema()

edit

You say there are 200+ columns, do you know them all already? I see two way forwards, and there's probably a number of ways to accomplish this. One is you define all the fields you can see in advance. What I've done in the past is create a json file with a single dummy record that has all the fields I want and is typed exactly as i want them. Then, you can just always load that record at the same time as your "Monday" or "Tuesday" dataset and strip it out post load. It probably isn't best practice, but this is how I've stumbled my way forward.

The alternative is to stop trying to load/save individual datasets in the correct schema, and set the schema once you've loaded all the data. Doesn't sound like the path you want to go down, but at least then you won't have this specific problem.

How to change Spark Dataframe column data type in an array

1 Answers