Spark DataFrame处理Byte数组：解压GZIP内容并存储到HDFS

嘿，我看你正在处理Spark DataFrame里的GZIP压缩内容，想要解压后存储到HDFS对吧？先帮你指出代码里的一个关键问题：你用origRow.getString(0)来获取filecontent是错误的——因为这个字段是binary类型，直接转字符串会破坏压缩数据的结构，应该用getAs[Array[Byte]]来获取原始的字节数组才行。另外，你提到的serialise函数其实是多余的，直接处理字节数组就可以完成解压。

下面给你一套完整的、可运行的解决方案，用Spark的DataFrame API结合UDF和HDFS文件操作来实现：

1. 导入必要的依赖包 #

首先要引入Spark和Java IO/压缩相关的包：

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._
import java.util.zip.GZIPInputStream
import java.io.{ByteArrayInputStream, BufferedReader, InputStreamReader}
import org.apache.hadoop.fs.{FileSystem, Path}
import java.io.OutputStreamWriter

2. 定义解压UDF #

先写一个安全的解压函数，把GZIP压缩的字节数组转成文本内容（如果你的解压结果是二进制数据，可以修改函数返回Array[Byte]）：

def gzipDecompressSafe(bytes: Array[Byte]): Option[String] = {
  try {
    val bis = new ByteArrayInputStream(bytes)
    val gzis = new GZIPInputStream(bis)
    val reader = new BufferedReader(new InputStreamReader(gzis))
    // 读取所有行并拼接成完整文本
    val content = Iterator.continually(reader.readLine()).takeWhile(_ != null).mkString("\n")
    // 关闭流资源
    reader.close()
    gzis.close()
    bis.close()
    Some(content)
  } catch {
    case e: Exception =>
      println(s"解压失败，文件名可能有问题：${e.getMessage}")
      None
  }
}

// 将函数注册为Spark UDF，方便在DataFrame中调用
val decompressUdf = udf(gzipDecompressSafe _)

3. 处理DataFrame生成解压内容 #

用UDF给原DataFrame添加一个解压后的内容列，同时过滤掉解压失败的记录：

// 添加解压列
val dfWithUnzipped = df1.withColumn("unzipped_content", decompressUdf(col("filecontent")))
// 过滤解压失败的行
val validDf = dfWithUnzipped.filter(col("unzipped_content").isNotNull)

4. 将解压内容按文件名保存到HDFS #

如果需要把每条记录的解压内容保存为HDFS上的单独文件，推荐用foreachPartition批量处理，避免频繁创建HDFS连接：

// 获取HDFS文件系统实例
val fs = FileSystem.get(spark.sparkContext.hadoopConfiguration)

validDf.foreachPartition { partition =>
  // 每个分区只初始化一次资源，提升性能
  partition.foreach { row =>
    val filename = row.getAs[String]("filename")
    val unzippedContent = row.getAs[String]("unzipped_content")
    // 定义HDFS输出路径，替换成你的实际路径
    val outputPath = new Path(s"/your/hdfs/output/directory/$filename")
    // 写入文件到HDFS
    val os = fs.create(outputPath)
    val writer = new OutputStreamWriter(os)
    writer.write(unzippedContent)
    // 关闭流
    writer.close()
    os.close()
  }
}

额外注意事项 #

• 文件名重复处理：如果DataFrame里有重复的filename，HDFS默认会覆盖已有文件，你可以给文件名加后缀（比如拼接时间戳）来避免覆盖。
• 二进制内容适配：如果解压后是二进制数据，把UDF的返回类型改成Option[Array[Byte]]，写入时用os.write(unzippedBytes)即可。
• 性能优化：foreachPartition比直接用foreach更高效，因为每个分区只建立一次HDFS连接，减少资源开销。

内容的提问来源于stack exchange，提问作者K S Nidhin