关于天云趋势天云趋势由宽带资本和趋势科技共同投资成立于 2010 年 3 月趋势科技是 Hadoop 的重度使用者 : 2006 年开始使用, 用于处理网页和邮件服务器评级五个数据中心, 近 1000 个节点, 最大集群约 500 台服务器日均处理 3.6T 日志数据亚洲最早, 也是最大的

Size: px

Start display at page:

Download "关于天云趋势天云趋势由宽带资本和趋势科技共同投资成立于 2010 年 3 月趋势科技是 Hadoop 的重度使用者 : 2006 年开始使用, 用于处理网页和邮件服务器评级五个数据中心, 近 1000 个节点, 最大集群约 500 台服务器日均处理 3.6T 日志数据亚洲最早, 也是最大的"

救击祝
5 years ago
Views:

1 TCloud Computing Hadoop 开发者培训

2 关于天云趋势天云趋势由宽带资本和趋势科技共同投资成立于 2010 年 3 月趋势科技是 Hadoop 的重度使用者 : 2006 年开始使用, 用于处理网页和邮件服务器评级五个数据中心, 近 1000 个节点, 最大集群约 500 台服务器日均处理 3.6T 日志数据亚洲最早, 也是最大的代码贡献者 HBase 0.92 新功能的主要开发者 (coprocessors, security) 天云趋势提供全面的 BigData 解决方案 TCloud BigData Platform, 集部署监控配置优化数据处理以及扩展 SDK 库于一体的集群管理和开发平台丰富的垂直行业开发经验, 包括电信银行电力安全等为多家大型传统软件开发厂商提供咨询服务可为企业内部进行定制化培训

3 第六章 MapReduce 入门 1. WordCount 功能介绍 2. 开发 MapReduce 程序步骤 3. WordCount 代码分析

4 6.1 WordCount功能介绍 WordCount之于MapReduce 就好比HelloWorld之于C或Java WordCount的输入多个文本文件 WordCount的输出所有出现的单词列表以及每个单词在所有文件中累计出现的次数

5 6.2 开发 MapReduce 程序步骤编写 Mapper 编写 Reducer 使用 JobConf 和 JobClient 提交作业

