标签: elastic-map-reduce

1) 有人告诉我，在 EMR 上安装了 git。这是真的 ？我不相信，因为我可以确认在我的 elastic-mapreduce ssh 终端中找不到“git”。参见：https : //raw.github.com/gist/3177009/6128554a2668367b72758be74885836e41898923/gistfile1.txt。

2）我需要在 EMR 上安装 git，但以下行：

hadoop@ip-10-235-11-52:/mnt$ sudo apt-get install git --assume-yes

结果是 ：

hadoop@ip-10-235-11-52:/mnt$ sudo apt-get install git --assume-yes 读取包列表...完成 构建依赖树
读取状态信息...完成 git已经是最新版本。0 升级，0 新安装，0 删除，3 未升级。

然而不幸的是，即使在发出以下命令后，git 也不会在此 EMR 实例上运行：

hadoop@ip-10-235-11-52:/mnt$ git -bash: git: 命令未找到 hadoop@ip-10-235-11-52:/mnt$ sudo git sudo: git: 命令未找到

我有一个EMR流媒体作业(Python)通常工作正常(例如10台机器处理200个输入).但是,当我针对大型数据集(12台机器处理总共6000个输入,每个输入大约20秒)运行它时,经过2.5小时的运算后,我得到以下错误:

java.lang.RuntimeException: PipeMapRed.waitOutputThreads(): subprocess failed with code 143
at org.apache.hadoop.streaming.PipeMapRed.waitOutputThreads(PipeMapRed.java:372)
at org.apache.hadoop.streaming.PipeMapRed.mapRedFinished(PipeMapRed.java:586)
at org.apache.hadoop.streaming.PipeMapper.close(PipeMapper.java:135)
at org.apache.hadoop.mapred.MapRunner.run(MapRunner.java:57)
at org.apache.hadoop.streaming.PipeMapRunner.run(PipeMapRunner.java:36)
at org.apache.hadoop.mapred.MapTask.runOldMapper(MapTask.java:441)
at org.apache.hadoop.mapred.MapTask.run(MapTask.java:377)
at org.apache.hadoop.mapred.Child$4.run(Child.java:255)
at java.security.AccessController.doPrivileged(Native Method)
at javax.security.auth.Subject.doAs(Subject.java:396)
at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1132)
at org.apache.hadoop.mapred.Child.main(Child.java:249)

如果我正确读取此内容,则子进程因代码143而失败,因为有人向流式作业发送了SIGTERM信号.

我的理解是否正确？如果是这样:EMR基础设施何时发送SIGTERM？

我想使用s3distcp将一个文件复制到HDFS.我已经尝试使用srcPattern参数,但它没有帮助,它继续抛出java.lang.Runtime异常.我正在使用的正则表达式可能是罪魁祸首,请帮忙.

我的代码如下:

elastic-mapreduce -j $jobflow --jar s3://us-east-1.elasticmapreduce/libs/s3distcp/1.latest/s3distcp.jar --args '--src,s3://<mybucket>/<path>' --args '--dest,hdfs:///output' --arg --srcPattern --arg '(filename)'

抛出异常:

Exception in thread "main" java.lang.RuntimeException: Error running job at com.amazon.external.elasticmapreduce.s3distcp.S3DistCp.run(S3DistCp.java:586) at com.amazon.external.elasticmapreduce.s3distcp.S3DistCp.run(S3DistCp.java:216) at org.apache.hadoop.util.ToolRunner.run(ToolRunner.java:65) at org.apache.hadoop.util.ToolRunner.run(ToolRunner.java:79) at com.amazon.external.elasticmapreduce.s3distcp.Main.main(Main.java:12) at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39) at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25) at java.lang.reflect.Method.invoke(Method.java:597) at org.apache.hadoop.util.RunJar.main(RunJar.java:156) Caused by: org.apache.hadoop.mapred.InvalidInputException: Input path does not exist: hdfs:/tmp/a088f00d-a67e-4239-bb0d-32b3a6ef0105/files at org.apache.hadoop.mapred.FileInputFormat.listStatus(FileInputFormat.java:197) at org.apache.hadoop.mapred.SequenceFileInputFormat.listStatus(SequenceFileInputFormat.java:40) at org.apache.hadoop.mapred.FileInputFormat.getSplits(FileInputFormat.java:208) at org.apache.hadoop.mapred.JobClient.writeOldSplits(JobClient.java:1036) at org.apache.hadoop.mapred.JobClient.writeSplits(JobClient.java:1028) at org.apache.hadoop.mapred.JobClient.access$700(JobClient.java:172) at org.apache.hadoop.mapred.JobClient$2.run(JobClient.java:944) at org.apache.hadoop.mapred.JobClient$2.run(JobClient.java:897) at java.security.AccessController.doPrivileged(Native Method) at javax.security.auth.Subject.doAs(Subject.java:396) at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1059) at org.apache.hadoop.mapred.JobClient.submitJobInternal(JobClient.java:897) at …

我想帮助理解算法.我首先粘贴了算法解释,然后是我的疑惑.

算法:(用于计算记录对之间的重叠)

给定用户定义的参数K,文件DR(*Format:record_id,data*)被分成K个近似等大小的块,使得文档Di的数据落入第i/K th块中.

我们覆盖了Hadoop的分区功能,该功能将映射器发出的密钥映射到reducer实例.每个键(i,j)都映射到j/Kth组中的reducer.

特殊键i,*及其相关值,即文档的数据最多被复制K次,以便可以在每个缩减器处传送文档的全部内容.因此,组中的每个缩减器仅需要在存储器中恢复并加载一个DR文件块,其大小可以通过改变K而任意设置为小.因此可以计算重叠.这是以复制通过MapReduce框架提供的文档为代价的.

释疑:

我做了一些假设:

声明:每个键(i,j)都映射到j/Kth组中的reducer.假设:存在K个减少节点,并且该密钥被映射到j/Kth reduce节点.

怀疑:是否有一些减少节点组合在一起？比如0,1,2节点被分组为0组？

声明:文档的数据最多可复制K次,因此可以在每个reducer上传送文档的全部内容.

所以这意味着K等于没有.减速机节点？如果没有,我们正在浪费计算节点,而不是正确使用它们？

主要疑问:K是否等于减速器节点的数量？

希望得到回应!

谢谢!

有没有办法指示Hive将数据拆分成多个输出文件？或者可能限制输出文件的大小.

我打算使用Redshift,它建议将数据拆分成多个文件以允许并行加载http://docs.aws.amazon.com/redshift/latest/dg/t_splitting-data-files.html

我们预先处理hive中的所有数据,我想知道是否有一种方法可以创建,例如10个1GB文件可能会使复制速度更快.

我正在查看https://cwiki.apache.org/Hive/adminmanual-configuration.html和https://cwiki.apache.org/confluence/display/Hive/Configuration+Properties但我找不到任何东西

当我使用m1.large作为要由作业流创建的hadoop实例的实例类型运行我的Amazon EMR作业时,我收到错误"设备上没有剩余空间".这项工作产生约.最大10 GB的数据,因为m1.large实例的容量应该是420GB*2(根据:EC2实例类型)).我很困惑,只有10GB的数据才能导致"磁盘空间已满"的消息.我知道如果我们已经完全耗尽了文件系统上允许的inode总数,也可能会产生这种错误,但这可能是数百万的大数字,我很确定我的工作不是产生那么多文件.我已经看到,当我尝试创建一个独立于m1.large类型的EC2实例时,它默认为它分配一个8GB的根卷.这可能是在EMR中配置实例的原因吗？然后,何时将大小为420GB的磁盘分配给实例？

另外,这里是"df -hi"和"mount"的输出

