Use spark SQL to read non-existing column with Parquet format

user_7557093 0
Viewed 2155

I have two month parquet file 2017_01.parquet and 2017_08.parquet and those schema is:

2017_01.parquet:

root
|-- value: struct (nullable = true)
|    |-- version: struct (nullable = true)
|    |    |-- major: integer (nullable = true)
|    |    |-- minor: integer (nullable = true)
|    |-- guid: string (nullable = true)

2017_08.parquet:

root
|-- value: struct (nullable = true)
|    |-- version: struct (nullable = true)
|    |    |-- major: integer (nullable = true)
|    |    |-- minor: integer (nullable = true)
|    |    |-- vnum: integer (nullable = true)
|    |-- guid: string (nullable = true)

and my code

SQL = """
SELECT value.version.major,
       value.version.minor,
       value.version.vnum
FROM OUT_TABLE 
LIMIT 10"""

parquetFile = spark.read.parquet("/mydata/2017_08.parquet")
parquetFile.createOrReplaceTempView("OUT_TABLE")
out_osce = spark.sql(SQL)
out_osce.show()

When I load 2017_08.parquet show:

+-----+-----+----+
|major|minor|vnum|
+-----+-----+----+
| 0001| 4610|1315|
| 0002| 4610|6206|
| 0003| 4610|6125|

but if I load 2017_01.parquet like parquetFile = spark.read.parquet("/mydata/2017_01.parquet")

SQL show error :

pyspark.sql.utils.AnalysisException: u'No such struct field vnum in major, minor; line 4 pos 11'

I know the reason is 2017_01.parquet didn't have vnum column, and I have two slove solutions, one is using mergeSchema another is using schema when reading parquet file, but those ways also have a big problem.

The first solution need read 2017_08.parquet, if I don't need 08's data it will be a problem, and if bad luck vnum is an option column and 08 didn't have this column it still error

The second solution is given schema when read, like spark.read.schema(schema).parquet("/mydata/2017_01.parquet"), this way need to write schema first, but if the file is a very complex nested table, user maybe can't write the schema, and schema will update.

I want ask anyone have the third solution then make only read 2017_01.parquet and output like:

+-----+-----+----+
|major|minor|vnum|
+-----+-----+----+
| 0001| 4600|null|
| 0002| 4600|null|
| 0003| 4600|null|
parquet pyspark apache-spark-sql apache-spark python
3 Answers

You can simply use case statement or coalesce when reading:

parquetFile = spark.read.parquet("") \
              .withColumn("vnum", coalesce("vnum"))

From docs:

coalesce(e: Column*): Column

Returns the first column that is not null, or null if all inputs are null.

If your Parquet file has this field, it will be used. If not, nulls will be used and new column will be in your schema

user_58118000

You can create a index table to store column situations for each parquet files, such as:

parquetfile1 column1, column2 parquetfile2 column1, column3 .....

when read parquet files, read index data firstly to filter some files, then execute the query operations.

user_15644870
Related