Different types of data in a same column in Parquet partition

user_7938749 0
Viewed 6290

I'm getting error while reading PARQUET file from S3, and the reason comes that "final_height" column got String and Double types in a same partition. For information, there are over 20 columns in parquet file. The errors I'm getting are:

ERROR 1: Failed merging schema of file ".../part1.gz.parquet":

ERROR 2: Caused by: org.apache.spark.SparkException:
Failed to merge fields 'final_height' and 'final_height'. Failed to merge incompatible data types double and string

ERROR 3: com.databricks.sql.io.FileReadException:
Error while reading file ".../part1.gz.parquet".
Parquet column cannot be converted. Column: [final_height], Expected: StringType, Found: DOUBLE

ERROR 4: com.databricks.sql.io.FileReadException:
Error while reading file ".../part1.gz.parquet".
Parquet column cannot be converted. Column: [final_height], Expected: DoubleType, Found: BINARY

ERROR 5: org.apache.spark.sql.execution.datasources.SchemaColumnConvertNotSupportedException

Found some solutions when "part1.gz.parquet" has String data for X column, and "part2.gz.parquet" has Double data in a same column. But they aren't working when different types in a same column found in a same partition.

Tried:

  • reading with mergeSchema and inferSchema options (got ERROR 1 & 2)
  • reading with predefined manual schema with type String for "final_height" column (got ERROR 3 & 5)
  • reading with predefined manual schema with type Double for "final_height" column (got ERROR 4 & 5)

Hoping for some solution :)

Thanks!

parquet databricks apache-spark amazon-s3 scala
1 Answers

An ideal approach would be to read entire dataframe as Binary (Array[Byte]) data type, and then casting corresponding values to their compatible data types, however, Spark does not allow to read Double Data type as Binary data type. Therefore could not proceed with this approach.

A hack could be, to set Spark property "spark.sql.files.ignoreCorruptFiles" to true and then read the files with the desired schema. Files that don’t match the specified schema are ignored. The resultant dataset contains only data from those files that match the specified schema. Thus read two dataframe one with String data type and other with Double data type and then cast any one of them to a single data type and then finally union them.

val stringSchema = StructType(StructField("final_height", StringType, false) :: Nil)
val doubleSchema = StructType(StructField("final_height", DoubleType, false) :: Nil)

spark.conf.set("spark.sql.files.ignoreCorruptFiles", "true")

val stringDF = spark.read.schema(stringSchema).parquet("path/")
val doubleDF = spark.read.schema(doubleSchema).parquet("path/")
//Cast to compatible type
val doubleToStringDF = doubleDF.select(col("final_height").cast(StringType))

val finalDF = stringDF.union(doubleToStringDF)
user_73896040
Related