如何用Spark读取HDFS文件夹文本文件并从DataFrame获取各文件内容

你当前用spark.read.text()读入HDFS路径生成的DataFrame，默认仅包含value列存储每行文本内容，无内置字段标记每行内容所属的源文件，只要追加文件来源标记字段，就能实现单个文件内容的提取。

具体操作步骤 #

• 引入Spark内置的文件元数据函数

  import org.apache.spark.sql.functions.input_file_name
  

• 读取文件时追加来源文件路径列

  val dfWithFileTag = spark.read.text("hdfs://master:9000/user/file/shakespeare")
    .withColumn("source_file_path", input_file_name())
  

  该函数会自动为每一行数据打上所属源文件的完整HDFS路径标识，哪怕文件夹下嵌套多级子目录的txt文件也能准确区分归属。
• 按文件路径分组聚合，拿到每个文件对应的完整内容

  // 聚合后每一行对应一个源文件，file_content_lines字段存储该文件拆分后的所有文本行
  val perFileDf = dfWithFileTag
    .groupBy("source_file_path")
    .agg(collect_list("value").as("file_content_lines"))
  

  如果需要把文件所有行拼接成完整的整段文本，可把聚合逻辑替换为：

  .agg(concat_ws("\n", collect_list("value")).as("full_file_content"))
  

低版本Spark兼容方案 #

如果使用的Spark版本低于2.0，没有input_file_name()函数的支持，可以直接用SparkContext的wholeTextFiles接口读取数据，该接口返回的RDD天然以“文件路径-文件全量内容”为键值对结构：

val wholeFileRDD = spark.sparkContext.wholeTextFiles("hdfs://master:9000/user/file/shakespeare")
val perFileDf = wholeFileRDD.toDF("source_file_path", "full_file_content")

注意：wholeTextFiles是按单个文件为单位读取，不会将大文件拆分为多个分片并行处理，大文件场景下并行度较低，优先推荐使用前面加文件路径标记的逐行读取方案。

内容的提问来源于stack exchange，提问作者Truong-Giang Nguyen