scala - AWS S3 中的 FileUtil.copyMerge()

Question

我已经使用以下代码将 DataFrame 作为 text 格式加载到 HDFS 中。 finalDataFrame 是 DataFrame

finalDataFrame.repartition(1).rdd.saveAsTextFile(targetFile)

执行上面的代码后，我发现用我提供的文件名创建了一个目录，并在该目录下创建了一个文件，但不是文本格式。文件名类似于 part-00000。

我已经使用下面的代码在 HDFS 中解决了这个问题。

val hadoopConfig = new Configuration()
val hdfs = FileSystem.get(hadoopConfig)
FileUtil.copyMerge(hdfs, new Path(srcPath), hdfs, new Path(dstPath), true, hadoopConfig, null)

现在我可以在上述路径中获取具有给定文件名的文本文件。

但是当我尝试在 S3 中执行相同的操作时，它显示出一些异常

FileUtil.copyMerge(hdfs, new Path(srcPath), hdfs, new Path(dstPath), true, hadoopConfig, null)

java.lang.IllegalArgumentException: Wrong FS:
s3a://globalhadoop/data, expected:
hdfs://*********.aws.*****.com:8050

这里好像不支持S3路径。任何人都可以协助如何做到这一点。

Answer 1

我已经使用下面的代码解决了这个问题。

def createOutputTextFile(srcPath: String, dstPath: String, s3BucketPath: String): Unit = {
    var fileSystem: FileSystem = null
    var conf: Configuration = null
    if (srcPath.toLowerCase().contains("s3a") || srcPath.toLowerCase().contains("s3n")) {
      conf = sc.hadoopConfiguration
      fileSystem = FileSystem.get(new URI(s3BucketPath), conf)
    } else {
      conf = new Configuration()
      fileSystem = FileSystem.get(conf)
    }
    FileUtil.copyMerge(fileSystem, new Path(srcPath), fileSystem, new Path(dstPath), true, conf, null)
  }

我已经为 S3 和 HDFS 的文件系统编写了代码，两者都运行良好。