6 6.3.1 Word Count Mapper import java.io.ioexception; import java.util.stringtokenizer; import org.apache.hadoop.io.longwritable; import org.apache.hadoop.io.text; import org.apache.hadoop.mapred.mapreducebase; import org.apache.hadoop.mapred.mapper; import org.apache.hadoop.mapred.outputcollector; import org.apache.hadoop.mapred.reporter; public class WordCountMapper extends MapReduceBase implements Mapper<LongWritable, Text, Text, LongWritable> { private final static LongWritable one = new LongWritable(1); private Text word = new public void map(longwritable key, Text value,outputcollector<text, LongWritable> output, Reporter reporter) throws IOException { String line = value.tostring(); for( String w : line.split("\\s+") ){ word.set(w); output.collect(word,one);

7 6.3.2 Word Count Mapper 依赖 import java.io.ioexception; import java.util.stringtokenizer; import org.apache.hadoop.io.longwritable; import org.apache.hadoop.io.text; import org.apache.hadoop.mapred.mapreducebase; import org.apache.hadoop.mapred.mapper; import org.apache.hadoop.mapred.outputcollector; import org.apache.hadoop.mapred.reporter; org.apache.hadoop.mapred 及 org.apache.hadoop.io 等是开发 Word Count 需要导入的包下面文档中,Mapper 导入部分不再单独列出

8 6.3.3 Word Count Mapper 类的继承与实现 public class WordCountMapper extends MapReduceBase implements Mapper<LongWritable, Text, Text, LongWritable> { private final static LongWritable one = new LongWritable(1); 用户 Mapper private 类需要继承 Text word MapReduceBase = new Text(); 类及实现 Mapper 接口 Mapper < 输入 key, 输入 value, 输出 key, 输出 value> public void map(longwritable key, Text value, OutputCollector<Text, LongWritable> output, Reporter reporter) throws IOException { String line = value.tostring(); for( String w : line.split("\\s+") ){ word.set(w); output.collect(word,one);

9 6.3.4 Word Count map 方法 public class WordCountMapper extends MapReduceBase implements Mapper<LongWritable, Text, Text, LongWritable> { private final static LongWritable one = new LongWritable(1); private Text word = new public void map(longwritable key, Text value, OutputCollector<Text, LongWritable> output, Reporter reporter) throws IOException { String line = value.tostring(); map 方法包括一个输入 for(string key, w 输入 : line.split("\\s+")){ value,output 和 reporter 四个参数 output 用来做输出, 但是要注意的一点是该类的参数要与 output.collect(word.set(w),one); Mapper 类的输出 key, 输出 value 的类型相同

10 6.3.5 value 类型转换 public class WordCountMapper extends MapReduceBase implements Mapper<LongWritable, Text, Text, LongWritable> { private final static LongWritable one = new LongWritable(1); private Text word = new public void map(longwritable key, Text value, OutputCollector<Text,LongWritable> output, Reporter reporter) throws IOException { String line = value.tostring(); for(string w : line.split("\\s+")){ value 是 Text 对象, 通过 tostring 方法 output.collect(word.set(w),one);, 可以获得 java.lang.string 的对象

11 6.3.6 遍历单词 public class WordCountMapper extends MapReduceBase implements Mapper<LongWritable, Text, Text, LongWritable> { private final static LongWritable one = new LongWritable(1); private Text word = new public void map(longwritable key, Text value, OutputCollector<Text,LongWritable> output, Reporter reporter) throws IOException { String line = value.tostring(); for( String w : line.split("\\s+") ){ output.collect(word.set(w),one); 遍历以空格分割的单词

12 6.3.7 遍历单词 public class WordCountMapper extends MapReduceBase implements Mapper<LongWritable, Text, Text, LongWritable> { private final static LongWritable one = new LongWritable(1); private Text word = new public void map(longwritable key, Text value, OutputCollector<Text, LongWritable> output, Reporter reporter) throws IOException { String line = value.tostring(); for( String w : line.split("\\s+") ){ word.set(w); output.collect(word,one); 调用 output 的 collect 方法这个 key, 是单词本身, 值是 1 需要注意, 输出的 key 及 value 一定要是一个 Writable 的子类

13 6.3.8 Word Count Reducer import java.io.ioexception; import java.util.iterator; import org.apache.hadoop.io.longwritable; import org.apache.hadoop.io.text; import org.apache.hadoop.mapred.mapreducebase; import org.apache.hadoop.mapred.outputcollector; import org.apache.hadoop.mapred.reducer; import org.apache.hadoop.mapred.reporter; public class WordCountReducer extends MapReduceBase implements Reducer<Text, LongWritable, Text, public void reduce(text key, Iterator<LongWritable> values, OutputCollector<Text, LongWritable> output, Reporter reporter) throws IOException { long sum = 0; while( values.hasnext() ) { sum += values.next().get(); output.collect(key, new LongWritable(sum));

14 6.3.9 Word Count Reducer 依赖 import java.io.ioexception; import java.util.iterator; import org.apache.hadoop.io.longwritable; import org.apache.hadoop.io.text; import org.apache.hadoop.mapred.mapreducebase; import org.apache.hadoop.mapred.outputcollector; import org.apache.hadoop.mapred.reducer; import org.apache.hadoop.mapred.reporter; org.apache.hadoop.mapred 及 org.apache.hadoop.io 是开发 Word Count 需要导入的包下面文档中,Reducer 导入部分不再单独列出

15 Word Count Reducer 类的继承与实现 public class WordCountReducer extends MapReduceBase implements Reducer<Text, LongWritable, Text, LongWritable>{ 用户类需要继承 MapReduceBase 类及实现 Reducer 接口 public void reduce(text key, Iterator<LongWritable> values,outputcollector<text, Reducer 接口有四个泛型 < 输入 Key, 输入 LongWritable> value, 输出 key, output, 输出 value> Reporter reporter) 需要注意的是这里面的输入 throws IOException key 及输入 value { 要与 Mapper 的输出 key 与输出 value 类型一致 long sum = 0; while( values.hasnext() ) { sum += values.next().get(); output.collect(key, new LongWritable(sum));

16 Word Count reduce 方法 public class WordCountReducer extends MapReduceBase implements Reducer<Text, LongWritable, Text, public void reduce(text key, Iterator<LongWritable> values, OutputCollector<Text, LongWritable> output, Reporter reporter) throws IOException { long sum = 0; reduce 方法包括一个输入 while key, (values.hasnext()) 输入集合,output{ 和 reporter 四个参数 sum += values.next().get(); output 用来做输出, 最终输出到 HDFS 但是要注意的一点是该类的参数要与 Reducer 类的输出 key, 输出 value output.collect(key, 的类型相同 new LongWritable(sum));

17 Word Count reduce 方法核心逻辑 public class WordCountReducer extends MapReduceBase implements Reducer<Text, LongWritable, Text, public void reduce(text key, Iterator<LongWritable> values, OutputCollector<Text, LongWritable> output, Reporter reporter) throws IOException { long sum = 0; while( values.hasnext() ) { sum += values.next().get(); output.collect(key, new LongWritable(sum)); 使用 values 对象的 hasnext() 与 next() 方法来遍历集合中所有的元素通过 next().get() 拿到单词的数量进行累加, 算出词频

18 Word Count 最终结果输出到文件 public class WordCountReducer extends MapReduceBase implements Reducer<Text, LongWritable, Text, public void reduce(text key, Iterator<LongWritable> values, OutputCollector<Text, LongWritable> output, Reporter reporter) throws IOException { long sum = 0; while( values.hasnext() ) { sum += values.next().get(); output.collect(key, new LongWritable(sum)); 核心逻辑处理完后, 用 output.collect 方法来输出单词与词频到 HDFS 文件单词与词频以 TAB 分割

19 使用 JobConf 和 JobClient 提交作业 public class WordCount { public static void main(string[] args) throws Exception { if (args.length!= 2) { System.err.println("Usage: WordCount <input path> <output path>"); System.exit(-1); Configuration conf= new Configuration(); JobConf jobconf = new JobConf(conf, WordCount.class); conf.setjobname("wordcount"); jobconf.setmapperclass(wordcountmapper.class); jobconf.setreducerclass(wordcountreducer.class); jobconf.setoutputkeyclass(text.class); jobconf.setoutputvalueclass(longwritable.class); FileInputFormat.setInputPaths(jobConf, new Path(args[0])); FileOutputFormat.setOutputPath(jobConf, new Path(args[1])); JobClient.runJob(jobConf);

20 Tool 与 ToolRunner 使用 Hadoop 自带的工具类可以方便的处理命令行参数用户程序实现 Tool 接口, 包括 getconf, setconf, run 方法 ToolRunner 是运行 Tool 实例的帮助类在命令行, 可以通过 -conf 参数, 方便指定配置文件

21 使用 ToolRunner public class WordCount extends Configured implements Tool public int run (String[] args) throws Exception { // 参数检查略, 同之前的 main JobConf conf = new JobConf(getConf(), WordCount.class); conf.setjobname("wordcount"); // 设置 Mapper 和 Reducer 类, 设置输入输出格式和路径, 略, 同之前的 main JobClient.runJob(conf); return 0; public static void main (String[] args) throws Exception{ int exitcode = ToolRunner.run (new WordCount(),args); System.exit (exitcode);

22 第十章 MapReduce 高级特性 1. MapReduce 单元测试 2. 新旧 API 对比 3. 如何用其它语言开发 MapReduce 应用 4. MapReduce 常用算法介绍

23 新旧 API 在 ( 不包含 ) 以前版本的 MapReduce API, 称为旧 API 0.20.x 以及后续的版本, 新建立了一套 MapReduce API, 称为新 API 旧 API 依然存在旧 API 与新 API 类型完全不兼容新 API 没有完全实现旧 API 的功能

24 新旧 API 主要区别 (1) 用抽象类还是接口旧 API 使用的是接口新 API 使用的是抽象类, 抽象类易于功能扩展, 而不会影响客户端代码抽象类实现了通用的功能 Mapper 及 Reducer 通信对象旧 API 的类 JobConf OutputCollector Reporter 新 API 使用 Context( 上下文 ) 对象, 同 MapReduce 框架通信

25 接口 ( 抽象类 ) 通信对象示例旧 API public class OldMapper extends MapReduceBase implements Mapper < >{ public void map(keytype k,valuetype v,outputcollector output,reporter report){ output.collect(k,v); 新 API public class NewMapper extends Mapper < >{ public void map(keytype k, ValueType v, Context context){ context.write(k,v);

26 新旧 API 主要区别 (2) 任务控制旧 API 的类使用 JobClient 新 API 的类使用 Job 配置对象旧 API 使用 JobConf 对象,JobConf 继承 Configuration 新 API, 直接使用 Configuration 配置任务参数

27 任务控制配置对象示例旧 API JobConf conf = new JobConf(new Configuration(), OldDriver.class); conf.setsomeproperty(...);... JobClient.runJob(conf); 新 API Configuration conf = new Configuration(); Job job = new Job(conf); job.setjarbyclass(newdriver.class); job.setsomeproperty(...);... job.waitforcompletion(true);

28 新旧 API 主要区别 (3) 输出文件命名旧 API 的 Mapper 及 Reducer 输出文件名格式 part-nnnnn 新 API 的 Mapper 输出文件名格式为 part-m-nnnnn,reducer 输出文件名格式为 part-r-nnnnn nnnnn 从 0 开始计数控制执行过程旧 API 的 Mapper 可以实现 MapRunnable 接口, 实现控制执行过程, 但 Reducer 不存在 ReduceRunnable 接口新 API 可以重写 Mapper 或 Reducer 的 run 方法, 控制执行过程

29 新旧 API 主要区别 (4) 异常处理旧 API 的 map,reduce 方法, 只抛出 IOException 新 API 的 map,reduce 方法, 抛出 IOException,InterruptedException, 可以使框架很优雅的取消长时间运行的操作包的位置旧 API 在 org.apache.hadoop.mapred 包 ( 及子包 ) 下新 API 在 org.apache.hadoop.mapreduce 包 ( 及子包 ) 下 reduce 方法遍历 value 方式旧 API 的 reduce 方法的参数使用 java.lang.iterator, 只能使用 hasnext,next 方法遍历新 API 的 reduce 方法的参数使用 java.lang.iterable, 可以方便使用 foreach 循环操作 value

30 reduce 方法遍历 value 方式旧 API public void reducer( KeyType k, Iterator<ValueType> v, OutputCollector output,reporter report){ while(v.hasnext()){ output.collect(k,v.next()); 新 API public void reducer(keytype k, Iterable<ValueType> values, Context context){ for(valuetype v:values){ context.write(k,v));

31 为什么我们推荐使用老 API 新版本的 API 没有完全实现旧版本的功能 ChainMapper ChainReducer MultiFileInputFormat 与 MultiFileOutputFormat SequenceFileAsBinaryInputFormat 与 SequenceFileAsBinaryOutputFormat... 在 Hadoop 2.0 版本成熟后再切换到新 API

Hadoop 集群 ( 第 6 期 ) WordCount 运行详解 1 MapReduce 理论简介 1.1 MapReduce 编程模型 MapReduce 采用分而治之的思想, 把对大规模数据集的操作, 分发给一个主节点管

Hadoop 集群 ( 第 6 期 ) WordCount 运行详解 1 MapReduce 理论简介 1.1 MapReduce 编程模型 MapReduce 采用分而治之的思想, 把对大规模数据集的操作, 分发给一个主节点管细细品味 Hadoop Hadoop 集群 ( 第 6 期 ) 精华集锦 csaxp 虾皮工作室 http://www.cnblogs.com/xia520pi/ 2012 年 5 月 15 日 Hadoop 集群 ( 第 6 期 ) WordCount 运行详解 1 MapReduce 理论简介 1.1 MapReduce 编程模型 MapReduce 采