Spark-scala：在读取之前检查 S3 目录是否存在答案

【问题标题】：Spark-scala : Check whether a S3 directory exists or not before reading itSpark-scala：在读取之前检查 S3 目录是否存在
【发布时间】：2018-04-18 17:01:20
【问题描述】：

如何在读取之前检查 S3 目录是否存在？

我正在尝试这个，这里给出http://bigdatatech.taleia.software/2015/12/21/check-if-exists-a-amazon-s3-path-from-apache-spark/

import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.hadoop.conf.Configuration
val fs = FileSystem.get(new Configuration())
fs.listStatus(new Path("s3://s3bucket/2017/10/31/*/*/"))

但收到此错误 Wrong FS: s3://s3bucket/2017/10/31/*/*, expected: hdfs://ip-172-31-55-167.ec2.internal:8020

我可以使用它的确切路径检查确切的文件是否存在，但我必须使用它不支持的通配符，例如“s3://s3bucket/2017/10/31/ */ *”。

检查了这个 StackOverflow 问题：Spark : Read file only if the path exists，但它与我的用例无关。

【问题讨论】：

使用fs.globStatus
也可以试试FileSystem.get(new java.net.URI("s3://s3bucket") ) 或类似的东西。我也相信 Hadoop 提供了一个专用的 API 来处理 S3 FS
试过了，还是一样的错误java.lang.IllegalArgumentException: Wrong FS: s3://myS3Bucket/2017/10/31, expected: hdfs://ip-172-31-55-167.ec2.internal:8020

标签： scala amazon-web-services apache-spark amazon-s3

【解决方案1】：

这是因为当您调用FileSystem.get(new Configuration()) 时，解析的文件系统是默认文件系统，在本例中为hdfs。

您首先需要通过从包含s3 方案和您的存储桶的路径提供一个 URI 来获取正确的文件系统。

最好使用当前 Spark 上下文的 Hadoop 配置，以确保使用与 Spark 相同的文件系统设置。

import java.net.URI
import org.apache.hadoop.fs.FileSystem
import org.apache.spark.sql.SparkSession

// Create a session with Spark master running locally (vs on a cluster)
// for development purposes
val sparkSession = SparkSession.builder
                               .appName("My App")
                               .master("local")
                               .getOrCreate

val conf = sparkSession.sparkContext.hadoopConfiguration
val fs = FileSystem.get(URI.create("s3://s3bucket/"), conf)

仅当您的类路径上有 S3 文件系统的实现时，这才有效。我可以看到您正在 EC2 实例上运行您的程序，因此您应该能够在不指定显式 AWS 凭证的情况下访问 S3。

【讨论】：

URI 的导入指令是什么？
import java.net.URI
还有FileSystem?
import org.apache.hadoop.fs.FileSystem
@Narfanator 我明白了。我在会话生成器中添加了一行来创建一个简单的本地会话。请注意，在生产环境中，您的会话配置会有所不同，并且很可能不会选择让主节点在客户端进程上运行。

【解决方案2】：

import org.apache.hadoop.fs.FileSystem
import org.apache.hadoop.fs.Path
import org.apache.spark.SparkContext
import java.net.URI

var sc = new SparkContext()
if(FileSystem.get(new URI(externalTableLocation), sc.hadoopConfiguration).exists(new Path(externalTableLocation)))
{
  println("File exists")
  val maxPopulatedDate = spark.sql(s"SELECT MAX(DateID) FROM tier_ppw.DistributionDailyFact_Stage")
      log.info("Reading maxPopulatedDate: "+maxPopulatedDate)
}
else
{
   val maxPopulatedDate = "2016-01-01"
   log.info("Reading maxPopulatedDate: "+maxPopulatedDate)
}

【讨论】：

【解决方案3】：

S3 是基于对象的存储，不支持这种语法。 /*/*。您还必须将 s3 文件系统设置为默认值，并检查集群是否可以访问存储桶。

【讨论】：

我至少可以知道一个特定的目录是否存在吗？或者有什么替代方法，我可以用spark.read.json(path) 尝试捕捉，但我不想做两次。
FileStatus status = fs.getFileStatus(path)。 FileNotFoundException => 没有。 status.isDirectory() => 它在那里和一个目录。其他任何东西（例如isFile()，它在那里但不是目录）

【解决方案4】：

这很简单。例如，如果您需要检查名为Testfolder 的目录是否存在，请使用以下代码。

val s3login = "s3a://Accesskey:Secretkey@Bucket"
val path = "/Myfolder/Testfolder"    
if(FileSystem.get(new java.net.URI(s3login + path), sc.hadoopConfiguration).exists(new Path(s3login + path)))
{
    println("Directory exists")
}

【讨论】：