问题标签 [s3distcp]

试图弄清楚如何从 Apache Spark Streaming 作业输出的 HDFS 导出数据。下图定义了解决方案架构：

Apache Spark 在 AWS EMR 集群中运行流作业并将结果存储在 HDFS 中。流式作业每小时使用窗口函数收集一次数据并执行计算。我需要将这些结果导出到 S3 和 RDS，这可以通过运行 S3Distcp 和 Sqoop 命令轻松完成，但是我希望在每次计算完成后准确运行这些结果。我想使用 cron 作业以外的其他东西更优雅地完成此操作。

有任何想法吗？

谢谢

我正在使用 s3distcp 将 500GB 数据集复制到我的 EMR 集群中。这是一个 12 节点 r4.4xlarge 集群，每个集群都有 750GB 磁盘。它使用 EMR 发布标签emr-5.13.0，我正在添加 Hadoop: Amazon 2.8.3、 Ganglia:3.7.2和 Spark 2.3.0。我正在使用以下命令将数据复制到集群中：

当我查看 Ganglia 或 namenode UI（EMR 集群上的端口 50070）中的磁盘使用情况时，我可以看到一个节点的大部分磁盘已填满，而其他节点的使用百分比相似。单击大量文件（〜50）我可以看到文件的副本始终出现在完整节点上。

我正在使用 Spark 转换这些数据，将其写入 HDFS，然后复制回 S3。我在处理这个数据集时遇到了问题，因为我的任务被杀死了。我不确定这是问题的原因。我不需要在本地复制数据，也不需要解压缩。最初我认为 BZIP2 编解码器不可拆分，解压缩有助于在我的 Spark 作业中获得并行性，但我错了，它是可拆分的。我还发现了hdfs balancer我用来重新分配副本的命令，看看这是否解决了我的 Spark 问题。

但是，现在我已经看到了我认为奇怪的行为，我想了解 s3distcp/HDFS 始终在一个节点上创建文件副本是否正常？

我正在将800 个 avro 文件，大小约为136 MB，从HDFS复制到EMR 集群上的S3，但我得到了这个异常：

EMR 集群的配置是：

任何帮助将不胜感激。

编辑：

运行hdfs dfsadmin -report命令，得到以下结果：

它表明该块丢失。这是否意味着我必须再次重新运行程序？如果我看到 Under replicated blocks 的输出，它会显示 126。这意味着将复制 126 个块。我怎么知道，它会复制丢失的块吗？

此外，在过去 30 分钟内， Under replicated blocks的值是 126。有没有办法强制它快速复制？

我有需要将文件从一个 S3 存储桶复制到另一个 S3 存储桶的要求。这些存储桶存在于不同的 AWS 账户中。我尝试使用 s3 sync 命令。但是，对于此目标 IAM 用户，应授予对源存储桶的读取访问权限。我没有使用此命令，因为我没有相同的 IAM 用户可以访问这两个帐户。目前有两个 AIM 用户，每个用户一个。

无论如何我可以使用 s3-dist-cp 命令来满足这个要求吗？

有人可以在将数据从 hdfs 移动到 S3 时帮助我进行身份验证。要连接到 S3，我正在使用 aws_key_gen（access_key、secret_key 和基于会话的令牌）生成基于会话的凭据

我测试过，distcp 可以很好地使用永久访问和密钥，但是使用基于会话的密钥会出现问题。此外，我使用 python 测试了基于会话的凭据，并且能够列出目录。

附件是代码（密钥和存储桶名称已更改）。

我想从 hadoop 将几个文件上传到 AWS 存储桶中。我有 AWS ACCESS KEY、SECRET KEY 和 S3 IMPORT PATH。

我无法通过 AWS CLI 命令访问。我在 aws 凭证文件中设置了密钥。我试图做“aws s3 ls”我收到错误

由于上面的代码不起作用，我尝试使用如下 distcp 命令。

对于上面的命令，我也遇到了错误。

18/11/09 00:55:40 信息 http.AmazonHttpClient：配置代理。代理主机：qypprdproxy02.ie.XXXX.net 代理端口：80 18/11/09 00:55:40 WARN s3a.S3AFileSystem：客户端：Amazon S3 错误 400：400 错误请求；错误请求（可重试）

