mapreducejoin_育才学习网

mapreduce怎么写

1.如何在Hadoop上编写MapReduce程序

用户配置并将一个Hadoop作业提到Hadoop框架中，Hadoop框架会把这个作业分解成一系列map tasks 和reduce tasks。

Hadoop框架负责task分发和执行，结果收集和作业进度监控。在编写MapReduce程序时，用户分别通过InputFormat和OutputFormat指定输入和输出格式，并定义Mapper和Reducer指定map阶段和reduce阶段的要做的工作。

在Mapper或者Reducer中，用户只需指定一对key/value的处理逻辑，Hadoop框架会自动顺序迭代解析所有key/value，并将每对key/value交给Mapper或者Reducer处理。表面上看来，Hadoop限定数据格式必须为key/value形式，过于简单，很难解决复杂问题，实际上，可以通过组合的方法使key或者value（比如在key或者value中保存多个字段，每个字段用分隔符分开，或者value是个序列化后的对象，在Mapper中使用时，将其反序列化等）保存多重信息，以解决输入格式较复杂的应用。

2.2 用户的工作用户编写MapReduce需要实现的类或者方法有：（1） InputFormat接口用户需要实现该接口以指定输入文件的内容格式。该接口有两个方法 public interface InputFormat { InputSplit[] getSplits(JobConf job, int numSplits) throws IOException; RecordReader getRecordReader(InputSplit split, JobConf job, Reporter reporter) throws IOException； } 其中getSplits函数将所有输入数据分成numSplits个split，每个split交给一个map task处理。

getRecordReader函数提供一个用户解析split的迭代器对象，它将split中的每个record解析成key/value对。Hadoop本身提供了一些InputFormat:(2)Mapper接口用户需继承Mapper接口实现自己的Mapper,Mapper中必须实现的函数是 void map(K1 key, V1 value, OutputCollector output, Reporter reporter) throws IOException 其中，是通过Inputformat中的RecordReader对象解析处理的，OutputCollector获取map（)的输出结果，Reporter保存了当前task处理进度。

Hadoop本身提供了一些Mapper供用户使用：（3）Partitioner接口用户需继承该接口实现自己的Partitioner以指定map task产生的key/value对交给哪个reduce task处理，好的Partitioner能让每个reduce task处理的数据相近，从而达到负载均衡。Partitioner中需实现的函数是 getPartition( K2 key, V2 value, int numPartitions) 该函数返回对应的reduce task ID。

用户如果不提供Partitioner,Hadoop会使用默认的（实际上是个hash函数）。（4）Combiner Combiner使得map task与reduce task之间的数据传输量大大减小，可明显提高性能。

大多数情况下，Combiner与Reducer相同。（5）Reducer接口用户需继承Reducer接口实现自己的Reducer,Reducer中必须实现的函数是 void reduce(K2 key, Iterator values, OutputCollector output, Reporter reporter) throws IOException Hadoop本身提供了一些Reducer供用户使用：（6）OutputFormat 用户通过OutputFormat指定输出文件的内容格式，不过它没有split。

每个reduce task将其数据写入自己的文件，文件名为part-nnnnn，其中nnnnn为reduce task的ID。Hadoop本身提供了几个OutputFormat:3. 分布式缓存 Haoop中自带了一个分布式缓存，即DistributedCache对象，方便map task之间或者reduce task之间共享一些信息，比如某些实际应用中，所有map task要读取同一个配置文件或者字典，则可将该配置文件或者字典放到分布式缓存中。

4. 多语言编写MapReduce作业 Hadoop采用java编写，因而Hadoop天生支持java语言编写作业，但在实际应用中，有时候，因要用到非java的第三方库或者其他原因，要采用C/C++或者其他语言编写MapReduce作业，这时候可能要用到Hadoop提供的一些工具。如果你要用C/C++编写MpaReduce作业，可使用的工具有Hadoop Streaming或者Hadoop Pipes。

如果你要用Python编写MapReduce作业，可以使用Hadoop Streaming或者Pydoop。如果你要使用其他语言，如shell,php,ruby等，可使用Hadoop Streaming。

关于Hadoop Streaming编程，可参见我的这篇博文：《Hadoop Streaming编程》（/projects/pydoop/ 关于Hadoop pipes编程，可参见《Hadoop Tutorial 2.2 — Running C++ Programs on Hadoop》。5. 编程方式比较（1）java。

Hadoop支持的最好最全面的语言，而且提供了很多工具方便程序员开发。（2）Hadoop Streaming。

它最大的优点是支持多种语言，但效率较低，reduce task需等到map 阶段完成后才能启动；它不支持用户自定义InputFormat，如果用户想指定输入文件格式，可使用java语言编写或者在命令行中指定分隔符；它采用标准输入输出让C/C++与java通信，因而只支持text数据格式。（3）Hadoop Pipes。

专门为C/C++语言设计，由于其采用了socket方式让C/C++与java通信，因而其效率较低（其优势在于，但作业需要大量，速度很快）。它支持用户（用C/C++）编写RecordReader。

