在基于目录的火花负载中避免“路径不存在”

Question

我正在使用通配符从一个目录加载多个文件，如下所示 -

val df: DataFrame = spark.read
          .format("csv")
          .option("delimiter", ",")
          .schema(schema)
          .load(inputPath + "/*.csv*")

这在大多数情况下都很好用。但是当 inputPath 中没有任何 csv 文件时，我得到 -

org.apache.spark.sql.AnalysisException: Path does not exist

有没有办法避免这个错误，以便我们在 csv 文件存在时加载，但在没有要加载的情况下不出错？

Answer 1

您可以将其放入 try 块和 catch 异常中

try {
       val df: DataFrame = spark.read
          .format("csv")
          .option("delimiter", ",")
          .schema(schema)
          .load(inputPath + "/*.csv*")
    }catch (Exception e) {
    print("Do something else here")
    e.getMessage();
  }

如果你想检查 csv 是否存在，你可以先检查文件是否存在

import java.nio.file.{Paths, Files}
exist = Files.exists(Paths.get(inputPath +  "/*.csv*"))
if (exist){
 val df: DataFrame = spark.read
          .format("csv")
          .option("delimiter", ",")
          .schema(schema)
          .load(inputPath + "/*.csv*")

 }

例如，如果您在inputPaths 中有多个路径，则可以将它们过滤为

inputPaths.filter(f => Files.exists(Paths.get(f +  "/*.csv*")))

对于hdfs文件系统，可以将上面的逻辑替换为

单个文件

val conf = sc.hadoopConfiguration
val fs = org.apache.hadoop.fs.FileSystem.get(conf)
val exists = fs.exists(new org.apache.hadoop.fs.Path(inputPath +  "/*.csv*"))
if (exist){
     val df: DataFrame = spark.read
              .format("csv")
              .option("delimiter", ",")
              .schema(schema)
              .load(inputPath + "/*.csv*")
    
     }

对于存储在数组中的多个文件位置。

val conf = sc.hadoopConfiguration
val fs = org.apache.hadoop.fs.FileSystem.get(conf)


inputPaths.filter(f => fs.exists(new org.apache.hadoop.fs.Path(f +  "/*.csv*")))