scala – Spark：仅在路径存在时读取文件

发布时间：2020-12-16 19:13:50 所属栏目：安全来源：网络整理

导读：我试图读取 scala中的Path of Paths中出现的文件.下面是示例(伪)代码： val paths = Seq[String] //Seq of pathsval dataframe = spark.read.parquet(paths: _*) 现在,在上面的序列中,存在一些路径而一些路径不存在.有没有办法在读取镶木地板文件时忽略丢失

我试图读取 scala中的Path of Paths中出现的文件.下面是示例(伪)代码：

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

现在,在上面的序列中,存在一些路径而一些路径不存在.有没有办法在读取镶木地板文件时忽略丢失的路径(以避免org.apache.spark.sql.AnalysisException：Path不存在)？

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

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

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

此外,这些路径可以是hdfs或s3(此Seq作为方法参数传递)并且在读取时,我不知道路径是s3还是hdfs,因此不能使用s3或hdfs特定的API来检查是否存在.

解决方法

你可以像在@ 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: _*)

（编辑：李大同）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!