标签: mapreduce

我遇到了一个奇怪的问题.当我在大型数据集(> 1TB压缩文本文件)上运行Hadoop作业时,一些reduce任务失败,堆栈跟踪如下:

java.io.IOException: Task: attempt_201104061411_0002_r_000044_0 - The reduce copier failed
    at org.apache.hadoop.mapred.ReduceTask.run(ReduceTask.java:385)
    at org.apache.hadoop.mapred.Child$4.run(Child.java:240)
    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:1115)
    at org.apache.hadoop.mapred.Child.main(Child.java:234)
Caused by: java.io.IOException: Intermediate merge failed
    at org.apache.hadoop.mapred.ReduceTask$ReduceCopier$InMemFSMergeThread.doInMemMerge(ReduceTask.java:2714)
    at org.apache.hadoop.mapred.ReduceTask$ReduceCopier$InMemFSMergeThread.run(ReduceTask.java:2639)
Caused by: java.lang.RuntimeException: java.io.EOFException
    at org.apache.hadoop.io.WritableComparator.compare(WritableComparator.java:128)
    at org.apache.hadoop.mapred.Merger$MergeQueue.lessThan(Merger.java:373)
    at org.apache.hadoop.util.PriorityQueue.downHeap(PriorityQueue.java:139)
    at org.apache.hadoop.util.PriorityQueue.adjustTop(PriorityQueue.java:103)
    at org.apache.hadoop.mapred.Merger$MergeQueue.adjustPriorityQueue(Merger.java:335)
    at org.apache.hadoop.mapred.Merger$MergeQueue.next(Merger.java:350)
    at org.apache.hadoop.mapred.Merger.writeFile(Merger.java:156)
    at org.apache.hadoop.mapred.ReduceTask$ReduceCopier$InMemFSMergeThread.doInMemMerge(ReduceTask.java:2698)
    ... 1 more
Caused by: java.io.EOFException
    at java.io.DataInputStream.readInt(DataInputStream.java:375)
    at com.__.hadoop.pixel.segments.IpCookieCountFilter$IpAndIpCookieCount.readFields(IpCookieCountFilter.java:241)
    at org.apache.hadoop.io.WritableComparator.compare(WritableComparator.java:125)
    ... 8 more

java.io.IOException: Task: attempt_201104061411_0002_r_000056_0 - The reduce copier failed
    at org.apache.hadoop.mapred.ReduceTask.run(ReduceTask.java:385)
    at org.apache.hadoop.mapred.Child$4.run(Child.java:240) …

我试图在Windows下使用hadoop,当我想启动tasktracker时遇到问题.例如:

$bin/start-all.sh

然后日志写道:

2011-06-08 16:32:18,157 ERROR org.apache.hadoop.mapred.TaskTracker: Can not start task tracker because java.io.IOException: Failed to set permissions of path: /tmp/hadoop-Administrator/mapred/local/taskTracker to 0755
    at org.apache.hadoop.fs.RawLocalFileSystem.checkReturnValue(RawLocalFileSystem.java:525)
    at org.apache.hadoop.fs.RawLocalFileSystem.setPermission(RawLocalFileSystem.java:507)
    at org.apache.hadoop.fs.RawLocalFileSystem.mkdirs(RawLocalFileSystem.java:318)
    at org.apache.hadoop.fs.FilterFileSystem.mkdirs(FilterFileSystem.java:183)
    at org.apache.hadoop.mapred.TaskTracker.initialize(TaskTracker.java:630)
    at org.apache.hadoop.mapred.TaskTracker.<init>(TaskTracker.java:1328)
    at org.apache.hadoop.mapred.TaskTracker.main(TaskTracker.java:3430)

有什么问题？我怎么解决这个问题？谢谢!

是否可以在多台计算机上运行F#代理作为群集？

我刚刚开始使用mongo db并试图做一些简单的事情.我用包含"item"属性的数据集填充了我的数据库.我想尝试计算每个项目在集合中的时间

文件的例子:

{ "_id" : ObjectId("50dadc38bbd7591082d920f0"), "item" : "Pons", "lines" : 37 }

所以我设计了这两个函数来做MapReduce(使用pymongo在python中编写)

all_map = Code("function () {"
           "    emit(this.item, 1);"
           "}")

all_reduce = Code("function (key, values) {"
                  " var sum = 0;"
                  " values.forEach(function(value){"
                  "     sum += value;"
                  " });"
                  " return sum;"
                  "}")

这就像一个魅力,所以我开始填充这个系列.在周围30.000文件,在映射缩减已经持续一秒钟以上......因为NoSQL的被吹嘘的速度,我想我一定是在做什么错了!