com.cloudera.com.amazonaws.services.s3.model.AmazonS3Exception：错误请求（服务：Amazon S3；状态代码：400；错误代码：400 错误请求；请求 ID：121931CAB75C3BB0），S3 扩展请求 ID：jn/iTngZS83 +A5U8e2gjQsyArDC68E+r0q/Sll0gkSCn0h5yDaG17TEb9HNSx7o590hmofguJIg= 在 com.cloudera.com.amazonaws.http.AmazonHttpClient.handleErrorResponse(AmazonHttpClient.java:1182) 在 com.cloudera.com.amazonaws.http.AmazonHttpClient.executeOneRequest(AmazonHttpClient.java:770) 在com.cloudera.com.amazonaws.http.AmazonHttpClient.executeHelper(AmazonHttpClient.java:489) 在 com.cloudera.com.amazonaws.http.AmazonHttpClient.execute(AmazonHttpClient.java:310) 在 com.cloudera.com.amazonaws。 services.s3.AmazonS3Client.invoke(AmazonS3Client.java:3785) 在 com.cloudera.com.amazonaws.services.s3.AmazonS3Client。headBucket(AmazonS3Client.java:1107) at com.cloudera.com.amazonaws.services.s3.AmazonS3Client.doesBucketExist(AmazonS3Client.java:1070) at org.apache.hadoop.fs.s3a.S3AFileSystem.verifyBucketExists(S3AFileSystem.java: 312) 在 org.apache.hadoop.fs.FileSystem.createFileSystem(FileSystem.java:2815) 在 org.apache.hadoop.fs 的 org.apache.hadoop.fs.s3a.S3AFileSystem.initialize(S3AFileSystem.java:260) .FileSystem.access$200(FileSystem.java:98) at org.apache.hadoop.fs.FileSystem$Cache.getInternal(FileSystem.java:2852) at org.apache.hadoop.fs.FileSystem$Cache.get(FileSystem. java:2834) 在 org.apache.hadoop.fs.FileSystem.get(FileSystem.java:387) 在 org.apache.hadoop.fs.Path.getFileSystem(Path.java:296) 在 org.apache.hadoop.tools .DistCp.setTargetPathExists(DistCp.java:205) 在 org.apache.hadoop.tools.DistCp。在 org.apache.hadoop.util.ToolRunner.run(ToolRunner.java:70) 上运行(DistCp.java:131) 在 org.apache.hadoop.tools.DistCp.main(DistCp.java:441) 18/11/ 09 00:55:40 错误 tools.DistCp：无效参数：org.apache.hadoop.fs.s3a.AWSS3IOException：segmentor-data-import-dev 上的 dosBucketExist：com.cloudera.com.amazonaws.services.s3.model。 AmazonS3Exception：错误请求（服务：Amazon S3；状态代码：400；错误代码：400 错误请求；请求 ID：121931CAB75C3BB0），S3 扩展请求 ID：jn/iTngZS83+A5U8e2gjQsyArDC68E+r0q/Sll0gkSCn0h5yDaG17TEb9HNSx7o590hmofguJIg=：错误请求S3；状态代码：400；错误代码：400 错误请求；请求 ID：121931CAB75C3BB0) 在 org.apache.hadoop.fs.s3a.S3AUtils.translateException(S3AUtils.java:178) 在 org.apache.hadoop.fs.s3a .S3AFileSystem.verifyBucketExists(S3AFileSystem.java:318) 在 org.apache.hadoop.fs.s3a.S3AFileSystem.initialize(S3AFileSystem.java:260) 在 org.apache.hadoop.fs.FileSystem.createFileSystem(FileSystem.java:2815) 在 org.apache.hadoop .fs.FileSystem.access$200(FileSystem.java:98) at org.apache.hadoop.fs.FileSystem$Cache.getInternal(FileSystem.java:2852) at org.apache.hadoop.fs.FileSystem$Cache.get( FileSystem.java:2834) 在 org.apache.hadoop.fs.FileSystem.get(FileSystem.java:387) 在 org.apache.hadoop.fs.Path.getFileSystem(Path.java:296) 在 org.apache.hadoop .tools.DistCp.setTargetPathExists(DistCp.java:205) 在 org.apache.hadoop.tools.DistCp.run(DistCp.java:131) 在 org.apache.hadoop.util.ToolRunner.run(ToolRunner.java:70 ) 在 org.apache.hadoop.tools.DistCp.main(DistCp.java:441) 引起：com.cloudera.com.amazonaws.services.s3.model.AmazonS3Exception：错误请求（服务：Amazon S3；状态代码：400；错误代码：400 错误请求；请求 ID：121931CAB75C3BB0），S3 扩展请求 ID：jn/iTngZS83+A5U8e2gjQsyArDC68E+r0q/Sll0gkSCn0h5yDaG17TEb9HNSx7o590hmofguzonamaJIg= 在 com.cloudera.com。 http.AmazonHttpClient.handleErrorResponse(AmazonHttpClient.java:1182) 在 com.cloudera.com.amazonaws.http.AmazonHttpClient.executeOneRequest(AmazonHttpClient.java:770) 在 com.cloudera.com.amazonaws.http.AmazonHttpClient.executeHelper(AmazonHttpClient. java:489) 在 com.cloudera.com.amazonaws.http.AmazonHttpClient.execute(AmazonHttpClient.java:310) 在 com.cloudera.com.amazonaws.services.s3.AmazonS3Client.invoke(AmazonS3Client.java:3785) 在 com .cloudera.com.amazonaws.services.s3.AmazonS3Client.headBucket(AmazonS3Client.java:1107) 在 com.cloudera.com.amazonaws。services.s3.AmazonS3Client.doesBucketExist(AmazonS3Client.java:1070) at org.apache.hadoop.fs.s3a.S3AFileSystem.verifyBucketExists(S3AFileSystem.java:312) ... 还有 11 个无效参数：segmentor-data-import 上的 dosBucketExist -dev：com.cloudera.com.amazonaws.services.s3.model.AmazonS3Exception：错误请求（服务：Amazon S3；状态代码：400；错误代码：400 错误请求；请求 ID：121931CAB75C3BB0），S3 扩展请求 ID： jn/iTngZS83+A5U8e2gjQsyArDC68E+r0q/Sll0gkSCn0h5yDaG17TEb9HNSx7o590hmofguJIg=：错误请求（服务：Amazon S3；状态代码：400；错误代码：400 错误请求；请求 ID：121931CAB75C3BB0）用法：distcp OPTIONS [source_path ...] OPTIONS -append 重用目标文件中的现有数据，并在可能的情况下将新数据附加到它们 -async distcp 执行是否应该阻塞 -atomic 提交所有更改或不提交 -bandwidth 指定每个映射的带宽 (MB) -delete
从目标中删除，源中缺少文件 -diff
使用快照差异报告来识别源和目标之间的差异 -f 需要复制的文件列表 -filelimit（已弃用！）限制复制到 <= n 的文件数 -filters 包含字符串列表的文件的路径要从副本中排除的路径。-i 在复制期间忽略失败 -log DFS 上保存 distcp 执行日志的文件夹 -m 用于复制的最大并发映射数 -mapredSslConf 用于 ssl 配置文件的配置，与 hftps 一起使用：//。必须在类路径中。-numListstatusThreads 用于构建文件列表的线程数（最多 40 个）。-overwrite 选择无条件覆盖目标文件，即使它们存在。-p 保留状态 (rbugpcaxt)（复制、块大小、用户、组、权限、校验和类型、ACL、XATTR、时间戳）。如果 -p 指定为 no ，则保留复制、块大小、用户、组、权限、校验和类型和时间戳。生的。* 当源路径和目标路径都在 /.reserved/raw 层次结构中（仅限 HDFS）时，会保留 xattrs。raw.* xattrpreservation 独立于 -p 标志。有关详细信息，请参阅 DistCp 文档。-rdiff 使用目标快照差异报告来识别对目标所做的更改 -sizelimit（已弃用！）限制复制到 <= n 字节的文件数 -skipcrccheck 是否跳过源路径和目标路径之间的 CRC 检查。-strategy 要使用的复制策略。

