列出电子病历上的S3文件夹

Question

我不明白如何在spark作业期间简单地在电子病历上列出S3存储桶的内容。我想做以下几件事

Configuration conf = spark.sparkContext().hadoopConfiguration();
FileSystem s3 = S3FileSystem.get(conf);
List<LocatedFileStatus> list = toList(s3.listFiles(new Path("s3://mybucket"), false))

此操作始终失败，并显示以下错误

java.lang.IllegalArgumentException: Wrong FS: s3://*********/, expected: hdfs://**********.eu-central-1.compute.internal:8020

在hadoopConfiguration fs.defaultFS -> hdfs://**********.eu-central-1.compute.internal:8020中

根据我的理解，如果我不使用协议，只使用/myfolder/myfile，而不是hdfs://myfolder/myfile，它将默认使用df.defaultFS。但是，如果我指定了s3://mybucket/，那么fs.defaultFS应该无关紧要。

如何访问目录信息？spark.read.parquet("s3://mybucket/*.parquet")工作得很好，但是对于这个任务，我需要检查一些文件是否存在，并希望删除一些文件。我认为org.apache.hadoop.fs.FileSystem会是正确的工具。

PS:我也不明白日志是如何工作的。如果我使用部署模式集群(我想从在客户机模式下不工作的s3部署jars )，我只能在s3://logbucket/j-.../containers/application.../conatiner...0001.中找到我的日志在这些在S3中展示之前，有相当长的延迟。如何在主机上通过ssh找到它？或者有没有更快/更好的方法来检查spark应用程序日志？更新:我在/mnt/var/log/hadoop-yarn/containers下找到了它们，但是它属于yarn:yarn，作为hadoop用户，我无法阅读它。:(想法？

Answer 1

我不认为您正确地选择了FS；只需使用静态FileSystem.get()方法或Path.get()

尝试如下所示：Path p = new Path("s3://bucket/subdir"); FileSystem fs = p.get(conf); FileStatus[] status= fs.listStatus(p);

关于日志，YARN UI应该允许您通过节点管理器查看日志。

Answer 2

在我的示例中，我需要读取由以前的电子病历作业生成的parquet文件，我需要查找给定s3前缀的文件列表，但好消息是我们不需要执行所有这些操作，我们只需执行以下操作: spark.read.parquet(bucket+prefix_directory)

Answer 3

应该使用URI.create()将其指向正确的文件系统。

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

val dirPaths = FileSystem.get(URI.create("<s3-path>"), fs.getConf).listStatus(new Path("<s3-path>"))```