Map Reduce案例项目实现（c编译和eclipse）

文章目录

• ​​实现Map/Reduce的实例​​

• ​​1、编写代码​​
• ​​2、编译代码，运行程序​​

• ​​安装Eclipse开发工具​​

• ​​1、eclipse准备​​
• ​​2、安装hadoop插件​​
• ​​3、运行Eclipse​​
• ​​4、检查是否安装成功​​

• ​​Map/Reduce项目​​

• ​​1、运行eclipse并创建项目​​
• ​​2、创建类​​
• ​​3、编写代码​​
• ​​4、导出jar包​​
• ​​5、编写需统计文档​​
• ​​6 、HDFS文件操作​​
• ​​7、运行WordCount​​
• ​​8、查看输出目录“/output”​​

实现Map/Reduce的实例

Map/Reduce操作代表了一大类的数据处理操作方式，为了让大家对Map/Reduce的工作过程有一个直观的了解，下面的程序采用C语言实现了一个简单经典的Map/Reduce计算，计算从控制台输入的字符串中单词的计数。

1、编写代码

[root@node1 ~]# vim mapreduce.c

#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#define BUF_SIZE        2048
int my_map(char *buffer,char (*mapbuffer)[100]);
int my_reduce(char (*mapbuffer)[100],char (*reducebuffer)[100],int *count,int num);

int main(int argc, char *argv[]) {
     char buffer[BUF_SIZE];             //定义存储字符串的缓冲区
     char mapbuffer[BUF_SIZE][100];     //定义存储map结果的缓冲区
     char reducebuffer[BUF_SIZE][100];  //定义存储reduce结果的缓冲区
     int count[BUF_SIZE]={0};           //定义每个单词计数数组
     int num;                           //单词总数
     int i;
     int countnum;                      //归约后的结果数
     fgets(buffer, BUF_SIZE - 1, stdin);
     buffer[strlen(buffer)-1]='\0';     //将字符串最后的回车符改为结束符
     num=my_map(buffer,mapbuffer);      //调用map函数处理字符串
printf("This is map results:\n");
     for(i=0;i<num;i++) {
         printf("<%s\t1>\n",mapbuffer[i]);
     }
     countnum=my_reduce(mapbuffer,reducebuffer,count,num);
//调用reduce函数处理字符串
     printf("This is reduce results:\n");
          for(i=0;i<countnum;i++) {
                   printf("<%s\t%d>\n",reducebuffer[i],count[i]);
                        }
                        }
int my_map(char *buffer,char (*mapbuffer)[100]) {
    char *p;
    int num=0;
    if(p=strtok(buffer," ")) {
         strcpy(mapbuffer[num],p);
         num++;
    }
    else
       return num;
    while(p=strtok(NULL," ")) {
        strcpy(mapbuffer[num],p);
        num++;
    }
    return num;
}

int my_reduce(char (*mapbuffer)[100],char (*reducebuffer)[100],int *count,int num) {
    int i,j;
    int flag[BUF_SIZE]={0};
    char tmp[100];
    int countnum=0;
    for(i=0;i<num;i++) {
       if(flag[i]==0) {
           strcpy(tmp,mapbuffer[i]);
           flag[i]=1;
           strcpy(reducebuffer[countnum],mapbuffer[i]);
           count[countnum]=1;
           for(j=0;j<num;j++) {
if(memcmp(tmp,mapbuffer[j],strlen(tmp))==0&&(strlen(tmp)==strlen(mapbuffer[j]))&&(flag[j]==0))            {
                    count[countnum]++;
                    flag[j]=1;
               }
           }
           countnum++;
           }
        }
    return countnum;
}

2、编译代码，运行程序

随便输入语句，查看统计结果

安装Eclipse开发工具

Eclipse 是一个开放源代码的、基于Java的可扩展开发平台。就其本身而言，它只是一个框架和一组服务，用于通过插件组件构建开发环境。使用Eclipse工具便于我们开发hadoop应用软件。

1、eclipse准备

1）下载Eclipse软件包。

用winscp上传

2）解压到/usr/local目录下。

[root@node1 ~]# cd /usr/local
[root@node1 local]# tar xzvf /root/eclipse-java-luna-SR2-linux-gtk-x86_64.tar.gz

2、安装hadoop插件

解压出hadoop-eclipse-plugin-2.7.2.jar后，放在Eclipse下的plugin目录下。

[root@node1 ~]# cd /usr/local/eclipse/plugins/
[root@node1 plugins]# mv /root/hadoop-eclipse-plugin-2.7.2.jar .

3、运行Eclipse

（只能在图形界面操作，不能用xshell）

[root@node1 ~]# /usr/local/eclipse/eclipse &

4、检查是否安装成功

单击“OK”后，进入工作界面。

