Spark读取Schema动态变化的Parquet数据的问题处理

你遇到的是Parquet数据中字段类型不一致（字符串与字符串数组交替）导致的Spark读取异常问题，以下是几种可行的解决思路：

1. 手动指定统一Schema，强制类型转换 #

先定义一个统一的目标Schema，读取时强制使用该Schema，再对数据进行适配转换：

统一为数组类型 #

如果业务逻辑需要数组格式，可将所有字段定义为字符串数组，读取后把单字符串转为单元素数组：

import org.apache.spark.sql.types.{StructType, StructField, StringType, ArrayType}
import org.apache.spark.sql.functions._

// 定义统一Schema
val unifiedSchema = new StructType()
  .add("EMPID", ArrayType(StringType, containsNull = true))
  .add("EMPNAME", ArrayType(StringType, containsNull = true))
  .add("CNTRY", ArrayType(StringType, containsNull = true))

// 按指定Schema读取数据
val df = spark.read.schema(unifiedSchema).parquet("/path/to/parquet")

// 适配单字符串数据，转为单元素数组
val processedDf = df.select(
  when(size(col("EMPID")) === 0, array(col("EMPID"))).otherwise(col("EMPID")).alias("EMPID"),
  when(size(col("EMPNAME")) === 0, array(col("EMPNAME"))).otherwise(col("EMPNAME")).alias("EMPNAME"),
  when(size(col("CNTRY")) === 0, array(col("CNTRY"))).otherwise(col("CNTRY")).alias("CNTRY")
)

统一为字符串类型 #

如果需要字符串格式，可将字段定义为字符串，读取后把数组元素拼接为字符串：

import org.apache.spark.sql.types.{StructType, StructField, StringType}
import org.apache.spark.sql.functions._

val unifiedSchema = new StructType()
  .add("EMPID", StringType)
  .add("EMPNAME", StringType)
  .add("CNTRY", StringType)

val df = spark.read.schema(unifiedSchema).parquet("/path/to/parquet")

// 适配数组数据，拼接为逗号分隔的字符串
val processedDf = df.select(
  when(col("EMPID").isArray, concat_ws(",", col("EMPID"))).otherwise(col("EMPID")).alias("EMPID"),
  when(col("EMPNAME").isArray, concat_ws(",", col("EMPNAME"))).otherwise(col("EMPNAME")).alias("EMPNAME"),
  when(col("CNTRY").isArray, concat_ws(",", col("CNTRY"))).otherwise(col("CNTRY")).alias("CNTRY")
)

2. 读取为通用类型后动态转换 #

先不指定Schema读取原始数据，再通过Spark的类型判断函数动态转换字段类型：

import org.apache.spark.sql.functions._

// 读取原始数据，不指定Schema
val rawDf = spark.read.parquet("/path/to/parquet")

// 遍历所有字段，统一转换为数组类型
val processedDf = rawDf.select(rawDf.columns.map { colName =>
  when(col(colName).isArray, col(colName))
    .otherwise(array(col(colName))).alias(colName)
}: _*)

3. 利用mergeSchema选项合并Schema（谨慎使用） #

设置mergeSchema=true让Spark自动合并不同分区的Schema，但此时字段类型会变为AnyType，需要后续手动转换：

import org.apache.spark.sql.functions._

val df = spark.read.option("mergeSchema", "true").parquet("/path/to/parquet")

// 转换AnyType字段为统一的数组类型
val processedDf = df.select(
  when(col("EMPID").isArray, col("EMPID"))
    .otherwise(array(col("EMPID"))).alias("EMPID"),
  when(col("EMPNAME").isArray, col("EMPNAME"))
    .otherwise(array(col("EMPNAME"))).alias("EMPNAME"),
  when(col("CNTRY").isArray, col("CNTRY"))
    .otherwise(array(col("CNTRY"))).alias("CNTRY")
)

内容的提问来源于stack exchange，提问作者Sr4