标签: hadoop-streaming

我有一个包含多达1000万条记录的数据框.我怎样才能快速计算？df.count需要很长时间.

Amazon EMR用于向集群添加步骤的文档表明,单个Elastic MapReduce步骤可以向Hadoop提交多个作业.但是,步骤配置的Amazon EMR文档表明,单个步骤只能容纳hadoop-streaming.jar的一次执行(即,HadoopJarStep是HadoopJarStepConfig而不是HadoopJarStepConfigs数组).

在一个步骤中向Hadoop提交多个作业的正确语法是什么？

我有用C#编写的mapper和reducer可执行文件.我想在Hadoop流媒体中使用它们.

这是我用来创建Hadoop作业的命令......

hadoop jar $HADOOP_HOME/contrib/streaming/hadoop-streaming-*.jar 
-input "/user/hduser/ss_waits" 
-output "/user/hduser/ss_waits-output" 
–mapper "mono mapper.exe" 
–reducer "mono reducer.exe" 
-file "mapper.exe" 
-file "reducer.exe"

这是每个映射器遇到的错误......

java.io.IOException: Type mismatch in key from map: expected org.apache.hadoop.io.Text, recieved org.apache.hadoop.io.LongWritable
at org.apache.hadoop.mapred.MapTask$MapOutputBuffer.collect(MapTask.java:1014)
at org.apache.hadoop.mapred.MapTask$OldOutputCollector.collect(MapTask.java:592)
at org.apache.hadoop.mapred.lib.IdentityMapper.map(IdentityMapper.java:38)
at org.apache.hadoop.mapred.MapRunner.run(MapRunner.java:50)
at org.apache.hadoop.mapred.MapTask.runOldMapper(MapTask.java:436)
at org.apache.hadoop.mapred.MapTask.run(MapTask.java:372)
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:415)
at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1121)
at org.apache.hadoop.mapred.Child.main(Child.java:249)

基于调用堆栈,问题似乎是(Java)IdentityMapper类被用作映射器.(这解释了导致类型不匹配错误的原因).映射器应该是可执行文件"mono mapper.exe".

任何想法为什么没有使用mono mapper.exe？

mapper.exe和reducer.exe具有以下权限:-rwxr-xr-x

我能够从unix命令shell 成功执行mono mapper.exe并让它从stdin中读取文本并写入stdout.

环境:

• Ubuntu Server 12.04 LTS(在Azure上运行的VM)
• Hadoop 1.0.4
• 单声道2.10

我已经为自己设置了Amazon Elastic MapReduce,以便执行各种标准的机器学习任务.我过去曾广泛使用Python进行本地机器学习,而且我不懂Java.

据我所知,没有用于分布式机器学习的完善的Python库.另一方面,Java有Apache Mahout和最近来自Cloudera的Oryx.

基本上我似乎必须在两个选项之间进行选择.通过并行化自己的算法来使用Hadoop流式传输或Hadoop的Python包装器之间的Slog,直到存在体面的库或跳转到Java以便我可以使用Mahout/Oryx.编写自己的MapReduce字数统计代码和编写自己的MapReduce SVM之间存在着天壤之别!即使与像伟大的教程帮助这个.

我不知道哪个是更明智的选择,所以我的问题是:

A)是否有一些我错过的Python库有用？如果没有,你知道在不久的将来是否有任何开发项目有用吗？

B)如果上面的答案是否定的,那么我的时间会更好地花在跳船上去Java吗？

我定义了一个MapReduce作业main.py,从中导入lib模块lib.py.我使用Hadoop Streaming将此作业提交到Hadoop集群,如下所示:

hadoop jar /usr/lib/hadoop-mapreduce/hadoop-streaming.jar -files lib.py,main.py 
    -mapper "./main.py map" -reducer "./main.py reduce" 
    -input input -output output

在我的理解,这应该把两者main.py并lib.py进入分布式缓存文件夹中的每个计算设备上,从而使模块lib可用main.但它没有发生:从日志中我看到文件真的被复制到同一目录,但main无法导入lib,抛出ImportError.

为什么会发生这种情况,我该如何解决？

UPD.将当前目录添加到路径不起作用:

import sys    
sys.path.append(os.path.realpath(__file__))
import lib
# ImportError

但是,手动加载模块可以解决问题:

import imp
lib = imp.load_source('lib', 'lib.py')

但这不是我想要的.那么为什么Python解释器会.py在同一目录中看到其他文件,但却无法导入它们？请注意,我已经尝试将空__init__.py文件添加到同一目录而不起作用.

我正在尝试使用R实现Hadoop Streaming的reducer.但是,我需要找到一种方法来访问某些不是用R,dplyr..etc构建的库.基于我的研究似乎有两种方法:

(1)在reducer代码中,将所需的库安装到临时文件夹中,它们将在会话完成后处理,如下所示:

.libPaths(c(.libPaths(), temp <- tempdir()))
install.packages("dplyr", lib=temp, repos='http://cran.us.r-project.org')
library(dplyr)
...

但是,这种方法会产生巨大的开销,具体取决于您尝试安装的库数量.因此,大多数时间都会浪费在安装库上(像dplyr这样的复杂库有很多依赖项,在vanilla R会话上安装需要几分钟).

所以听起来我需要事先安装它,这导致我们接近2.

(2)我的集群相当大.而且我必须使用像Ansible这样的工具来使它工作.所以我更喜欢使用一个Linux shell命令来安装库.我之前看过R CMD INSTALL...,感觉就像只从源文件安装软件包而不是install.packages()在R控制台中进行,找出镜像,拉出源文件,在一个命令中安装它.