击菜单栏的“Window—>Open Perspective—>Other…”菜单项，显示“Open perspective”窗口。

点击“Other…”后，可以看到Map Reduce视图，该环境已安装成功。

Map/Reduce项目

WordCount实例统计hdfs系统中文件中出现的单词数目，这是一个典型的Map/Reduce项目，类似于C语言中的hello world。

1、运行eclipse并创建项目

选择菜单栏的“File—New—Other…”菜单项

选择“Map/Reduce Project”。

单击“Next”按钮后进入对话框

单击“Configure Hadoop install directory…”，打开对话框，输入Hadoop安装路径。

单击“OK”按钮，打开对话框输入项目名称WordCount。

单击“Next>”按钮，选择“Libraries”。

单击“Add External JARs…”

把目录“/home/hadoop/share/hadoop/common”、

“/home/hadoop/share/hadoop/mapreduce”、“/home/hadoop/share/hadoop/hdfs”、

“/home/hadoop/share/hadoop/common/lib/commons-cli”和

“/home/hadoop/share/hadoop/yarn”下的jar包添加到库中。

单击“Finish”按钮，完成WordCount项目的创建。

单击左边大象图标

在eclipse的项目管理器中显示wordcount项目，新建MapReduce项目WordCount创建成功。

2、创建类

右击列表中的项目WordCount，新建classs，输入类名WordCount。

3、编写代码

输入类名后，打开代码编辑窗口。

输入代码，代码如下所示：

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.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 {
    //继承mapper接口，设置map的输入类型为<Object,Text>  
    //输出类型为<Text,IntWritable>  
    public static class Map extends Mapper<Object,Text,Text,IntWritable>{  
        //one表示单词出现一次  
        private static IntWritable one = new IntWritable(1);  
        //word存储切下的单词  
        private Text word = new Text();  
        public void map(Object key,Text value,Context context) throws IOException,InterruptedException{  
            //对输入的行切词  
            StringTokenizer st = new StringTokenizer(value.toString());//以默认的分隔符将value切分  
            while(st.hasMoreTokens()){  
                word.set(st.nextToken());//切下的单词存入word  
                context.write(word, one); //将word为key，one为value写入磁盘 
            }  
        }  
    } 
//继承reducer接口，设置reduce的输入类型<Text,IntWritable>  
    //输出类型为<Text,IntWritable>  
    public static class Reduce extends Reducer<Text,IntWritable,Text,IntWritable>{  
        //result记录单词的频数  
        private static IntWritable result = new IntWritable();  
        public void reduce(Text key,Iterable<IntWritable> values,Context context) throws IOException,InterruptedException{  
            int sum = 0;  
            //对获取的<key,value-list>计算value的和  
            for(IntWritable val:values){  
                sum += val.get();  
            }  
            //将频数设置到result  
            result.set(sum);  
            //收集结果  
            context.write(key, result);  
        }  
    } 
    /** 
     * @param args 
     */  
    public static void main(String[] args) throws Exception{  
        // TODO Auto-generated method stub  
        Configuration conf = new Configuration();  
        //检查运行命令  
        String[] otherArgs = new GenericOptionsParser(conf,args).getRemainingArgs();  
        if(otherArgs.length != 2){  
            System.err.println("Usage WordCount <int> <out>");  
            System.exit(2);  
        }  
        //配置作业名  
        Job job = new Job(conf,"word count");  
        //配置作业各个类  
        job.setJarByClass(WordCount.class);  
        job.setMapperClass(Map.class);  
        job.setCombinerClass(Reduce.class);  
        job.setReducerClass(Reduce.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);  
    }  
}

4、导出jar包

在eclipse的项目管理器列表中右击项目名WordCount，选择export，选择导出为“JAR file”格式。

单击 “Next>”按钮，输入JAR包名称WordCount。

单击“Finish”按钮后，生成JAR包WordCount.jar，可以在/root/workspace/目录下查看所生成的JAR包。

5、编写需统计文档

在hadoop用户目录下新建wordcount目录

拷贝WordCount.jar文件到该目录下，在该目录下新建文件gdu1.txt和gdu2.txt，gdu1.txt文件内容如下：

gdu2.txt文件内容如下：

修改该目录下文件的属性。

6 、HDFS文件操作

切换到hadoop用户

先开启Hadoop服务，在hadoop系统上新建目录input。

将文件gdu1.txt和gdu2.txt文件上传至input目录下。

7、运行WordCount

其中的“/input”是输入目录，“/output”是输出目录。

8、查看输出目录“/output”

（output目录只能使用一次，若下次还要使用请用别名或者把这个目录删掉）。

“/output”目录生成两个新文件，其中part-r-00000文件保存WordCount的输出结果。

查看part-r-00000文件内容。