大连公司地址,seo网络优化推广,网页制作模板兼职,富阳做网站的1 以词频统计为例子介绍 mapreduce怎么写出来的
弄清楚MapReduce的各个过程#xff1a; 将文件输入后#xff0c;返回的k1,v1代表的含义是#xff1a;k1表示偏移量#xff0c;即v1的第一个字母在文件中的索引#xff08;从0开始数的#xff09;#xff1b;v1表…1 以词频统计为例子介绍 mapreduce怎么写出来的
弄清楚MapReduce的各个过程 将文件输入后返回的k1,v1代表的含义是k1表示偏移量即v1的第一个字母在文件中的索引从0开始数的v1表示对应的一整行的值
map阶段将每一行的内容按照空格进行分割后作为k2将v2的值写为1后输出
reduce阶段将相同的k2合并后输出
1.1 创建Mapper、Reducer、Driver类
创建这三种类用的是一种方法用Mapper举例如下 注意选择父类 1.2 map阶段代码书写
1mapper源码
本来可以按住ctrl键后点击open 后查看mapper源代码但是在虚拟机里一直调不出来。所以从网上搜索出具体代码如下
/*** Licensed to the Apache Software Foundation (ASF) under one* or more contributor license agreements. See the NOTICE file* distributed with this work for additional information* regarding copyright ownership. The ASF licenses this file* to you under the Apache License, Version 2.0 (the* License); you may not use this file except in compliance* with the License. You may obtain a copy of the License at** http://www.apache.org/licenses/LICENSE-2.0** Unless required by applicable law or agreed to in writing, software* distributed under the License is distributed on an AS IS BASIS,* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.* See the License for the specific language governing permissions and* limitations under the License.*/package org.apache.hadoop.mapreduce;import java.io.IOException;import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.RawComparator;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.mapreduce.task.MapContextImpl;InterfaceAudience.Public
InterfaceStability.Stable
public class MapperKEYIN, VALUEIN, KEYOUT, VALUEOUT {/*** The codeContext/code passed on to the {link Mapper} implementations.*/public abstract class Contextimplements MapContextKEYIN,VALUEIN,KEYOUT,VALUEOUT {}/*** Called once at the beginning of the task.*/protected void setup(Context context) throws IOException, InterruptedException {// NOTHING}/*** Called once for each key/value pair in the input split. Most applications* should override this, but the default is the identity function.*/SuppressWarnings(unchecked)protected void map(KEYIN key, VALUEIN value, Context context) throws IOException, InterruptedException {context.write((KEYOUT) key, (VALUEOUT) value);}/*** Called once at the end of the task.*/protected void cleanup(Context context) throws IOException, InterruptedException {// NOTHING}/*** Expert users can override this method for more complete control over the* execution of the Mapper.* param context* throws IOException*/public void run(Context context) throws IOException, InterruptedException {setup(context);try {while (context.nextKeyValue()) {map(context.getCurrentKey(), context.getCurrentValue(), context);}} finally {cleanup(context);}}
}
2修改的注意事项
注意我们需要修改的只是map方法
1. Mapper组件开发方式自定义一个类继承Mapper
2. Mapper组件的作用是定义每一个MapTask具体要怎么处理数据。例如一个文件256MB会生成2个MapTask(每个切片大小,默认是128MB,所以MapTask的多少有处理的数据大小来决定)。即2个MapTask处理逻辑是一样的只是每个MapTask处理的数据不一样。
3. 下面是Mapper类中的4个泛型含义:a.泛型一:KEYINLongWritable对应的Mapper的输入key。输入key是每行的行首偏移量b.泛型二: VALUEINText对应的Mapper的输入Value。输入value是每行的内容c.泛型三:KEYOUT对应的Mapper的输出key根据业务来定义d.泛型四:VALUEOUT对应的Mapper的输出value根据业务来定义
4. 注意初学时KEYIN和VALUEIN写死(LongWritable,Text)。KEYOUT和VALUEOUT不固定,根据业务来定
5. Writable机制是Hadoop自身的序列化机制常用的类型a. LongWritable b. Text(String)c. IntWritabled. NullWritable
6. 定义MapTask的任务逻辑是通过重写map()方法来实现的。
读取一行数据就会调用一次此方法同时会把输入key和输入value进行传递
7. 在实际开发中最重要的是拿到输入value(每行内容)
8. 输出方法通过context.write(输出key输出value)
9. 开发一个MapReduce程序jobMapper可以单独存储此时最后的输出的结果文件内容就是Mapper的输出。
10. Reducer组件不能单独存在因为Reducer要依赖于Mapper的输出。当引入了Reducer之后最后输出的结果文件的结果就是Reducer的输出。3具体实例
重写map方法:输入map后 按住alt加 后就可以自动补全代码!
然后进行编写
import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;public class WordMapper extends MapperLongWritable, Text, Text, IntWritable {Overrideprotected void map(LongWritable key, Text value, MapperLongWritable, Text, Text,IntWritable.Context context)throws IOException, InterruptedException {//将value转换成字符串再将其转化成字符串数组String line value.toString(); //hello wordString[] wordarr line.split( );for (String word:wordarr) {context.write(new Text(word), new IntWritable(1));} }} 1.3 reducer阶段代码的书写
1reducer源码
和mapper差不多
2修改时的注意事项
1. Reducer组件用于接收Mapper组件的输出
2. reduce的输入key,value需要和mapper的输出key,value类型保持一致
3. reduce的输出key,value类型根据具体业务决定
4. reduce收到map的输出会按相同的key做聚合
形成:key Iterable 形式然后通过reduce方法进行传递
5. reduce方法中的Iterable是一次性的即遍历一次之后再遍历里面就没有数据了。
所以在某些业务场景会涉及到多次操作此迭代器处理的方法是
①先创建一个List ②把Iterable装到List ③多次去使用List即可3具体案例
注意IntWriter是一个迭代器context负责输出
import java.io.IOException;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.examples.SecondarySort.Reduce;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;public class WordReducer extends ReducerText, IntWritable, Text, IntWritable {Overrideprotected void reduce(Text key, IterableIntWritable values,ReducerText, IntWritable, Text, IntWritable.Context context) throws IOException, InterruptedException {int total 0; for (IntWritable value:values) {total total value.get(); }context.write(key, new IntWritable(total));}1.4 主函数代码的书写
【1】还未进行reducer阶段时
1主函数也就是驱动函数一般包含以下几个阶段
注意实例化job、设置输入文件地址、输出文件地址。这三个代码是固定的每次都这样哦
import java.io.IOException;
public class WordDriver {public static void main(String[] args) throws Exception {//1.实例化jobConfiguration conf new Configuration();String[] otherArgs (new GenericOptionsParser(conf, args)).getRemainingArgs();if(otherArgs.length 2) {System.err.println(Usage: wordcount in [in...] out);System.exit(2);}Job job Job.getInstance(conf, word count);//2.关联class文件job.setJarByClass(WordDriver.class);job.setMapperClass(WordMapper.class);//3.设置mapper的输出数据类型job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(IntWritable.class);//4.设置reducer的是输出数据类型//5.设置输入文件路径for(int i 0; i otherArgs.length - 1; i) {FileInputFormat.addInputPath(job, new Path(otherArgs[i]));}//6.设置输出文件路径FileOutputFormat.setOutputPath(job, new Path(otherArgs[otherArgs.length - 1])); //7.提交job文件System.exit(job.waitForCompletion(true)?0:1);}}输出的结果为 就是我们map阶段应该产生的结果
【2】进行reducer阶段后
import java.io.IOException;
public class WordDriver {public static void main(String[] args) throws Exception {//1.实例化jobConfiguration conf new Configuration();String[] otherArgs (new GenericOptionsParser(conf, args)).getRemainingArgs();if(otherArgs.length 2) {System.err.println(Usage: wordcount in [in...] out);System.exit(2);}Job job Job.getInstance(conf, word count);//2.关联class文件job.setJarByClass(WordDriver.class);job.setMapperClass(WordMapper.class);job.setReducerClass(WordReducer.class);//3.设置mapper的输出数据类型job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(IntWritable.class);//4.设置reducer的是输出数据类型job.setOutputKeyClass(Text.class);job.setOutputValueClass(IntWritable.class);//5.设置输入文件路径for(int i 0; i otherArgs.length - 1; i) {FileInputFormat.addInputPath(job, new Path(otherArgs[i]));}//6.设置输出文件路径FileOutputFormat.setOutputPath(job, new Path(otherArgs[otherArgs.length - 1])); //7.提交job文件System.exit(job.waitForCompletion(true)?0:1);}}1.5 在Ubuntu上运行
1.5.1 编译打包程序
现在就可以编译上面编写的代码。可以直接点击Eclipse工作界面上部的运行程序的快捷按钮当把鼠标移动到该按钮上时在弹出的菜单中选择“Run as”继续在弹出来的菜单中选择“Java Application”如下图所示。 然后会弹出如下图所示界面。 点击界面右下角的“OK”按钮开始运行程序。程序运行结束后会在底部的“Console”面板中显示运行结果信息如下图所示。 下面就可以把Java应用程序打包生成JAR包部署到Hadoop平台上运行。现在可以把词频统计程序放在“/usr/local/hadoop/myapp”目录下。如果该目录不存在可以使用如下命令创建
cd /usr/local/hadoop
mkdir myapp
首先请在Eclipse工作界面左侧的“Package Explorer”面板中在工程名称“WordCount”上点击鼠标右键在弹出的菜单中选择“Export”如下图所示。 然后会弹出如下图所示界面。 在该界面中选择“Runnable JAR file”然后点击“Next”按钮弹出如下图所示界面。 在该界面中“Launch configuration”用于设置生成的JAR包被部署启动时运行的主类需要在下拉列表中选择刚才配置的类“WordCount-WordCount”。在“Export destination”中需要设置JAR包要输出保存到哪个目录比如这里设置为“/usr/local/hadoop/myapp/WordCount.jar”。在“Library handling”下面选择“Extract required libraries into generated JAR”。然后点击“Finish”按钮会出现如下图所示界面。 可以忽略该界面的信息直接点击界面右下角的“OK”按钮启动打包过程。打包过程结束后会出现一个警告信息界面如下图所示。 可以忽略该界面的信息直接点击界面右下角的“OK”按钮。至此已经顺利把WordCount工程打包生成了WordCount.jar。可以到Linux系统中查看一下生成的WordCount.jar文件可以在Linux的终端中执行如下命令
cd /usr/local/hadoop/myapp
ls
1.5.2 运行程序
在运行程序之前需要启动Hadoop命令如下
cd /usr/local/hadoop
./sbin/start-dfs.sh
在启动Hadoop之后需要首先删除HDFS中与当前Linux用户hadoop对应的input和output目录即HDFS中的“/user/hadoop/input”和“/user/hadoop/output”目录这样确保后面程序运行不会出现问题具体命令如下
cd /usr/local/hadoop
./bin/hdfs dfs -rm -r input
./bin/hdfs dfs -rm -r output
然后再在HDFS中新建与当前Linux用户hadoop对应的input目录即“/user/hadoop/input”目录具体命令如下
cd /usr/local/hadoop
./bin/hdfs dfs -mkdir input
然后把之前在中在Linux本地文件系统中新建的文件wordfile1.txt假设这个文件位于“/usr/local/hadoop”目录下并且里面包含了一些英文语句上传到HDFS中的“/user/hadoop/input”目录下命令如下
cd /usr/local/hadoop
./bin/hdfs dfs -put ./wordfile1.txt input如果HDFS中已经存在目录“/user/hadoop/output”则使用如下命令删除该目录
cd /usr/local/hadoop
./bin/hdfs dfs -rm -r /user/hadoop/output
现在就可以在Linux系统中使用hadoop jar命令运行程序命令如下
cd /usr/local/hadoop
./bin/hadoop jar ./myapp/WordDriver.jar input output
上面命令执行以后当运行顺利结束时屏幕上会显示类似如下的信息 词频统计结果已经被写入了HDFS的“/user/hadoop/output”目录中可以执行如下命令查看词频统计结果
cd /usr/local/hadoop
./bin/hdfs dfs -cat output/*
上面命令执行后会在屏幕上显示如下词频统计结果
Hadoop 2
I 2
Spark 2
fast 1
good 1
is 2
love 2
至此词频统计程序顺利运行结束。需要注意的是如果要再次运行WordCount.jar需要首先删除HDFS中的output目录否则会报错。
最后关闭hadoop程序
cd /usr/local/hadoop
./sbin/stop-dfs.sh
