如何在Scala中不使用Spark替换Parquet指定列的符合条件值为Null

要实现不依赖Spark的Parquet文件修改，我们可以直接使用Apache Parquet的原生Scala/Java API（基于Hadoop文件系统API）完成读取、修改、写入的全流程。以下是具体实现步骤和代码：

1. 添加依赖 #

首先在你的SBT项目中添加必要的Parquet和Hadoop依赖（版本可根据实际需求调整）：

libraryDependencies ++= Seq(
  "org.apache.parquet" % "parquet-hadoop" % "1.14.0",
  "org.apache.parquet" % "parquet-scala" % "1.14.0",
  "org.apache.hadoop" % "hadoop-common" % "3.3.6"
)

2. 完整实现代码 #

import org.apache.parquet.generic.GenericRecord
import org.apache.parquet.hadoop.{ParquetReader, ParquetWriter}
import org.apache.parquet.hadoop.read.GenericParquetReader
import org.apache.parquet.hadoop.write.GenericParquetWriter
import org.apache.parquet.schema.MessageType
import org.apache.hadoop.fs.Path

object ParquetNullUpdater {
  def main(args: Array[String]): Unit = {
    // 配置参数
    val parquetPath = new Path("path/to/my.parquet")
    val nullableColumns = Seq("column1", "column2")
    val targetId = "your_target_id" // 替换为实际需要匹配的ID值
    val idColumn = "id_column"

    // 读取原Parquet文件的Schema和所有记录
    val reader = GenericParquetReader.builder(parquetPath).build()
    val schema: MessageType = reader.getFooter.getFileMetaData.getSchema
    val records = collection.mutable.ArrayBuffer[GenericRecord]()
    
    var currentRecord: GenericRecord = null
    while ({currentRecord = reader.read(); currentRecord != null}) {
      records += currentRecord
    }
    reader.close()

    // 遍历并修改符合条件的记录
    val updatedRecords = records.map { record =>
      val currentId = record.get(idColumn)
      if (currentId != null && currentId == targetId) {
        // 创建新记录，复制原数据并将指定列设为Null
        val updatedRecord = new GenericRecord(schema)
        schema.getFields.forEach { field =>
          val fieldName = field.getName
          updatedRecord.put(
            fieldName,
            if (nullableColumns.contains(fieldName)) null else record.get(fieldName)
          )
        }
        updatedRecord
      } else {
        // 不符合条件，直接保留原记录
        record
      }
    }

    // 覆盖写入原文件（先删除原文件，再写入新数据）
    val fs = parquetPath.getFileSystem(new org.apache.hadoop.conf.Configuration())
    if (fs.exists(parquetPath)) {
      fs.delete(parquetPath, true)
    }

    val writer = GenericParquetWriter.builder(parquetPath)
      .withSchema(schema)
      .withConf(new org.apache.hadoop.conf.Configuration())
      .build()
    
    updatedRecords.foreach(writer.write)
    writer.close()
  }
}

3. 关键注意事项 #

• 内存限制：该方案会将所有Parquet记录加载到内存中处理，仅适合小到中等大小的文件。如果是超大文件，需要实现分批读取/写入的逻辑，避免内存溢出。
• Schema兼容性：使用GenericRecord可以兼容任意结构的Parquet文件，包括嵌套类型、数组等复杂Schema。如果你的文件是基于自定义case class生成的，也可以改用SpecificRecord来获得类型安全的操作。
• 文件覆盖逻辑：由于Parquet文件不支持原地修改，必须先读取所有数据到内存，删除原文件后再写入修改后的内容。
• 版本兼容：确保Parquet和Hadoop的版本匹配，避免出现依赖冲突或API兼容问题。

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