包含子文件夹的文件夹顶部的分区表，其中包含火花中的 json 文件

Posted 2023-04-17

【中文标题】包含子文件夹的文件夹顶部的分区表，其中包含火花中的 json 文件

【英文标题】：Partition Table on top of folders containing sub-folders which contains json files in spark

【发布时间】：2021-06-07 18:54:48

【问题描述】：

我正在 Databricks 中研究 spark。我的存储位置有一个挂载点，指向我的目录。让我们将目录称为“/mnt/abc1/abc2” - 路径。 在这个“abc2”目录中，假设我有 10 个文件夹，名为“xyz1”..“xyz10”。所有这些“xyz%”文件夹都包含 json 文件，我们称它们为“xyz1_1.json”，依此类推。 我需要构建一个表，以便我可以通过将其引用为路径 +“abc2.xyz1.xyz1_1.json”来访问我的 json 到 spark 表中

var path = "/mnt/abc1/"
var data = spark.read.json(path)

当 json 文件直接位于路径内而不是路径中的文件夹内时，此方法有效。 我想找出一种方法，可以自动检测底层文件夹和包含 jsons 的子文件夹，并在其上构建表。

【参考方案1】：

使用 spark 3+ 您可以将选项 recursiveFileLookup 添加为 true 以搜索子目录

var path = "/mnt/abc1/"
var data = spark.read.option("recursiveFileLookup","true").json(path)

【参考方案2】：

试试下面的代码。

import org.apache.hadoop.fs.FileSystem, Path, RemoteIterator
import org.apache.spark.sql.SparkSession
import scala.util.Failure, Success, Try

case class Hdfs(fs: FileSystem) 
  implicit def convertToScalaIterator[T](remoteIterator: RemoteIterator[T]): Iterator[T] = 
    case class wrapper(remoteIterator: RemoteIterator[T]) extends Iterator[T] 
      override def hasNext: Boolean = remoteIterator.hasNext
      override def next(): T = remoteIterator.next()
    
    wrapper(remoteIterator)
  
  def listFiles(path: String): List[String] = 
    Try(
      fs
        .listFiles(new Path(path), true)
        .toList
        .map(_.getPath)
        .filter(!_.toString.contains("_spark_metadata"))
        .map(_.toString)
    ) match 
      case Success(files) => files
      case Failure(ex) => Nil
    
  

使用 spark session 获取 hdfs 对象。

val hdfs = Hdfs(FileSystem.get(spark.sparkContext.hadoopConfiguration))

使用listFiles函数递归获取文件列表。

val files = hdfs.listFiles("/mnt/abc1/")

检查文件在 hdfs 路径中是否可用。

if(!files.isEmpty) val data = spark.read.json(files:_*)