Hadoop界的Hello World!

2019-05-20  19:50:09

应用平台：Eclipse+ubantu+hadoop包

注：例分析的形式给宝宝们解释一下，详细运行过程省略。

实例：定义一个进行统计的原始文件

Hello MrZhangxd Hello Yootk

Hello Bye Bye Bye

Hello MrZhangxd

预期结果：

Bye 3

Hello 4

MrZhangxd 2

Yootk 1

主要实现利用MapReduce,那么什么是MapReduce?

MapReduce是一种可用于数据处理的编程模型。MapReduce程序本质是并行运行的。

对于MapReduce而言有两个阶段

Map阶段：对数据的处理阶段

Reduce阶段：对处理后的数据进行计算

以上实例如果使用MapReduce处理的话其流程如下：

Map处理

排序处理

合并处理

Reduce处理

以上整个操作称作一个完整的作业“Job”

实现单词统计的代码：

package org.mrzhangxd.com.linux;

import java.io.IOException;

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;

/**

* 本操作主要是进行Map的数据处理

* @author MrZhangxd

* 在Mapper父类里面接受的内容如下：

* Object:输入数据的具体内容

* Text:每行的文本数据

* Text:每个单词分解后的统计结果

* IntWritable:输出记录的结果

*/

public class WordCount {//本处要求实现单词统计的处理操作

//在整个代码中最为关键部分就是Map和Reduce部分，而且这两个部分是需要用户自己了实现的

private static class WordCountReducer extends Reducer{

@Override

protected void reduce(Text key, Iterable values,

Reducer.Context context)

throws IOException, InterruptedException {

// TODO Auto-generated method stub

super.reduce(key, values, context);

int sum = 0;//保存每个单词出现的数据

for(IntWritable count : values) {

sum += count.get();

}

context.write(key, new IntWritable(sum));

}

}

@SuppressWarnings("unused")

private static class WordCountMapper extends Mapper{

@Override

protected void map(Object key, Text value, Mapper.Context context)

throws IOException, InterruptedException {

// TODO Auto-generated method stub

super.map(key, value, context);

//默认情况下是提取每一行数据，所以每行数据里面都会存在空格，那么要按照空格进行分割，每当出现一个单词就需要做一个统计的1

String lineContent = value.toString();    //取出每行的数据

String result [] = lineContent.split(" ");//按空格进行数据拆分

for(int x = 0;x < result.length;x++) {    //循环每行每个单词而后进行数据的生成

//每一个单词最终生成的保存个数是1

context.write(new Text(result[x]), new IntWritable(1));

}

}

}

public static void main(String[] args) throws IOException {

// TODO Auto-generated method stub

if(args.length != 2) {

System.out.println("本程序需要两个参数，执行，hadoop yootk.jar /input/info.txt /output");

System.exit(1);

}

//每一次的执行实际上都属于一个作业（Job）,但是现在希望可以通过初始化参数来设置HDFS的文件存储路径

//假设现在的文件保存在HDFS上的“input/info.txt”上，而且最终输出结果也将保存在HDFS的“output”目录中

Configuration conf = new Configuration();

//考虑到最终要使用HDFS进行内容的处理操作，并且输入的时候不带有HDFS地址

String[] argArray = new GenericOptionsParser(conf, args).getRemainingArgs();//对输入的参数进行处理

//后面就需要作业进行处理了，而且Map与Reduces操作必须通过作业来配置

Job job = Job.getInstance(conf,"hadoop");//定义一个hadoop作业

job.setMapperClass(WordCountMapper.class);//设置执行的jar文件的程序类

job.setJarByClass(WordCount.class);    //指定Mapper的处理类

job.setMapOutputKeyClass(Text.class); //设置输出key的类型

job.setMapOutputValueClass(IntWritable.class);//设置输出的value类型

job.setReducerClass(WordCountReducer.class);//设置reduce操作的处理类

//设置Map-Reduce最终的执行结果

job.setOutputKeyClass(Text.class);//信息设置为文本

job.setOutputValueClass(IntWritable.class);//最终将内容设置为一个数值

//设置输入以及输出路径

//FileInputFormat.addInputPath(job, new Path(argArray[0]));

FileInputFormat.addInputPath(job, new Path(argArray[0]));

FileOutputFormat.setOutputPath(job,new Path(argArray[1]));

//等待执行完毕

try {

System.exit(job.waitForCompletion(true) ? 0 : 1);

} catch (ClassNotFoundException | InterruptedException e) {

// TODO Auto-generated catch block

e.printStackTrace();

}//执行完毕并且退出

}

}

注：代码的编写是需要使用到Hadoop中提供的*.jar文件的。

C:\Users\ XXX \Desktop\大数据\hadoop-3.2.0\share\hadoop

需要配置如下几个路劲的开发包：

|--Common组件包：

|   |--C:\Users\XXX\Desktop\大数据\hadoop-3.2.0\share\hadoop\common

|   |--C:\Users\XXX\Desktop\大数据\hadoop-3.2.0\share\hadoop\common\lib;

|--Mapreduce组件包：

|   |--C:\Users\XXX\Desktop\大数据\hadoop-3.2.0\share\hadoop\mapreduce

|   |--C:\Users\XXX\Desktop\大数据\hadoop-3.2.0\share\hadoop\mapreduce\lib;