当前位置：文档之家 › educoder平台MapReduce基础实战

educoder平台MapReduce基础实战

  • 格式：doc
  • 大小：45.50 KB
  • 文档页数：12

MapReduce第1关：成绩统计过关代码：import java.io.IOException;import java.util.StringTokenizer;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.*;import org.apache.hadoop.io.Text;import org.apache.hadoop.mapreduce.Job;import org.apache.hadoop.mapreduce.Mapper;import org.apache.hadoop.mapreduce.Reducer;import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;import org.apache.hadoop.util.GenericOptionsParser;public class WordCount {/********** Begin **********/public static class TokenizerMapper extends Mapper<LongWritable, Text, Text, IntWritable> {private final static IntWritable one = new IntWritable(1);private Text word = new Text();private int maxValue = 0;public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {StringTokenizer itr = new StringTokenizer(value.toString(),"\n");while (itr.hasMoreTokens()) {String[] str = itr.nextToken().split(" ");String name = str[0];one.set(Integer.parseInt(str[1]));word.set(name);context.write(word,one);}//context.write(word,one);}}public static class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable> {private IntWritable result = new IntWritable();public void reduce(Text key, Iterable<IntWritable> values, Context context)throws IOException, InterruptedException {int maxAge = 0;int age = 0;for (IntWritable intWritable : values) {maxAge = Math.max(maxAge, intWritable.get());}result.set(maxAge);context.write(key, result);}public static void main(String[] args) throws Exception {Configuration conf = new Configuration();Job job = new Job(conf, "word count");job.setJarByClass(WordCount.class);job.setMapperClass(TokenizerMapper.class);job.setCombinerClass(IntSumReducer.class);job.setReducerClass(IntSumReducer.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(IntWritable.class);String inputfile = "/user/test/input";String outputFile = "/user/test/output/";FileInputFormat.addInputPath(job, new Path(inputfile));FileOutputFormat.setOutputPath(job, new Path(outputFile)); job.waitForCompletion(true);/********** End **********/}}命令行touch file01echo Hello World Bye Worldcat file01echo Hello World Bye World >file01touch file02echo Hello Hadoop Goodbye Hadoop >file02cat file02start-dfs.shhadoop fs -mkdir /usrhadoop fs -mkdir /usr/inputhadoop fs -ls /usr/outputhadoop fs -ls /hadoop fs -ls /usrhadoop fs -put file01 /usr/inputhadoop fs -put file02 /usr/inputhadoop fs -ls /usr/input测评——————————————————————————————————MapReduce第2关：文件内容合并去重代码import java.io.IOException;import java.util.*;import org.apache.hadoop.conf.Configuration;import org.apache.hadoop.fs.Path;import org.apache.hadoop.io.*;import org.apache.hadoop.mapreduce.Job;import org.apache.hadoop.mapreduce.Mapper;import org.apache.hadoop.mapreduce.Reducer;import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;import org.apache.hadoop.util.GenericOptionsParser;public class Merge {/*** @param args* 对A,B两个文件进行合并，并剔除其中重复的内容，得到一个新的输出文件C*///在这重载map函数，直接将输入中的value复制到输出数据的key上注意在map方法中要抛出异常：throws IOException,InterruptedException/********** Begin **********/public static class Map extends Mapper<LongWritable, Text, Text, Text >{protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context)throws IOException, InterruptedException {String str = value.toString();String[] data = str.split(" ");Text t1= new Text(data[0]);Text t2 = new Text(data[1]);context.write(t1,t2);}}/********** End **********///在这重载reduce函数，直接将输入中的key复制到输出数据的key上注意在reduce 方法上要抛出异常：throws IOException,InterruptedException/********** Begin **********/public static class Reduce extends Reducer<Text, Text, Text, Text>{protected void reduce(Text key, Iterable<Text> values, Reducer<Text, Text, Text, Text>.Context context)throws IOException, InterruptedException {List<String> list = new ArrayList<>();for (Text text : values) {String str = text.toString();if(!list.contains(str)){list.add(str);}}Collections.sort(list);for (String text : list) {context.write(key, new Text(text));}}/********** End **********/}public static void main(String[] args) throws Exception{Configuration conf = new Configuration();Job job = new Job(conf, "word count");job.setJarByClass(Merge.class);job.setMapperClass(Map.class);job.setCombinerClass(Reduce.class);job.setReducerClass(Reduce.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(Text.class);String inputPath = "/user/tmp/input/"; //在这里设置输入路径String outputPath = "/user/tmp/output/"; //在这里设置输出路径FileInputFormat.addInputPath(job, new Path(inputPath));FileOutputFormat.setOutputPath(job, new Path(outputPath));System.exit(job.waitForCompletion(true) ? 0 : 1);}}测评———————————————————————————————————————MapReduce第3关：信息挖掘 - 挖掘父子关系代码import java.io.IOException;import java.util.*;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.mapreduce.Job;import org.apache.hadoop.mapreduce.Mapper;import org.apache.hadoop.mapreduce.Reducer;import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;import org.apache.hadoop.util.GenericOptionsParser;public class simple_data_mining {public static int time = 0;/*** @param args* 输入一个child-parent的表格* 输出一个体现grandchild-grandparent关系的表格*///Map将输入文件按照空格分割成child和parent，然后正序输出一次作为右表，反序输出一次作为左表，需要注意的是在输出的value中必须加上左右表区别标志public static class Map extends Mapper<Object, Text, Text, Text>{public void map(Object key, Text value, Context context) throws IOException,InterruptedException{/********** Begin **********/String child_name = new String();String parent_name = new String();String relation_type = new String();String line = value.toString();int i = 0;while(line.charAt(i) != ' '){i++;}String[] values = {line.substring(0,i),line.substring(i+1)};if(values[0].compareTo("child") != 0){child_name = values[0];parent_name = values[1];relation_type = "1";//左右表区分标志context.write(new Text(values[1]), new Text(relation_type+"+"+child_name+"+"+parent_name));//左表relation_type = "2";context.write(new Text(values[0]), new Text(relation_type+"+"+child_name+"+"+parent_name));//右表}/********** End **********/}}public static class Reduce extends Reducer<Text, Text, Text, Text>{public void reduce(Text key, Iterable<Text> values,Context context) throws IOException,InterruptedException{/********** Begin **********/if(time == 0){ //输出表头context.write(new Text("grand_child"), new Text("grand_parent"));time++;}int grand_child_num = 0;String grand_child[] = new String[10];int grand_parent_num = 0;String grand_parent[]= new String[10];Iterator ite = values.iterator();while(ite.hasNext()){String record = ite.next().toString();int len = record.length();int i = 2;if(len == 0) continue;char relation_type = record.charAt(0);String child_name = new String();String parent_name = new String();//获取value-list中value的childwhile(record.charAt(i) != '+'){child_name = child_name + record.charAt(i);i++;}i=i+1;//获取value-list中value的parentwhile(i<len){parent_name = parent_name+record.charAt(i);i++;}//左表，取出child放入grand_childif(relation_type == '1'){grand_child[grand_child_num] = child_name;grand_child_num++;}else{//右表，取出parent放入grand_parentgrand_parent[grand_parent_num] = parent_name;grand_parent_num++;}}if(grand_parent_num != 0 && grand_child_num != 0 ){for(int m = 0;m<grand_child_num;m++){for(int n=0;n<grand_parent_num;n++){context.write(new Text(grand_child[m]), new Text(grand_parent[n]));//输出结果}}}/********** End **********/}}public static void main(String[] args) throws Exception{// TODO Auto-generated method stubConfiguration conf = new Configuration();Job job = Job.getInstance(conf,"Single table join");job.setJarByClass(simple_data_mining.class);job.setMapperClass(Map.class);job.setReducerClass(Reduce.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(Text.class);String inputPath = "/user/reduce/input"; //设置输入路径String outputPath = "/user/reduce/output"; //设置输出路径FileInputFormat.addInputPath(job, new Path(inputPath));FileOutputFormat.setOutputPath(job, new Path(outputPath));System.exit(job.waitForCompletion(true) ? 0 : 1);}}测评（资料素材和资料部分来自网络，供参考。

合集下载

educoder平台MapReduce基础实战

educoder平台MapReduce基础实战

MapReduce第1关：成绩统计过关代码：importimport class WordCount {/********** Begin **********/public static class TokenizerMapper extends Mapper<LongWritable, Text, Text, IntWritable> {private final static IntWritable one = new IntWritable(1);private Text word = new Text();private int maxValue = 0;public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {StringTokenizer itr = new StringTokenizer(),"\n");￥while ()) {String[] str = ().split(" ");String name = str[0];(str[1]));(name);(word,one);};import class Merge {/*** @param args—* 对A,B两个文件进行合并，并剔除其中重复的内容，得到一个新的输出文件C */ontext context)throws IOException, InterruptedException {String str = ();String[] data = (" ");Text t1= new Text(data[0]);Text t2 = new Text(data[1]);(t1,t2);}},/********** End **********/ontext context)throws IOException, InterruptedException {List<String> list = new ArrayList<>();for (Text text : values) {String str = ();if(!(str)){(str);}}(list);《for (String text : list) {(key, new Text(text));}}/********** End **********/}public static void main(String[] args) throws Exception{ Configuration conf = new Configuration();Job job = new Job(conf, "word count");;;$;;;;String inputPath = "/user/tmp/input/"; ;import class simple_data_mining {public static int time = 0;/*** @param args* 输入一个child-parent的表格* 输出一个体现grandchild-grandparent关系的表格:*/oString();int len = ();int i = 2;if(len == 0) continue;char relation_type = (0);String child_name = new String();String parent_name = new String();//获取value-list中value的childwhile(i) != '+'){child_name = child_name + (i);—i++;}i=i+1;//获取value-list中value的parentwhile(i<len){parent_name = parent_name+(i);i++;}//左表，取出child放入grand_childif(relation_type == '1'){grand_child[grand_child_num] = child_name; ~grand_child_num++;}else{//右表，取出parent放入grand_parentgrand_parent[grand_parent_num] = parent_name;grand_parent_num++;}}if(grand_parent_num != 0 && grand_child_num != 0 ){for(int m = 0;m<grand_child_num;m++){for(int n=0;n<grand_parent_num;n++){(new Text(grand_child[m]), new Text(grand_parent[n])); }//输出结果}}}/********** End **********/}}public static void main(String[] args) throws Exception{// TODO Auto-generated method stubConfiguration conf = new Configuration();—Job job = (conf,"Single table join");;;;;;String inputPath = "/user/reduce/input"; //设置输入路径 String outputPath = "/user/reduce/output"; //设置输出路径 (job, new Path(inputPath));(job, new Path(outputPath));(true) 0 : 1);}}测评。

实验5-MapReduce实验：单词计数

实验5-MapReduce实验：单词计数

实验五MapReduce实验：单词计数5.1 实验目的基于MapReduce思想，编写WordCount程序。

5.2 实验要求1．理解MapReduce编程思想；2．会编写MapReduce版本WordCount；3．会执行该程序；4．自行分析执行过程。

5.3 实验原理MapReduce是一种计算模型，简单的说就是将大批量的工作（数据）分解（MAP）执行，然后再将结果合并成最终结果（REDUCE）。

这样做的好处是可以在任务被分解后，可以通过大量机器进行并行计算，减少整个操作的时间。

适用范围：数据量大，但是数据种类小可以放入内存。

基本原理及要点：将数据交给不同的机器去处理，数据划分，结果归约。

理解MapReduce和Yarn：在新版Hadoop中，Yarn作为一个资源管理调度框架，是Hadoop下MapReduce程序运行的生存环境。

其实MapRuduce除了可以运行Yarn框架下，也可以运行在诸如Mesos，Corona之类的调度框架上，使用不同的调度框架，需要针对Hadoop做不同的适配。

一个完成的MapReduce程序在Yarn中执行过程如下：（1）ResourcManager JobClient向ResourcManager提交一个job。

（2）ResourcManager向Scheduler请求一个供MRAppMaster运行的container，然后启动它。

（3）MRAppMaster启动起来后向ResourcManager注册。

（4）ResourcManagerJobClient向ResourcManager获取到MRAppMaster相关的信息，然后直接与MRAppMaster进行通信。

（5）MRAppMaster算splits并为所有的map构造资源请求。

（6）MRAppMaster做一些必要的MR OutputCommitter的准备工作。

（7）MRAppMaster向RM(Scheduler)发起资源请求，得到一组供map/reduce task运行的container，然后与NodeManager一起对每一个container执行一些必要的任务，包括资源本地化等。

educoder平台MapReduce基础实战教案资料

educoder平台MapReduce基础实战教案资料

private final static IntWritable one = new IntWritable(1);
private Text word = new Text();
private int maxValue = 0;
public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
private IntWritable result = new IntWritable();
public void reduce(Text key, Iterable<IntWritable> values, Context context)
throws IOException, InterruptedException {
FileOutputFormat.setOutputPath(job, new Path(outputFile));
job.waitForCompletion(true);
/********** End **********/
}
}
命令行
touch file01
echo Hello World Bye World
import org.apache.hadoop.util.GenericOptionsParser;
public class WordCount {
/********** Begin **********/
public static class TokenizerMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
mapreduce基础编程

mapreduce基础编程

mapreduce基础编程MapReduce是一种用于大规模数据处理的编程模型和软件框架。

它可以将大数据集分成多个小数据集，并通过多个计算节点并行处理，最后汇总处理结果。

MapReduce将数据处理过程分成两个阶段：Map阶段和Reduce阶段。

在Map阶段中，数据被分成多个小数据集，每个小数据集上运行相同的计算任务，然后产生中间结果。

在Reduce阶段中，中间结果被合并，最终产生处理结果。

MapReduce的基础编程模型可以分为以下几个步骤：1. 输入数据的读取：输入数据可以来自于Hadoop Distributed File System (HDFS)、本地文件系统或其他数据源。

2. Map阶段的编写：开发者需要编写Map函数，该函数将输入数据切分成多个小数据集，并在每个小数据集上运行相同的计算任务，生成中间结果。

Map函数的输出通常是一个键值对（key-value pair），其中键表示中间结果的类型，值表示中间结果的值。

3. Reduce阶段的编写：开发者需要编写Reduce函数，该函数将中间结果根据键值进行合并，生成最终的处理结果。

Reduce函数的输出通常是一个键值对（key-value pair），其中键表示最终处理结果的类型，值表示最终处理结果的值。

4. 输出数据的写入：最终处理结果可以写入到HDFS或其他数据源中。

MapReduce程序的开发需要掌握Java或其他编程语言。

除了基础编程模型外，还需要了解MapReduce的一些高级编程技术，如Combiner、Partitioner、InputFormat、OutputFormat等。

通过这些技术，可以进一步提高MapReduce程序的性能和可扩展性。

总之，MapReduce是一种强大的大数据处理工具，掌握基础编程模型是进行大数据分析和处理的必要条件。