Spark异构JSON按类型分区写入Parquet的效率优化方案咨询

最优方案1：单次遍历RDD + 自定义多路径输出（性能最高） #

该方案仅需扫描一次原始RDD，完全避免了N次全量遍历的开销。你可以直接使用Hadoop的MultipleOutputs实现按类型输出到不同Parquet路径，不需要多次转换RDD。
代码示例：

import org.apache.hadoop.mapred.lib.MultipleOutputFormat
import org.apache.parquet.hadoop.ParquetOutputFormat
import org.apache.spark.sql.Row
import org.apache.spark.sql.types.StructType

// 预加载所有类型对应的Schema和Row构建函数，避免重复计算
val typeSchemaMap: Map[Type, (StructType, JSON => Row)] = types.map { t =>
  val schema = getSchema(t)
  t -> (schema, buildRow(schema))
}.toMap

// 自定义多输出格式，用Type作为输出路径的分区键
class TypePartitionParquetOutputFormat extends MultipleOutputFormat[Type, Row] {
  override def generateFileNameForKeyValue(key: Type, value: Row, name: String): String = {
    s"type=${key.toString}/part-$name"
  }
}

// 单次遍历转换所有数据直接输出
rdd.flatMap { case (t, json) =>
  // 可在此处添加脏数据过滤逻辑，过滤解析失败的记录
  typeSchemaMap.get(t).map { case (_, buildFunc) => (t, buildFunc(json)) }
}.saveAsHadoopFile(
  path = path,
  keyClass = classOf[Type],
  valueClass = classOf[Row],
  outputFormatClass = classOf[TypePartitionParquetOutputFormat],
  conf = spark.sparkContext.hadoopConfiguration
)

方案2：合并Schema统一输出DataFrame（适配Spark SQL API习惯） #

之前合并DataFrame失败是因为没有对齐Schema，你可以先合并所有类型的StructType，缺失字段默认设为nullable=true，所有类型的Row都可以适配统一的全局Schema：

import org.apache.spark.sql.Row

// 第一步：合并所有类型的Schema得到全局统一Schema
val globalSchema = types.map(getSchema).foldLeft(new StructType()) { (acc, schema) =>
  schema.foldLeft(acc) { (innerAcc, field) =>
    if (innerAcc.fieldNames.contains(field.name)) innerAcc
    // 新增字段统一设为nullable，适配缺失该字段的类型
    else innerAcc.add(field.copy(nullable = true))
  }
}.add("type", "string", nullable = false) // 新增分区字段

// 第二步：单次遍历转换所有数据为对齐全局Schema的Row
val allRows = rdd.map { case (t, json) =>
  val (schema, buildFunc) = typeSchemaMap(t)
  val rawRow = buildFunc(json)
  // 对齐到全局Schema，缺失字段填充null
  val fullValues = globalSchema.map { field =>
    if (field.name == "type") t.toString
    else if (schema.fieldNames.contains(field.name)) rawRow.getAs(field.name)
    else null
  }
  Row.fromSeq(fullValues)
}

// 第三步：生成DataFrame按type动态分区写入，一次完成全量输出
spark.createDataFrame(allRows, globalSchema)
  .write
  .partitionBy("type")
  .parquet(path)

注意：如果不同类型的同名字段类型冲突，需要提前做类型兼容处理（例如统一转成String或者更宽的数值类型），避免后续转换报错。

原有写法的低成本优化 #

如果不想调整整体逻辑，你可以通过持久化+主动释放资源的方式提升性能、降低内存占用：

• 循环前先持久化原始RDD，避免每次filter都重新计算上游数据
• 每个类型写完后主动调用unpersist释放中间资源
  代码示例：

// 循环前先持久化原始RDD
rdd.persist()

types.foreach { jsonType =>
  val sparkSchema: StructType = getSchema(jsonType)
  val rows: RDD[Row] = rdd
    .filter(k => k == jsonType)
    .map { case (_, json) => buildRow(sparkSchema)(json) }
    .persist() // 持久化中间RDD，避免写过程中重复计算
  val df = spark.createDataFrame(rows, sparkSchema)
  df.write.parquet(s"$path/type=$jsonType")
  // 写完立即释放当前批次的资源
  df.unpersist()
  rows.unpersist()
}

// 全量写入完成后释放原始RDD
rdd.unpersist()

该优化可以将原有写法的耗时降低到原来的1/N左右（N为类型数量）。

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