$ df -hi
Filesystem            Inodes   IUsed   IFree IUse% Mounted on
/dev/xvda1              640K    100K    541K   16% /
tmpfs                   932K       3    932K    1% /lib/init/rw
udev                    930K     454    929K    1% /dev
tmpfs                   932K       3    932K    1% /dev/shm
ip-10-182-182-151.ec2.internal:/mapr
                        100G     50G     50G   50% /mapr

$ mount
/dev/xvda1 on / type ext3 (rw,noatime)
tmpfs on /lib/init/rw type tmpfs (rw,nosuid,mode=0755)
proc on /proc type proc (rw,noexec,nosuid,nodev)
sysfs on /sys type sysfs (rw,noexec,nosuid,nodev)
udev on /dev type tmpfs (rw,mode=0755)
tmpfs …

我想在我的 EMR Hadoop 作业中从 S3 读取文件。我正在使用自定义 JAR 选项。

我尝试过两种解决方案：

• org.apache.hadoop.fs.S3FileSystem: 抛出一个NullPointerException.
• com.amazonaws.services.s3.AmazonS3Client：抛出异常，显示“访问被拒绝”。

我无法理解的是，我是从控制台开始这项工作的，所以显然我应该拥有必要的权限。System.getenv()但是，映射器可用的环境变量 ( ) 中缺少 AWS_*_KEY 键。

我确信我做错了什么，只是不确定是什么。

如何在AWS Elastic MapReduce主节点上静音DEBUG消息？

hbase(main):003:0> list
TABLE                                                                                                               
mydb                                                                                                                
1 row(s) in 0.0510 seconds

hbase(main):004:0> 00:25:17.104 [main-SendThread(ip-172-31-14-206.ec2.internal:2181)] DEBUG org.apache.zookeeper.ClientCnxn - Got ping response for sessionid: 0x1493a5c3b78001b after 1ms

hbase(main):005:0* 00:26:17.165 [main-SendThread(ip-172-31-14-206.ec2.internal:2181)] DEBUG org.apache.zookeeper.ClientCnxn - Got ping response for sessionid: 0x1493a5c3b78001b after 1ms

我正在使用 Amazon Elastic Map Reduce 4.7.1、Hadoop 2.7.2、Hive 1.0.0 和 Spark 1.6.1。

用例：我有一个用于处理数据的 Spark 集群。该数据作为 Parquet 文件存储在 S3 中。我希望工具能够使用在 Hive Metastore 中注册的名称来查询数据（例如，查找foo表而不是parquet.`s3://bucket/key/prefix/foo/parquet`做事的风格）。我还希望这些数据在 Hive Metastore（一个单独的 RDS 实例）的生命周期内持续存在，即使我拆除 EMR 集群并启动一个连接到同一个 Metastore 的新集群也是如此。

问题：如果我这样做sqlContext.saveAsTable("foo")，默认情况下，会在 Hive Metastore 中创建一个托管表（请参阅https://spark.apache.org/docs/latest/sql-programming-guide.html）。这些托管表将数据从 S3 复制到 EMR 集群上的 HDFS，这意味着在拆除 EMR 集群后元数据将无用。

• 总实例:我创建了一个总共11个节点的EMR(1个主实例,10个核心实例).
• 工作提交: spark-submit myApplication.py

• 容器图:接下来,我有这些图,它们引用了"容器",而我并不完全是容器在EMR环境中的含义,所以这告诉我它并不明显:

• 实际运行的执行程序:然后我在我的spark历史UI中得到了这个,这表明我只创建了4个执行程序.
• 动态分配:然后我有了spark.dynamicAllocation.enabled=True,我可以在我的环境细节中看到.

• 执行程序内存:此外,默认执行程序内存为5120M.

• 执行者:接下来,我有我的执行器选项卡,显示我有3个活动和1个死执行程序:

所以,从表面上看,在我看来,我没有使用我的所有节点或可用内存.

1. 我怎么知道我是否使用了我可用的所有资源？
2. 如果我没有充分利用所有可用资源,我该如何改变我正在做的事情,以便充分发挥可用资源的潜力？