Stack Overflow的一个问题让我查看了mongodb的聚合功能.所以我尝试使用group + sum + sort thingies.得出这个:

db.wikipedia.aggregate(
 { $group: { _id: "$item", count: { $sum: 1  }  } }, 
 { $sort: {count: 1}  }
)

这段代码工作正常,并给我与mapreduce集相同的结果,但它同样慢.难道我做错了什么？我真的需要使用像hadoop …

我试图使用Hadoop API - DFSCleint.getFileChecksum()复制到HDFS后检查文件的一致性.

我得到以上代码的以下输出:

Null
HDFS : null
Local : null

谁能指出错误或错误？这是代码:

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileChecksum;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocalFileSystem;
import org.apache.hadoop.fs.Path;


public class fileCheckSum {

    /**
     * @param args
     * @throws IOException 
     */
    public static void main(String[] args) throws IOException {
        // TODO Auto-generated method stub

        Configuration conf = new Configuration();

         FileSystem hadoopFS = FileSystem.get(conf);
    //  Path hdfsPath = new Path("/derby.log");

        LocalFileSystem localFS = LocalFileSystem.getLocal(conf);
    //      Path localPath = new Path("file:///home/ubuntu/derby.log");


    //  System.out.println("HDFS PATH : "+hdfsPath.getName()); …

我有两个1 MB的文本文件,每个文件存储在HDFS中作为MapReduce程序的输入.在下一行中,我们考虑map()的输入对.

class Map extends MapReduceBase implements Mapper<LongWritable, Text, Text, IntWritable> {

什么是LongWritable密钥？MapReduce如何假设？输入文件中的每行文本是否都有一个键或单个键是否为文件中的完整文本(即一次所有行)？

我是Hadoop生态系统的新手.

我最近在单节点集群上尝试了Hadoop(2.7.1)而没有任何问题,并决定转向具有1个namenode和2个datanode的多节点集群.

但是我面临一个奇怪的问题.无论我尝试运行什么工作,都会遇到以下消息:

在网络界面上:

YarnApplicationState: ACCEPTED: waiting for AM container to be allocated, launched and register

在cli中:

16/01/05 17:52:53 INFO mapreduce.Job: Running job: job_1451083949804_0001

他们甚至没有开始,在这一点上,我不确定我需要做出哪些改变才能使其发挥作用.

这是我试图解决的问题:

1. 在所有节点上禁用防火墙
2. 设置较低的资源限制
3. 在不同的机器,路由器和发行版下进行配置

我真的很感激任何帮助(即使是一分钟的提示)正确的方向.

我已按照这些说明(配置):

• 在Ubuntu Linux上运行Hadoop(多节点集群)

• 如何设置多节点Hadoop 2(YARN)群集

我正在努力深刻理解火花洗牌过程.当我开始阅读时,我遇到了以下几点.

Spark在完成时将Map任务(ShuffleMapTask)输出直接写入磁盘.

我想了解Hadoop MapReduce的以下内容.

1. 如果Map-Reduce和Spark都将数据写入本地磁盘,那么spark shuffle进程与Hadoop MapReduce有何不同？

2. 由于数据在Spark中表示为RDD,为什么这些输出不会保留在节点执行程序内存中？

3. Hadoop MapReduce和Spark的Map任务输出有何不同？

4. 如果有很多小的中间文件作为输出,火花如何处理网络和I/O瓶颈？

我正在分析2015年美国国内航班的准时性能记录.我需要按尾号分组,并将每个尾号的所有航班的日期分类列表存储在数据库中,以便我的应用程序检索.我不确定实现这一目标的两个选项中哪一个是最好的选择.

# Load the parquet file
on_time_dataframe = sqlContext.read.parquet('../data/on_time_performance.parquet')

# Filter down to the fields we need to identify and link to a flight
flights = on_time_dataframe.rdd.map(lambda x: 
  (x.Carrier, x.FlightDate, x.FlightNum, x.Origin, x.Dest, x.TailNum)
  )

我可以通过减少排序来实现这一目标......

# Group flights by tail number, sorted by date, then flight number, then 
origin/dest
flights_per_airplane = flights\
  .map(lambda nameTuple: (nameTuple[5], [nameTuple]))\
  .reduceByKey(lambda a, b: sorted(a + b, key=lambda x: (x[1],x[2],x[3],x[4])))

或者我可以在随后的地图工作中实现它......

# Do same in a map step, more efficient or does pySpark know …

我想问一下,在使用Hadoop/MapReduce和Spark时,数据分区有什么显着差异吗？它们都在HDFS(TextInputFormat)上工作,所以它在理论上应该是相同的.

是否有任何数据分区程序可能不同的情况？任何见解对我的学习都非常有帮助.

谢谢