‘壹’ hadoop的maprece常见算法案例有几种
基本MapRece模式
计数与求和
问题陈述:
有许多文档，每个文档都有一些字段组成。需要计算出每个字段在所有文档中的出现次数或者这些字段的其他什么统计值。例如，给定一个log文件，其中的每条记录都包含一个响应时间，需要计算出平均响应时间。
解决方案:
让我们先从简单的例子入手。在下面的代码片段里，Mapper每遇到指定词就把频次记1，Recer一个个遍历这些词的集合然后把他们的频次加和。
1 class Mapper
2 method Map(docid id, doc d)
3 for all term t in doc d do
4 Emit(term t, count 1)
5
6 class Recer
7 method Rece(term t, counts [c1, c2,...])
8 sum = 0
9 for all count c in [c1, c2,...] do
10 sum = sum + c
11 Emit(term t, count sum)
这种方法的缺点显而易见，Mapper提交了太多无意义的计数。它完全可以通过先对每个文档中的词进行计数从而减少传递给Recer的数据量:
1 class Mapper
2 method Map(docid id, doc d)
3 H = new AssociativeArray
4 for all term t in doc d do
5 H{t} = H{t} + 1
6 for all term t in H do
7 Emit(term t, count H{t})
如果要累计计数的的不只是单个文档中的内容，还包括了一个Mapper节点处理的所有文档，那就要用到Combiner了:
1 class Mapper
2 method Map(docid id, doc d)
3 for all term t in doc d do
4 Emit(term t, count 1)
5
6 class Combiner
7 method Combine(term t, [c1, c2,...])
8 sum = 0
9 for all count c in [c1, c2,...] do
10 sum = sum + c
11 Emit(term t, count sum)
12
13 class Recer
14 method Rece(term t, counts [c1, c2,...])
15 sum = 0
16 for all count c in [c1, c2,...] do
17 sum = sum + c
18 Emit(term t, count sum)
应用：Log 分析, 数据查询
整理归类
问题陈述:
有一系列条目，每个条目都有几个属性，要把具有同一属性值的条目都保存在一个文件里，或者把条目按照属性值分组。 最典型的应用是倒排索引。
解决方案：
解决方案很简单。 在 Mapper 中以每个条目的所需属性值作为 key，其本身作为值传递给 Recer。 Recer 取得按照属性值分组的条目，然后可以处理或者保存。如果是在构建倒排索引，那么 每个条目相当于一个词而属性值就是词所在的文档ID。
应用：倒排索引， ETL
过滤 (文本查找)，解析和校验
问题陈述:
假设有很多条记录，需要从其中找出满足某个条件的所有记录，或者将每条记录传换成另外一种形式（转换操作相对于各条记录独立，即对一条记录的操作与其他记录无关）。像文本解析、特定值抽取、格式转换等都属于后一种用例。
解决方案:
非常简单，在Mapper 里逐条进行操作，输出需要的值或转换后的形式。
应用：日志分析，数据查询，ETL，数据校验
分布式任务执行
问题陈述:
大型计算可以分解为多个部分分别进行然后合并各个计算的结果以获得最终结果。
解决方案: 将数据切分成多份作为每个 Mapper 的输入，每个Mapper处理一份数据，执行同样的运算，产生结果，Recer把多个Mapper的结果组合成一个。
案例研究： 数字通信系统模拟
像 WiMAX 这样的数字通信模拟软件通过系统模型来传输大量的随机数据，然后计算传输中的错误几率。 每个 Mapper 处理样本 1/N 的数据，计算出这部分数据的错误率，然后在 Recer 里计算平均错误率。
应用：工程模拟，数字分析，性能测试
排序
问题陈述:
有许多条记录，需要按照某种规则将所有记录排序或是按照顺序来处理记录。
解决方案: 简单排序很好办 – Mappers 将待排序的属性值为键，整条记录为值输出。 不过实际应用中的排序要更加巧妙一点， 这就是它之所以被称为MapRece 核心的原因（“核心”是说排序？因为证明Hadoop计算能力的实验是大数据排序？还是说Hadoop的处理过程中对key排序的环节？）。在实践中，常用组合键来实现二次排序和分组。
MapRece 最初只能够对键排序， 但是也有技术利用可以利用Hadoop 的特性来实现按值排序。想了解的话可以看这篇博客。
按照BigTable的概念，使用 MapRece来对最初数据而非中间数据排序，也即保持数据的有序状态更有好处，必须注意这一点。换句话说，在数据插入时排序一次要比在每次查询数据的时候排序更高效。
应用：ETL，数据分析
非基本 MapRece 模式
迭代消息传递 (图处理)
问题陈述：
假设一个实体网络，实体之间存在着关系。 需要按照与它比邻的其他实体的属性计算出一个状态。这个状态可以表现为它和其它节点之间的距离， 存在特定属性的邻接点的迹象， 邻域密度特征等等。
解决方案：
网络存储为系列节点的结合，每个节点包含有其所有邻接点ID的列表。按照这个概念，MapRece 迭代进行，每次迭代中每个节点都发消息给它的邻接点。邻接点根据接收到的信息更新自己的状态。当满足了某些条件的时候迭代停止，如达到了最大迭代次数（网络半径）或两次连续的迭代几乎没有状态改变。从技术上来看，Mapper 以每个邻接点的ID为键发出信息，所有的信息都会按照接受节点分组，recer 就能够重算各节点的状态然后更新那些状态改变了的节点。下面展示了这个算法：
1 class Mapper
2 method Map(id n, object N)
3 Emit(id n, object N)
4 for all id m in N.OutgoingRelations do
5 Emit(id m, message getMessage(N))
6
7 class Recer
8 method Rece(id m, [s1, s2,...])
9 M = null
10 messages = []
11 for all s in [s1, s2,...] do
12 if IsObject(s) then
13 M = s
14 else // s is a message
15 messages.add(s)
16 M.State = calculateState(messages)
17 Emit(id m, item M)
一个节点的状态可以迅速的沿着网络传全网，那些被感染了的节点又去感染它们的邻居，整个过程就像下面的图示一样：
案例研究： 沿分类树的有效性传递
问题陈述：
这个问题来自于真实的电子商务应用。将各种货物分类，这些类别可以组成一个树形结构，比较大的分类（像男人、女人、儿童）可以再分出小分类（像男裤或女装），直到不能再分为止（像男式蓝色牛仔裤）。这些不能再分的基层类别可以是有效（这个类别包含有货品）或者已无效的（没有属于这个分类的货品）。如果一个分类至少含有一个有效的子分类那么认为这个分类也是有效的。我们需要在已知一些基层分类有效的情况下找出分类树上所有有效的分类。
解决方案：
这个问题可以用上一节提到的框架来解决。我们咋下面定义了名为 getMessage和 calculateState 的方法：
1 class N
2 State in {True = 2, False = 1, null = 0},
3 initialized 1 or 2 for end-of-line categories, 0 otherwise
4 method getMessage(object N)
5 return N.State
6 method calculateState(state s, data [d1, d2,...])
7 return max( [d1, d2,...] )
案例研究：广度优先搜索
问题陈述：需要计算出一个图结构中某一个节点到其它所有节点的距离。
解决方案： Source源节点给所有邻接点发出值为0的信号，邻接点把收到的信号再转发给自己的邻接点，每转发一次就对信号值加1：
1 class N
2 State is distance,
3 initialized 0 for source node, INFINITY for all other nodes
4 method getMessage(N)
5 return N.State + 1
6 method calculateState(state s, data [d1, d2,...])
7 min( [d1, d2,...] )
案例研究：网页排名和 Mapper 端数据聚合
这个算法由Google提出，使用权威的PageRank算法，通过连接到一个网页的其他网页来计算网页的相关性。真实算法是相当复杂的，但是核心思想是权重可以传播，也即通过一个节点的各联接节点的权重的均值来计算节点自身的权重。
1 class N
2 State is PageRank
3 method getMessage(object N)
4 return N.State / N.OutgoingRelations.size()
5 method calculateState(state s, data [d1, d2,...])
6 return ( sum([d1, d2,...]) )
要指出的是上面用一个数值来作为评分实际上是一种简化，在实际情况下，我们需要在Mapper端来进行聚合计算得出这个值。下面的代码片段展示了这个改变后的逻辑 （针对于 PageRank 算法）：
1 class Mapper
2 method Initialize
3 H = new AssociativeArray
4 method Map(id n, object N)
5 p = N.PageRank / N.OutgoingRelations.size()
6 Emit(id n, object N)
7 for all id m in N.OutgoingRelations do
8 H{m} = H{m} + p
9 method Close
10 for all id n in H do
11 Emit(id n, value H{n})
12
13 class Recer
14 method Rece(id m, [s1, s2,...])
15 M = null
16 p = 0
17 for all s in [s1, s2,...] do
18 if IsObject(s) then
19 M = s
20 else
21 p = p + s
22 M.PageRank = p
23 Emit(id m, item M)
应用：图分析，网页索引
值去重 （对唯一项计数）
问题陈述: 记录包含值域F和值域 G，要分别统计相同G值的记录中不同的F值的数目 (相当于按照 G分组).
这个问题可以推而广之应用于分面搜索（某些电子商务网站称之为Narrow Search）
Record 1: F=1, G={a, b}
Record 2: F=2, G={a, d, e}
Record 3: F=1, G={b}
Record 4: F=3, G={a, b}
Result:
a -> 3 // F=1, F=2, F=3
b -> 2 // F=1, F=3
d -> 1 // F=2
e -> 1 // F=2
解决方案 I:
第一种方法是分两个阶段来解决这个问题。第一阶段在Mapper中使用F和G组成一个复合值对，然后在Recer中输出每个值对，目的是为了保证F值的唯一性。在第二阶段，再将值对按照G值来分组计算每组中的条目数。
第一阶段：
1 class Mapper
2 method Map(null, record [value f, categories [g1, g2,...]])
3 for all category g in [g1, g2,...]
4 Emit(record [g, f], count 1)
5
6 class Recer
7 method Rece(record [g, f], counts [n1, n2, ...])
8 Emit(record [g, f], null )
第二阶段：
1 class Mapper
2 method Map(record [f, g], null)
3 Emit(value g, count 1)
4
5 class Recer
6 method Rece(value g, counts [n1, n2,...])
7 Emit(value g, sum( [n1, n2,...] ) )
解决方案 II:
第二种方法只需要一次MapRece 即可实现，但扩展性不强。算法很简单-Mapper 输出值和分类，在Recer里为每个值对应的分类去重然后给每个所属的分类计数加1，最后再在Recer结束后将所有计数加和。这种方法适用于只有有限个分类，而且拥有相同F值的记录不是很多的情况。例如网络日志处理和用户分类，用户的总数很多，但是每个用户的事件是有限的，以此分类得到的类别也是有限的。值得一提的是在这种模式下可以在数据传输到Recer之前使用Combiner来去除分类的重复值。
1 class Mapper
2 method Map(null, record [value f, categories [g1, g2,...] )
3 for all category g in [g1, g2,...]
4 Emit(value f, category g)
5
6 class Recer
7 method Initialize
8 H = new AssociativeArray : category -> count
9 method Rece(value f, categories [g1, g2,...])
10 [g1', g2',..] = ExcludeDuplicates( [g1, g2,..] )
11 for all category g in [g1', g2',...]
12 H{g} = H{g} + 1
13 method Close
14 for all category g in H do
15 Emit(category g, count H{g})
应用：日志分析，用户计数
互相关
问题陈述：有多个各由若干项构成的组，计算项两两共同出现于一个组中的次数。假如项数是N，那么应该计算N*N。
这种情况常见于文本分析（条目是单词而元组是句子），市场分析（购买了此物的客户还可能购买什么）。如果N*N小到可以容纳于一台机器的内存，实现起来就比较简单了。
配对法
第一种方法是在Mapper中给所有条目配对，然后在Recer中将同一条目对的计数加和。但这种做法也有缺点：
使用 combiners 带来的的好处有限，因为很可能所有项对都是唯一的
不能有效利用内存
1 class Mapper
2 method Map(null, items [i1, i2,...] )
3 for all item i in [i1, i2,...]
4 for all item j in [i1, i2,...]
5 Emit(pair [i j], count 1)
6
7 class Recer
8 method Rece(pair [i j], counts [c1, c2,...])
9 s = sum([c1, c2,...])
10 Emit(pair[i j], count s)
Stripes Approach（条方法？不知道这个名字怎么理解）
第二种方法是将数据按照pair中的第一项来分组，并维护一个关联数组，数组中存储的是所有关联项的计数。The second approach is to group data by the first item in pair and maintain an associative array (“stripe”) where counters for all adjacent items are accumulated. Recer receives all stripes for leading item i, merges them, and emits the same result as in the Pairs approach.
中间结果的键数量相对较少，因此减少了排序消耗。
可以有效利用 combiners。
可在内存中执行，不过如果没有正确执行的话也会带来问题。
实现起来比较复杂。
一般来说， “stripes” 比 “pairs” 更快
1 class Mapper
2 method Map(null, items [i1, i2,...] )
3 for all item i in [i1, i2,...]
4 H = new AssociativeArray : item -> counter
5 for all item j in [i1, i2,...]
6 H{j} = H{j} + 1
7 Emit(item i, stripe H)
8
9 class Recer
10 method Rece(item i, stripes [H1, H2,...])
11 H = new AssociativeArray : item -> counter
12 H = merge-sum( [H1, H2,...] )
13 for all item j in H.keys()
14 Emit(pair [i j], H{j})
应用：文本分析，市场分析
参考资料：Lin J. Dyer C. Hirst G. Data Intensive Processing MapRece
用MapRece 表达关系模式
在这部分我们会讨论一下怎么使用MapRece来进行主要的关系操作。
筛选（Selection）
1 class Mapper
2 method Map(rowkey key, tuple t)
3 if t satisfies the predicate
4 Emit(tuple t, null)
投影（Projection）
投影只比筛选稍微复杂一点，在这种情况下我们可以用Recer来消除可能的重复值。
1 class Mapper
2 method Map(rowkey key, tuple t)
3 tuple g = project(t) // extract required fields to tuple g
4 Emit(tuple g, null)
5
6 class Recer
‘贰’ 怎么优化hadoop任务调度算法
首先介绍了Hadoop平台下作业的分布式运行机制，然后对Hadoop平台自带的4种任务调度器做分析和比较，最后在分析JobTracker类文件的基础上指出了创建自定义任务调度器所需完成的工作。
首先Hadoop集群式基于单服务器的，只有一个服务器节点负责调度整个集群的作业运行，主要的具体工作是切分大数据量的作业，指定哪些Worker节点做Map工作、哪些Worker节点做Rece工作、与Worker节点通信并接受其心跳信号、作为用户的访问入口等等。其次，集群中的每个Worker节点相当于一个器官，运行着主节点所指派的具体作业。这些节点会被分为两种类型，一种是接收分块之后的作业并做映射工作。另一种是负责把前面所做的映射工作按照约定的规则做一个统计。
Task－Tracker通过运行一个简单循环来定期地发送心跳信号（heartbeat）给JobTracker．这个心跳信号会把TaskTracker是否还在存活告知JobTracker，TaskTracker通过信号指明自己是否已经准备
好运行新的任务．一旦TaskTracker已经准备好接受任务，JobTracker就会从作业优先级表中选定一个作业并分配下去．至于到底是执行Map任务还是Rece任务，是由TaskTracker的任务槽所决定的．默认的任务调度器在处理Rece任务之前，会优先填满空闲的Map任务槽．因此，如果TaskTracker满足存在至少一个空闲任务槽时，JobTracker会为它分配Map任务，否则为它选择一个Rece任务．TaskTracker在运行任务的时候，第一步是从共享文件系统中把作业的JAR文件复制过来，从而实现任务文件的本地化．第二步是TaskTracker为任务新建一个本地文件夹并把作业文件解压在此目录中．第三步是由Task－Tracker新建一个TaskRunner实例来运行该任务．
Hadoop平台默认的调度方案就是JobQueueTaskScheler，这是一种按照任务到来的时间先后顺序而执行的调度策略．这种方式比较简单，JobTracker作为主控节点，仅仅是依照作业到来的先后顺序而选择将要执行的作业．当然，这有一定的缺陷，由于Hadoop平台是默认将作业运行在整个集群上的，那么如果一个耗时非常大的作业进入执行期，将会导致其余大量作业长时间得不到运行．这种长时间运行的优先级别并不高的作业带来了严重的作业阻塞，使得整个平台的运行效率处在较低的水平．Hadoop平台对这种FIFO（FirstINAndFirstOut）机制所给出的解决办法是调用SetJobPriority（）方法，通过设置作业的权重级别来做平衡调度．
FairScheler是一种“公平”调度器，它的目标是让每个用户能够公平地共享Hadoop集群计算能力．当只有一个作业运行的时候，它会得到整个集群的资源．随着提交到作业表中作业的增多，Hadoop平台会把集群中空闲出来的时间槽公平分配给每个需要执行的作业．这样即便其中某些作业需要较长时间运行，平台仍然有能力让那些短作业在合理时间内完成［3］．FairScheler支持资源抢占，当一个资源池在一定时段内没有得到公平共享时，它会终止该资源池所获得的过多的资源，同时把这些释放的资源让给那些资源不足的资源池．
Hadoop平台中的CapacityScheler是由Yahoo贡献的，在调度器上，设置了三种粒度的对象：queue，job，task．在该策略下，平台可以有多个作业队列，每个作业队列经提交后，都会获得一定数量的TaskTracker资源．具体调度流程如下．
（1）选择queue，根据资源库的使用情况从小到大排序，直到找到一个合适的job．
（2）选择job，在当前所选定的queue中，按照作业提交的时间先后以及作业的权重优先级别进行排序，选择合适的job．当然，在job选择时还需要考虑所选作业是否超出目前现有的资源上限，以及资源池中的内存是否够该job的task用等因素．
（3）选择task，根据本地节点的资源使用情况来选择合适的task．
虽然Hadoop平台自带了几种调度器，但是上述3种调度方案很难满足公司复杂的应用需求．因此作为平台的个性化使用者，往往需要开发自己的调度器．Hadoop的调度器是在JobTracker中加载和调用的，因此开发一个自定义的调度器就必须搞清楚JobTracker类文件的内部机制．作为Hadoop平台的核心组件，JobTracker监控着整个集群的作业运行情况并对资源进行管理调度．每个Task－Tracker每隔3s通过heartbeat向JobTracker汇报自己管理的机器的一些基本信息，包括内存使用量、内存的剩余量以及空闲的slot数目等等［5］．一
旦JobTracker发现了空闲slot，便会调用调度器中的AssignTask方法为该TaskTracker分配task。
‘叁’ 如何使用python为Hadoop编写一个简单的MapRece程序
在这个实例中，我将会向大家介绍如何使用Python 为 Hadoop编写一个简单的MapRece
程序。
尽管Hadoop 框架是使用java编写的但是我们仍然需要使用像C++、Python等语言来实现Hadoop程序。尽管Hadoop官方网站给的示例程序是使用Jython编写并打包成Jar文件，这样显然造成了不便，其实，不一定非要这样来实现，我们可以使用Python与Hadoop 关联进行编程，看看位于/src/examples/python/WordCount.py 的例子，你将了解到我在说什么。
我们想要做什么？
我们将编写一个简单的 MapRece 程序，使用的是C-Python，而不是Jython编写后打包成jar包的程序。
我们的这个例子将模仿 WordCount 并使用Python来实现，例子通过读取文本文件来统计出单词的出现次数。结果也以文本形式输出，每一行包含一个单词和单词出现的次数，两者中间使用制表符来想间隔。
先决条件
编写这个程序之前，你学要架设好Hadoop 集群，这样才能不会在后期工作抓瞎。如果你没有架设好，那么在后面有个简明教程来教你在Ubuntu linux 上搭建（同样适用于其他发行版linux、unix）
如何使用Hadoop Distributed File System (HDFS)在Ubuntu Linux 建立单节点的 Hadoop 集群
如何使用Hadoop Distributed File System (HDFS)在Ubuntu Linux 建立多节点的 Hadoop 集群
Python的MapRece代码
使用Python编写MapRece代码的技巧就在于我们使用了 HadoopStreaming 来帮助我们在Map 和 Rece间传递数据通过STDIN (标准输入)和STDOUT (标准输出).我们仅仅使用Python的sys.stdin来输入数据，使用sys.stdout输出数据，这样做是因为HadoopStreaming会帮我们办好其他事。这是真的，别不相信！
Map: mapper.py
将下列的代码保存在/home/hadoop/mapper.py中，他将从STDIN读取数据并将单词成行分隔开，生成一个列表映射单词与发生次数的关系：
注意：要确保这个脚本有足够权限（chmod +x /home/hadoop/mapper.py）。
#!/usr/bin/env python
import sys
# input comes from STDIN (standard input)
for line in sys.stdin:
# remove leading and trailing whitespace
line = line.strip()
# split the line into words
words = line.split()
# increase counters
for word in words:
# write the results to STDOUT (standard output);
# what we output here will be the input for the
# Rece step, i.e. the input for recer.py
#
# tab-delimited; the trivial word count is 1
print '%s\\t%s' % (word, 1)在这个脚本中，并不计算出单词出现的总数，它将输出 "<word> 1" 迅速地，尽管<word>可能会在输入中出现多次，计算是留给后来的Rece步骤（或叫做程序）来实现。当然你可以改变下编码风格，完全尊重你的习惯。
Rece: recer.py
将代码存储在/home/hadoop/recer.py 中，这个脚本的作用是从mapper.py 的STDIN中读取结果，然后计算每个单词出现次数的总和，并输出结果到STDOUT。
同样，要注意脚本权限：chmod +x /home/hadoop/recer.py
#!/usr/bin/env python
from operator import itemgetter
import sys
# maps words to their counts
word2count = {}
# input comes from STDIN
for line in sys.stdin:
# remove leading and trailing whitespace
line = line.strip()
# parse the input we got from mapper.py
word, count = line.split('\\t', 1)
# convert count (currently a string) to int
try:
count = int(count)
word2count[word] = word2count.get(word, 0) + count
except ValueError:
# count was not a number, so silently
# ignore/discard this line
pass
# sort the words lexigraphically;
#
# this step is NOT required, we just do it so that our
# final output will look more like the official Hadoop
# word count examples
sorted_word2count = sorted(word2count.items(), key=itemgetter(0))
# write the results to STDOUT (standard output)
for word, count in sorted_word2count:
print '%s\\t%s'% (word, count)
测试你的代码（cat data | map | sort | rece）
我建议你在运行MapRece job测试前尝试手工测试你的mapper.py 和 recer.py脚本，以免得不到任何返回结果
这里有一些建议，关于如何测试你的Map和Rece的功能：
——————————————————————————————————————————————
\r\n
# very basic test
hadoop@ubuntu:~$ echo "foo foo quux labs foo bar quux" | /home/hadoop/mapper.py
foo 1
foo 1
quux 1
labs 1
foo 1
bar 1
——————————————————————————————————————————————
hadoop@ubuntu:~$ echo "foo foo quux labs foo bar quux" | /home/hadoop/mapper.py | sort | /home/hadoop/recer.py
bar 1
foo 3
labs 1
——————————————————————————————————————————————
# using one of the ebooks as example input
# (see below on where to get the ebooks)
hadoop@ubuntu:~$ cat /tmp/gutenberg/20417-8.txt | /home/hadoop/mapper.py
The 1
Project 1
Gutenberg 1
EBook 1
of 1
[...]
(you get the idea)
quux 2
quux 1
——————————————————————————————————————————————
在Hadoop平台上运行Python脚本
为了这个例子，我们将需要三种电子书：
The Outline of Science, Vol. 1 (of 4) by J. Arthur Thomson\r\n
The Notebooks of Leonardo Da Vinci\r\n
Ulysses by James Joyce
下载他们，并使用us-ascii编码存储 解压后的文件，保存在临时目录，比如/tmp/gutenberg.
hadoop@ubuntu:~$ ls -l /tmp/gutenberg/
total 3592
-rw-r--r-- 1 hadoop hadoop 674425 2007-01-22 12:56 20417-8.txt
-rw-r--r-- 1 hadoop hadoop 1423808 2006-08-03 16:36 7ldvc10.txt
-rw-r--r-- 1 hadoop hadoop 1561677 2004-11-26 09:48 ulyss12.txt
hadoop@ubuntu:~$
复制本地数据到HDFS
在我们运行MapRece job 前，我们需要将本地的文件复制到HDFS中：
hadoop@ubuntu:/usr/local/hadoop$ bin/hadoop dfs -FromLocal /tmp/gutenberg gutenberg
hadoop@ubuntu:/usr/local/hadoop$ bin/hadoop dfs -ls
Found 1 items
/user/hadoop/gutenberg <dir>
hadoop@ubuntu:/usr/local/hadoop$ bin/hadoop dfs -ls gutenberg
Found 3 items
/user/hadoop/gutenberg/20417-8.txt <r 1> 674425
/user/hadoop/gutenberg/7ldvc10.txt <r 1> 1423808
/user/hadoop/gutenberg/ulyss12.txt <r 1> 1561677
执行 MapRece job
现在，一切准备就绪，我们将在运行Python MapRece job 在Hadoop集群上。像我上面所说的，我们使用的是
HadoopStreaming 帮助我们传递数据在Map和Rece间并通过STDIN和STDOUT，进行标准化输入输出。
hadoop@ubuntu:/usr/local/hadoop$ bin/hadoop jar contrib/streaming/hadoop-0.19.1-streaming.jar
-mapper /home/hadoop/mapper.py -recer /home/hadoop/recer.py -input gutenberg/*
-output gutenberg-output
在运行中，如果你想更改Hadoop的一些设置，如增加Rece任务的数量，你可以使用“-jobconf”选项：
hadoop@ubuntu:/usr/local/hadoop$ bin/hadoop jar contrib/streaming/hadoop-0.19.1-streaming.jar
-jobconf mapred.rece.tasks=16 -mapper ...
一个重要的备忘是关于Hadoop does not honor mapred.map.tasks
这个任务将会读取HDFS目录下的gutenberg并处理他们，将结果存储在独立的结果文件中，并存储在HDFS目录下的
gutenberg-output目录。
之前执行的结果如下：
hadoop@ubuntu:/usr/local/hadoop$ bin/hadoop jar contrib/streaming/hadoop-0.19.1-streaming.jar
-mapper /home/hadoop/mapper.py -recer /home/hadoop/recer.py -input gutenberg/*
-output gutenberg-output
additionalConfSpec_:null
null=@@@userJobConfProps_.get(stream.shipped.hadoopstreaming
packageJobJar: [/usr/local/hadoop-datastore/hadoop-hadoop/hadoop-unjar54543/]
[] /tmp/streamjob54544.jar tmpDir=null
[...] INFO mapred.FileInputFormat: Total input paths to process : 7
[...] INFO streaming.StreamJob: getLocalDirs(): [/usr/local/hadoop-datastore/hadoop-hadoop/mapred/local]
[...] INFO streaming.StreamJob: Running job: job_200803031615_0021
[...]
[...] INFO streaming.StreamJob: map 0% rece 0%
[...] INFO streaming.StreamJob: map 43% rece 0%
[...] INFO streaming.StreamJob: map 86% rece 0%
[...] INFO streaming.StreamJob: map 100% rece 0%
[...] INFO streaming.StreamJob: map 100% rece 33%
[...] INFO streaming.StreamJob: map 100% rece 70%
[...] INFO streaming.StreamJob: map 100% rece 77%
[...] INFO streaming.StreamJob: map 100% rece 100%
[...] INFO streaming.StreamJob: Job complete: job_200803031615_0021
[...] INFO streaming.StreamJob: Output: gutenberg-output hadoop@ubuntu:/usr/local/hadoop$
正如你所见到的上面的输出结果，Hadoop 同时还提供了一个基本的WEB接口显示统计结果和信息。
当Hadoop集群在执行时，你可以使用浏览器访问 http://localhost:50030/ ，如图：
检查结果是否输出并存储在HDFS目录下的gutenberg-output中：
hadoop@ubuntu:/usr/local/hadoop$ bin/hadoop dfs -ls gutenberg-output
Found 1 items
/user/hadoop/gutenberg-output/part-00000 <r 1> 903193 2007-09-21 13:00
hadoop@ubuntu:/usr/local/hadoop$
可以使用dfs -cat 命令检查文件目录
hadoop@ubuntu:/usr/local/hadoop$ bin/hadoop dfs -cat gutenberg-output/part-00000
"(Lo)cra" 1
"1490 1
"1498," 1
"35" 1
"40," 1
"A 2
"AS-IS". 2
"A_ 1
"Absoluti 1
[...]
hadoop@ubuntu:/usr/local/hadoop$
注意比输出，上面结果的(")符号不是Hadoop插入的。
转载仅供参考，版权属于原作者。祝你愉快，满意请采纳哦
‘肆’ 如何快速地编写和运行一个属于自己的MapRece例子程序
大数据的时代， 到处张嘴闭嘴都是Hadoop, MapRece, 不跟上时代怎么行？ 可是对一个hadoop的新手， 写一个属于自己的MapRece程序还是小有点难度的， 需要建立一个maven项目， 还要搞清楚各种库的依赖， 再加上编译运行， 基本上头大两圈了吧。 这也使得很多只是想简单了解一下MapRece的人望而却步。
本文会教你如何用最快最简单的方法编写和运行一个属于自己的MapRece程序， let's go!
首先有两个前提：
1. 有一个已经可以运行的hadoop 集群（也可以是伪分布系统)， 上面的hdfs和maprece工作正常 （这个真的是最基本的了， 不再累述， 不会的请参考 http://hadoop.apache.org/docs/current/)
2. 集群上安装了JDK (编译运行时会用到）
正式开始
1. 首先登入hadoop 集群里面的一个节点， 创建一个java源文件， 偷懒起见， 基本盗用官方的word count (因为本文的目的是教会你如何快编写和运行一个MapRece程序， 而不是如何写好一个功能齐全的MapRece程序）
内容如下：
import java.io.IOException;
import java.util.StringTokenizer;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.maprece.Job;
import org.apache.hadoop.maprece.Mapper;
import org.apache.hadoop.maprece.Recer;
import org.apache.hadoop.maprece.lib.input.FileInputFormat;
import org.apache.hadoop.maprece.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
public class myword {
public static class TokenizerMapper
extends Mapper<Object, Text, Text, IntWritable>{
private final static IntWritable one = new IntWritable(1);
private Text word = new Text();
public void map(Object key, Text value, Context context
) throws IOException, InterruptedException {
StringTokenizer itr = new StringTokenizer(value.toString());
while (itr.hasMoreTokens()) {
word.set(itr.nextToken());
context.write(word, one);
}
}
}
public static class IntSumRecer
extends Recer<Text,IntWritable,Text,IntWritable> {
private IntWritable result = new IntWritable();
public void rece(Text key, Iterable<IntWritable> values,
Context context
) throws IOException, InterruptedException {
int sum = 0;
for (IntWritable val : values) {
sum += val.get();
}
result.set(sum);
context.write(key, result);
}
}
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
if (otherArgs.length != 2) {
System.err.println('Usage: wordcount <in> <out>');
System.exit(2);
}
Job job = new Job(conf, 'word count');
job.setJarByClass(myword.class);
job.setMapperClass(TokenizerMapper.class);
job.setCombinerClass(IntSumRecer.class);
job.setRecerClass(IntSumRecer.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}
与官方版本相比， 主要做了两处修改
1） 为了简单起见，去掉了开头的 package org.apache.hadoop.examples;
2） 将类名从 WordCount 改为 myword, 以体现是我们自己的工作成果 :)
2. 拿到hadoop 运行的class path, 主要为编译所用
运行命令
hadoop classpath
保存打出的结果，本文用的hadoop 版本是Pivotal 公司的Pivotal hadoop, 例子：
/etc/gphd/hadoop/conf:/usr/lib/gphd/hadoop/lib/*:/usr/lib/gphd/hadoop/.//*:/usr/lib/gphd/hadoop-hdfs/./:/usr/lib/gphd/hadoop-hdfs/lib/*:/usr/lib/gphd/hadoop-hdfs/.//*:/usr/lib/gphd/hadoop-yarn/lib/*:/usr/lib/gphd/hadoop-yarn/.//*:/usr/lib/gphd/hadoop-maprece/lib/*:/usr/lib/gphd/hadoop-maprece/.//*::/etc/gphd/pxf/conf::/usr/lib/gphd/pxf/pxf-core.jar:/usr/lib/gphd/pxf/pxf-api.jar:/usr/lib/gphd/publicstage:/usr/lib/gphd/gfxd/lib/gemfirexd.jar::/usr/lib/gphd/zookeeper/zookeeper.jar:/usr/lib/gphd/hbase/lib/hbase-common.jar:/usr/lib/gphd/hbase/lib/hbase-protocol.jar:/usr/lib/gphd/hbase/lib/hbase-client.jar:/usr/lib/gphd/hbase/lib/hbase-thrift.jar:/usr/lib/gphd/hbase/lib/htrace-core-2.01.jar:/etc/gphd/hbase/conf::/usr/lib/gphd/hive/lib/hive-service.jar:/usr/lib/gphd/hive/lib/libthrift-0.9.0.jar:/usr/lib/gphd/hive/lib/hive-metastore.jar:/usr/lib/gphd/hive/lib/libfb303-0.9.0.jar:/usr/lib/gphd/hive/lib/hive-common.jar:/usr/lib/gphd/hive/lib/hive-exec.jar:/usr/lib/gphd/hive/lib/postgresql-jdbc.jar:/etc/gphd/hive/conf::/usr/lib/gphd/sm-plugins/*:
3. 编译
运行命令
javac -classpath xxx ./myword.java
xxx部分就是上一步里面取到的class path
运行完此命令后， 当前目录下会生成一些.class 文件， 例如：
myword.class myword$IntSumRecer.class myword$TokenizerMapper.class
4. 将class文件打包成.jar文件
运行命令
jar -cvf myword.jar ./*.class
至此, 目标jar 文件成功生成
5. 准备一些文本文件， 上传到hdfs, 以做word count的input
例子：
随意创建一些文本文件， 保存到mapred_test 文件夹
运行命令
hadoop fs -put ./mapred_test/
确保此文件夹成功上传到hdfs 当前用户根目录下
6. 运行我们的程序
运行命令
hadoop jar ./myword.jar myword mapred_test output
顺利的话， 此命令会正常进行， 一个MapRece job 会开始工作， 输出的结果会保存在 hdfs 当前用户根目录下的output 文件夹里面。
至此大功告成！
如果还需要更多的功能， 我们可以修改前面的源文件以达到一个真正有用的MapRece job。
但是原理大同小异， 练手的话， 基本够了。
一个抛砖引玉的简单例子， 欢迎板砖。
转载