(4)Pydoop。它是专门方便python。

2.怎么用Python写mapreduce,请举例说明,初学者,请赐教,不胜感激

1.lambda# 匿名函数# 基本用法 lambda x: x**2 # 第一个参数，然后是表达式# 也可以使用如下（lambda x: x**2）(5)2. map()def map(function, sequence, *sequence_1): # real signature unknown; restored from __doc__ """ map(function, sequence[, sequence, 。

]) -> list Return a list of the results of applying the function to the items of the argument sequence(s). If more than one sequence is given, the function is called with an argument list consisting of the corresponding item of each sequence, substituting None for missing values when not all sequences have the same length. If the function is None, return a list of the items of the sequence (or a list of tuples if more than one sequence). """ return []# 两个参数，一个处理函数，一个可迭代的序列# 返回一个列表# 例如计算1到10的平方，并以列表的形式返回map(lambda x: x**2, range(1, 11)）# 结果如下[1, 4, 9, 16, 25, 36, 49, 64, 81, 100]# 当然也可以如下这样使用def square(x): return x**2map(square, range(1, 11))3.reduce()def reduce(function, sequence, initial=None): # real signature unknown; restored from __doc__ """ reduce(function, sequence[, initial]) -> value Apply a function of two arguments cumulatively to the items of a sequence, from left to right, so as to reduce the sequence to a single value. For example, reduce(lambda x, y: x+y, [1, 2, 3, 4, 5]) calculates ((((1+2)+3)+4)+5). If initial is present, it is placed before the items of the sequence in the calculation, and serves as a default when the sequence is empty. """ pass# 两个参数，一个接受两个参数的函数，一个序列参数# 例如计算 1到10 的和reduce(lambda x, y: x+y, range(1, 11)）# 当然，不适用lambda匿名函数也可以def add(x, y): return x+yreduce(add, range(1, 11)）# 结果如下454.filter()def filter(function_or_none, sequence): # known special case of filter """ filter(function or None, sequence) -> list, tuple, or string Return those items of sequence for which function(item) is true. If function is None, return the items that are true. If sequence is a tuple or string, return the same type, else return a list. """ pass# 接受两个参数，一个过滤函数，返回True 或者 False，以及一个序列# 例如，计算100以内的偶数filter(lambda x: x % 2 == 0, range(100)）# 如上def div2(x): if x % 2 == 0: return True else: return Falsefilter(div2, range(100)）# 结果如下 [0, 2, 4, 6, 8, 10, 12, 14, 16, 。 ]。

3.怎么用Python写mapreduce,请举例说明,初学者,请赐教,不胜感激

1.lambda # 匿名函数# 基本用法 lambda x: x**2 # 第一个参数，然后是表达式# 也可以使用如下（lambda x: x**2）(5)2. map() def map(function, sequence, *sequence_1): # real signature unknown; restored from __doc__ """ map(function, sequence[, sequence, 。

]) -> list Return a list of the results of applying the function to the items of the argument sequence(s). If more than one sequence is given, the function is called with an argument list consisting of the corresponding item of each sequence, substituting None for missing values when not all sequences have the same length. If the function is None, return a list of the items of the sequence (or a list of tuples if more than one sequence). """ return []# 两个参数，一个处理函数，一个可迭代的序列# 返回一个列表# 例如计算1到10的平方，并以列表的形式返回map(lambda x: x**2, range(1, 11)）# 结果如下[1, 4, 9, 16, 25, 36, 49, 64, 81, 100]# 当然也可以如下这样使用def square(x): return x**2map(square, range(1, 11))3.reduce() def reduce(function, sequence, initial=None): # real signature unknown; restored from __doc__ """ reduce(function, sequence[, initial]) -> value Apply a function of two arguments cumulatively to the items of a sequence, from left to right, so as to reduce the sequence to a single value. For example, reduce(lambda x, y: x+y, [1, 2, 3, 4, 5]) calculates ((((1+2)+3)+4)+5). If initial is present, it is placed before the items of the sequence in the calculation, and serves as a default when the sequence is empty. """ pass# 两个参数，一个接受两个参数的函数，一个序列参数# 例如计算 1到10 的和reduce(lambda x, y: x+y, range(1, 11)）# 当然，不适用lambda匿名函数也可以def add(x, y): return x+yreduce(add, range(1, 11)）# 结果如下454.filter() def filter(function_or_none, sequence): # known special case of filter """ filter(function or None, sequence) -> list, tuple, or string Return those items of sequence for which function(item) is true. If function is None, return the items that are true. If sequence is a tuple or string, return the same type, else return a list. """ pass# 接受两个参数，一个过滤函数，返回True 或者 False，以及一个序列# 例如，计算100以内的偶数filter(lambda x: x % 2 == 0, range(100)）# 如上def div2(x): if x % 2 == 0: return True else: return Falsefilter(div2, range(100)）# 结果如下 [0, 2, 4, 6, 8, 10, 12, 14, 16, 。 ]。

mapreduce怎么写