Spark:仅在路径存在时读取文件

Question

我正在尝试阅读Sequencescala中Paths中的文件.下面是示例(伪)代码:

val paths = Seq[String] //Seq of paths
val dataframe = spark.read.parquet(paths: _*)

现在,在上面的序列中,存在一些路径而一些路径不存在.有没有办法在读取parquet文件时忽略丢失的路径(避免org.apache.spark.sql.AnalysisException: Path does not exist)？

我已经尝试了以下它似乎工作,但是,然后,我最终读取相同的路径两次,这是我想避免做的事情:

val filteredPaths = paths.filter(p => Try(spark.read.parquet(p)).isSuccess)

我检查了options方法,DataFrameReader但似乎没有任何类似的选项ignore_if_missing.

此外,这些路径可以是hdfs或s3(这Seq是作为一个方法参数传递),并一边读书,我不知道一个路径是s3或hdfs因此无法使用s3或hdfs特定的API,以检查是否存在.

Answer 1

您可以像在@ Psidom的答案中过滤掉不相关的文件.在spark中,最好的方法是使用内部spark hadoop配置.鉴于spark session变量被称为"spark",你可以这样做:

import org.apache.hadoop.fs.FileSystem
import org.apache.hadoop.fs.Path

val hadoopfs: FileSystem = FileSystem.get(spark.sparkContext.hadoopConfiguration)

def testDirExist(path: String): Boolean = {
  val p = new Path(path)
  hadoopfs.exists(p) && hadoopfs.getFileStatus(p).isDirectory
}
val filteredPaths = paths.filter(p => testDirExists(p))
val dataframe = spark.read.parquet(filteredPaths: _*)

Answer 2

过滤第一个怎么样paths：

paths.filter(f => new java.io.File(f).exists)

例如：

Seq("/tmp", "xx").filter(f => new java.io.File(f).exists)
// res18: List[String] = List(/tmp)