标签: elastic-map-reduce

我注意到mrjob和boto都不支持在Amazon Elastic MapReduce(EMR)上提交和运行Hive作业的Python接口.是否有其他支持在EMR上运行Hive的Python客户端库？

在尝试在AWS EMR集群上使用Spark时,我最近遇到了一些问题.

我正在使用以下内容创建集群:

./elastic-mapreduce --create --alive \
--name "ll_Spark_Cluster" \
--bootstrap-action s3://elasticmapreduce/samples/spark/1.0.0/install-spark-shark.rb \
--bootstrap-name "Spark/Shark" \
--instance-type m1.xlarge \
--instance-count 2 \
--ami-version 3.0.4

问题是每当我尝试从S3获取数据时,我都会遇到异常.所以,如果我启动spark-shell并尝试类似:

val data = sc.textFile("s3n://your_s3_data")

我得到以下异常:

WARN storage.BlockManager: Putting block broadcast_1 failed
java.lang.NoSuchMethodError:
com.google.common.hash.HashFunction.hashInt(I)Lcom/google/common/hash/HashCode;

我想知道如何在使用自定义jar运行流作业时指定mapreduce配置,例如mapred.task.timeout,mapred.min.split.size等.

当我们使用外部脚本语言(如ruby或python)运行时,我们可以使用以下方式指定这些配置:

ruby elastic-mapreduce -j --stream --step-name"mystream"--jobconf mapred.task.timeout = 0 --jobconf mapred.min.split.size = 52880 --mapper s3:// somepath/mapper. rb --reducer s3:somepath/reducer.rb --input s3:// somepath/input --output s3:// somepath/output

我尝试了以下方法,但没有一个工作:

1. ruby elastic-mapreduce --jobflow --jar s3://somepath/job.jar --arg s3:// somepath/input --arg s3:// somepath/output --args -m,mapred.min.split .size = 52880 -m,mapred.task.timeout = 0

2. ruby elastic-mapreduce --jobflow --jar s3://somepath/job.jar --arg s3:// somepath/input --arg s3:// somepath/output --args -jobconf,mapred.min.split .size = 52880 -jobconf,mapred.task.timeout = 0

我还想知道如何使用EMR中的自定义jar将java选项传递给流作业.在hadoop上本地运行时,我们可以按如下方式传递它:

bin/hadoop jar job.jar input_path output_path -D <some_java_parameter> = <some_value>

我有一个巨大的DynamoDB表,我想分析它来聚合存储在其属性中的数据.然后,聚合数据应由Java应用程序处理.虽然我理解MapReduce背后的基本概念,但我以前从未使用过它.

在我的情况,让我们说,我有一个customerId和orderNumbers属性在每个DynamoDB项目,我可以有多个项目为同一客户.喜欢:

customerId: 1, orderNumbers: 2
customerId: 1, orderNumbers: 6
customerId: 2, orderNumbers: -1

基本上我想总结每个customerId的orderNumbers,然后用聚合在Java中执行一些操作.

AWS Elastic MapReduce可能对我有帮助,但我不明白如何将自定义JAR与DynamoDB连接.我的自定义JAR可能需要公开a map和reduce函数,在哪里可以找到正确的接口来实现？

另外我对文档感到有点困惑,似乎我应该先运行自定义JAR之前将数据导出到S3.它是否正确？

谢谢

我刚刚发现使用Amazon的Elastic Map Reduce,我可以指定一个步骤来选择三个ActionOnFailure中的一个:

• TERMINATE_JOB_FLOW
• CANCEL_AND_WAIT
• 继续

TERMINATE_JOB_FLOW是默认的并且显而易见 - 它在步骤失败时关闭整个群集.

CANCEL_AND_WAIT和CONTINUE有什么区别？在我看来,两者都将保持群集运行,并在添加时继续进行下一步.

我有一个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？

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

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

给定用户定义的参数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 …

如何在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