请让我知道如何实现这一目标。

我有一个 jar 文件，它在 jar 中的方法中提供给 spark-submit.With 。我正在尝试做一个

我还与 master 一起在所有药膏上安装了 s3-dist-cp。应用程序启动并成功且没有错误，但不会将数据移动到 S3。

我正在尝试使用 distcp 命令将数据从一个 hdfs 集群复制到另一个集群。以下是我提交的命令

hadoop distcp hdfs://sourcenamenodehostname:50070/var/lib/hadoop-hdfs/distcptest.txt hdfs://destinationnamenodehostname:50070/var/lib/hadoop-hdfs

提交时我收到错误消息。请仔细阅读错误消息并指导我正确的方法。

19/02/27 04:28:19 INFO tools.OptionsParser: parseChunkSize: blocksperchunk false 19/02/27 04:28:20 ERROR tools.DistCp: Invalid arguments: org.apache.hadoop.ipc.RemoteException(org.apache .hadoop.ipc.StandbyException)：状态待机不支持操作类别 READ。访问https://s.apache.org/sbnn-error 在 org.apache.hadoop.hdfs.server.namenode.ha.StandbyState.checkOperation(StandbyState.java:88) 在 org.apache.hadoop.hdfs.server.namenode.NameNode$NameNodeHAContext.checkOperation(NameNode.java:1835)在 org.apache.hadoop.hdfs.server.namenode.FSNamesystem.checkOperation(FSNamesystem.java:1515) 在 org.apache.hadoop.hdfs.server.namenode.FSNamesystem.getFileInfo(FSNamesystem.java:4448) 在 org.apache .hadoop.hdfs.server.namenode.NameNodeRpcServer.getFileInfo(NameNodeRpcServer.java:912) 在 org.apache.hadoop.hdfs.server.namenode.AuthorizationProviderProxyClientProtocol.getFileInfo(AuthorizationProviderProxyClientProtocol.java:533) 在 org.apache.hadoop.hdfs .protocolPB.ClientNamenodeProtocolServerSideTranslatorPB.getFileInfo(ClientNamenodeProtocolServerSideTranslatorPB.java:862) 在 org.apache。hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos$ClientNamenodeProtocol$2.callBlockingMethod(ClientNamenodeProtocolProtos.java) at org.apache.hadoop.ipc.ProtobufRpcEngine$Server$ProtoBufRpcInvoker.call(ProtobufRpcEngine.java:617) at org.apache.hadoop.ipc .RPC$Server.call(RPC.java:1073) at org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:2281) at org.apache.hadoop.ipc.Server$Handler$1.run (Server.java:2277) 在 java.security.AccessController.doPrivileged(Native Method) 在 javax.security.auth.Subject.doAs(Subject.java:422) 在 org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation .java:1924) 在 org.apache.hadoop.ipc.Server$Handler.run(Server.java:2275)callBlockingMethod(ClientNamenodeProtocolProtos.java) at org.apache.hadoop.ipc.ProtobufRpcEngine$Server$ProtoBufRpcInvoker.call(ProtobufRpcEngine.java:617) at org.apache.hadoop.ipc.RPC$Server.call(RPC.java:1073)在 org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:2281) 在 org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:2277) 在 java.security.AccessController .doPrivileged(Native Method) at javax.security.auth.Subject.doAs(Subject.java:422) at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1924) at org.apache.hadoop.ipc .Server$Handler.run(Server.java:2275)callBlockingMethod(ClientNamenodeProtocolProtos.java) at org.apache.hadoop.ipc.ProtobufRpcEngine$Server$ProtoBufRpcInvoker.call(ProtobufRpcEngine.java:617) at org.apache.hadoop.ipc.RPC$Server.call(RPC.java:1073)在 org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:2281) 在 org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:2277) 在 java.security.AccessController .doPrivileged(Native Method) at javax.security.auth.Subject.doAs(Subject.java:422) at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1924) at org.apache.hadoop.ipc .Server$Handler.run(Server.java:2275)1073) 在 org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:2281) 在 org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:2277) 在 java.security .AccessController.doPrivileged(Native Method) at javax.security.auth.Subject.doAs(Subject.java:422) at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1924) at org.apache.hadoop .ipc.Server$Handler.run(Server.java:2275)1073) 在 org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:2281) 在 org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:2277) 在 java.security .AccessController.doPrivileged(Native Method) at javax.security.auth.Subject.doAs(Subject.java:422) at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1924) at org.apache.hadoop .ipc.Server$Handler.run(Server.java:2275)服务器$Handler.run(Server.java:2275)服务器$Handler.run(Server.java:2275)

