如何在spark scala中重命名S3文件而不是HDFS

2023-12-08

我在 S3 中存储了大约 100 万个文本文件。 我想根据文件夹名称重命名所有文件。

我怎样才能在 Spark-scala 中做到这一点?

我正在寻找一些示例代码。

我正在使用 zeppelin 来运行我的 Spark 脚本。

下面的代码我已经按照答案的建议进行了尝试

import org.apache.hadoop.fs._

val src = new Path("s3://trfsmallfffile/FinancialLineItem/MAIN")
val dest = new Path("s3://trfsmallfffile/FinancialLineItem/MAIN/dest")
val conf = sc.hadoopConfiguration   // assuming sc = spark context
val fs = Path.getFileSystem(conf)
fs.rename(src, dest)

但低于错误

<console>:110: error: value getFileSystem is not a member of object org.apache.hadoop.fs.Path
       val fs = Path.getFileSystem(conf)

您可以使用普通的 HDFS API,例如(输入,未测试)

val src = new Path("s3a://bucket/data/src")
val dest = new Path("s3a://bucket/data/dest")
val conf = sc.hadoopConfiguration   // assuming sc = spark context
val fs = src.getFileSystem(conf)
fs.rename(src, dest)

S3A客户端伪造重命名的方式是copy + delete每个文件的数量,因此所需的时间与文件数量和数据量成正比。 S3 会限制你:如果你尝试并行执行此操作,它可能会减慢你的速度。如果需要“一段时间”,请不要感到惊讶。

您还需要按每次 COPY 调用付费,每 1,000 次调用 0.005 美元,因此尝试费用约为 5 美元。在一个小目录上进行测试,直到确定一切正常

本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)

如何在spark scala中重命名S3文件而不是HDFS 的相关文章

随机推荐