谁能告诉我如何在shell中使用一个命令行来非交互式安装R包？(对不起这么多背景知识,如果有人认为我甚至没有遵循正确的哲学,请随意留下评论如何管理整个集群R包.)

对于python Hadoop流作业,如何将参数传递给例如reducer脚本,以便根据传入的参数使其行为不同？

我知道流媒体作业的格式为:

hadoop jar hadoop-streaming.jar -input -output -mapper mapper.py -reducer reducer.py ...

我想影响reducer.py.

嘿,我是大数据世界的新手.我在http://musicmachinery.com/2011/09/04/how-to-process-a-million-songs-in-20-minutes/上看到了这个教程.

它详细描述了如何在本地和Elastic Map Reduce上使用mrjob运行MapReduce作业.

好吧,我正试图在我自己的Hadoop cluser上运行它.我使用以下命令运行作业.

python density.py tiny.dat -r hadoop --hadoop-bin /usr/bin/hadoop > outputmusic

这就是我得到的:

HADOOP: Running job: job_1369345811890_0245
HADOOP: Job job_1369345811890_0245 running in uber mode : false
HADOOP:  map 0% reduce 0%
HADOOP: Task Id : attempt_1369345811890_0245_m_000000_0, Status : FAILED
HADOOP: Error: java.lang.RuntimeException: PipeMapRed.waitOutputThreads(): subprocess failed with code 1
HADOOP:         at org.apache.hadoop.streaming.PipeMapRed.waitOutputThreads(PipeMapRed.java:320)
HADOOP:         at org.apache.hadoop.streaming.PipeMapRed.mapRedFinished(PipeMapRed.java:533)
HADOOP:         at org.apache.hadoop.streaming.PipeMapper.close(PipeMapper.java:130)
HADOOP:         at org.apache.hadoop.mapred.MapRunner.run(MapRunner.java:61)
HADOOP:         at org.apache.hadoop.streaming.PipeMapRunner.run(PipeMapRunner.java:34)
HADOOP:         at org.apache.hadoop.mapred.MapTask.runOldMapper(MapTask.java:428)
HADOOP:         at org.apache.hadoop.mapred.MapTask.run(MapTask.java:340)
HADOOP:         at org.apache.hadoop.mapred.YarnChild$2.run(YarnChild.java:157)
HADOOP:         at java.security.AccessController.doPrivileged(Native …

我有一个hadoop流作业,其输出不包含键/值对.您可以将其视为仅限值对或仅键对.

我的流式缩减器(PHP脚本)输出由换行符分隔的记录.Hadoop流将其视为没有值的密钥,并在换行符之前插入一个选项卡.这个额外的标签是不需要的

我该如何删除它？

我正在使用带有AWS EMR的hadoop 1.0.3.我下载了hadoop 1.0.3的源代码,并在hadoop-1.0.3/src/contrib/streaming/src/java/org/apache/hadoop/streaming/PipeReducer.java中找到了这段代码:

reduceOutFieldSeparator = job_.get("stream.reduce.output.field.separator", "\t").getBytes("UTF-8");

所以我试着把工作-D stream.reduce.output.field.separator=作为一个参数传递给没有运气的工作.我也尝试过-D mapred.textoutputformat.separator=和-D mapreduce.output.textoutputformat.separator=没有运气.

我当然搜索谷歌,我找不到任何工作.一个搜索结果甚至表明,没有任何论据可以传递以达到预期的结果(但是,在这种情况下,hadoop版本真的很老).

这是我的代码(为了便于阅读,添加了换行符):

hadoop jar streaming.jar -files s3n://path/to/a/file.json#file.json
    -D mapred.output.compress=true -D stream.reduce.output.field.separator=
    -input s3n://path/to/some/input/*/* -output hdfs:///path/to/output/dir
    -mapper 'php my_mapper.php' -reducer 'php my_reducer.php'

问题:当我向hadoop 2.2.0群集提交作业时,它不会显示在作业跟踪器中,但作业成功完成.通过这个我可以看到输出,它正确运行并在运行时打印输出.

我尝试过多种选择,但是求职者没有看到这份工作.如果我使用2.2.0 hadoop运行流媒体作业,它会显示在任务跟踪器中,但是当我通过hadoop-client api提交它时,它不会显示在作业跟踪器中.我正在查看端口8088上的ui接口以验证该作业

环境 OSX Mavericks,Java 1.6,Hadoop 2.2.0单节点集群,Tomcat 7.0.47

码

    try {
        configuration.set("fs.defaultFS", "hdfs://127.0.0.1:9000");
        configuration.set("mapred.jobtracker.address", "localhost:9001");

        Job job = createJob(configuration);
        job.waitForCompletion(true);
    } catch (Exception e) {
        logger.log(Level.SEVERE, "Unable to execute job", e);
    }

    return null;

等/ hadoop的/ mapred-site.xml中

<configuration>
    <property>
        <name>mapreduce.framework.name</name>
        <value>yarn</value>
    </property>

    <property>
         <name>mapred.job.tracker</name>
         <value>localhost:9001</value>
    </property> 
</configuration>

等/ hadoop的/芯-site.xml中

<configuration>
     <property>
       <name>hadoop.tmp.dir</name>
       <value>/tmp/hadoop-${user.name}</value>
       <description>A base for other temporary directories.</description>
    </property>

    <property> 
      <name>fs.default.name</name> 
      <value>hdfs://localhost:9000</value> 
    </property>

</configuration>