无效参数：状态待机不支持操作类别 READ。访问https://s.apache.org/sbnn-error 在 org.apache.hadoop.hdfs.server.namenode.ha.StandbyState.checkOperation(StandbyState.java:88) 在 org.apache.hadoop.hdfs.server.namenode.NameNode$NameNodeHAContext.checkOperation(NameNode.java:1835)在 org.apache.hadoop.hdfs.server.namenode.FSNamesystem.checkOperation(FSNamesystem.java:1515) 在 org.apache.hadoop.hdfs.server.namenode.FSNamesystem.getFileInfo(FSNamesystem.java:4448) 在 org.apache .hadoop.hdfs.server.namenode.NameNodeRpcServer.getFileInfo(NameNodeRpcServer.java:912) 在 org.apache.hadoop.hdfs.server.namenode.AuthorizationProviderProxyClientProtocol.getFileInfo(AuthorizationProviderProxyClientProtocol.java:533) 在 org.apache.hadoop.hdfs .protocolPB.ClientNamenodeProtocolServerSideTranslatorPB.getFileInfo(ClientNamenodeProtocolServerSideTranslatorPB.java:862) 在 org.apache。hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos$ClientNamenodeProtocol$2.callBlockingMethod(ClientNamenodeProtocolProtos.java) at org.apache.hadoop.ipc.ProtobufRpcEngine$Server$ProtoBufRpcInvoker.call(ProtobufRpcEngine.java:617) at org.apache.hadoop.ipc .RPC$Server.call(RPC.java:1073) at org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:2281) at org.apache.hadoop.ipc.Server$Handler$1.run (Server.java:2277) 在 java.security.AccessController.doPrivileged(Native Method) 在 javax.security.auth.Subject.doAs(Subject.java:422) 在 org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation .java:1924) 在 org.apache.hadoop.ipc.Server$Handler.run(Server.java:2275)callBlockingMethod(ClientNamenodeProtocolProtos.java) at org.apache.hadoop.ipc.ProtobufRpcEngine$Server$ProtoBufRpcInvoker.call(ProtobufRpcEngine.java:617) at org.apache.hadoop.ipc.RPC$Server.call(RPC.java:1073)在 org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:2281) 在 org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:2277) 在 java.security.AccessController .doPrivileged(Native Method) at javax.security.auth.Subject.doAs(Subject.java:422) at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1924) at org.apache.hadoop.ipc .Server$Handler.run(Server.java:2275)callBlockingMethod(ClientNamenodeProtocolProtos.java) at org.apache.hadoop.ipc.ProtobufRpcEngine$Server$ProtoBufRpcInvoker.call(ProtobufRpcEngine.java:617) at org.apache.hadoop.ipc.RPC$Server.call(RPC.java:1073)在 org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:2281) 在 org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:2277) 在 java.security.AccessController .doPrivileged(Native Method) at javax.security.auth.Subject.doAs(Subject.java:422) at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1924) at org.apache.hadoop.ipc .Server$Handler.run(Server.java:2275)1073) 在 org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:2281) 在 org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:2277) 在 java.security .AccessController.doPrivileged(Native Method) at javax.security.auth.Subject.doAs(Subject.java:422) at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1924) at org.apache.hadoop .ipc.Server$Handler.run(Server.java:2275)1073) 在 org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:2281) 在 org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:2277) 在 java.security .AccessController.doPrivileged(Native Method) at javax.security.auth.Subject.doAs(Subject.java:422) at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1924) at org.apache.hadoop .ipc.Server$Handler.run(Server.java:2275)服务器$Handler.run(Server.java:2275)服务器$Handler.run(Server.java:2275)

在 EMR 上，我使用s3-dist-cp --groupBy它来将文件夹中的随机fileName文件命名为我希望在 S3 中将其重命名为的名称：

例子：

运行 s3-dist-cp 后，

但是，我想使用hadoop distcp命令在 Dataproc 上实现此功能并将文件写入 GCS 位置gs://testLocation/folder_in_gs/file.csv

任何帮助表示赞赏。

我正在尝试使用 s3-distcp 将数据从 EMR 集群复制到 S3。我可以将减速器的数量指定为大于默认值以加快我的流程吗？