Hadoop实战 一 WordCount

Hadoop实战 一 WordCount

Hadoop实战 一 WordCount

• 准备工作
• • 启动虚拟机Hadoop
  • Windows Hadoop 环境配置
  • 准备数据文本
  • 修改Maven pom文件
  • Hadoop配置文件
• 编码环节
• • Mapper文件
  • Reduce 文件
  • Mian方法
• 结果

准备工作

这里我们使用 IDEA + Maven 作为演示，做一个简单的单词计数统计

启动虚拟机Hadoop

如何启动以及配置问题见 – Hadoop入门

Windows Hadoop 环境配置

下载 winutils

链接：https://github.com/steveloughran/winutils

配置环境变量

在path里 添加 HADOOP_HOME

准备数据文本

一个非常简单的 txt 文本

修改Maven pom文件

这里我们添加两个依赖 hadoop-common 以及 hadoop-core

		
            org.apache.hadoop
            hadoop-common
            3.3.1
        
        
        
            org.apache.hadoop
            hadoop-mapreduce-client-core
            3.3.1
        
        
        
            org.apache.hadoop
            hadoop-hdfs
            3.3.1
            test
        

        
            org.apache.hadoop
            hadoop-mapreduce-client-common
            3.3.1
        

        
            org.apache.hadoop
            hadoop-mapreduce-client-jobclient
            3.3.1
            provided
        

Hadoop配置文件

将 Hadoop 配置文件 拷贝到Java 工程 resources文件夹下

编码环节 Mapper文件

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;



public class WcMapper extends Mapper {

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        // 结合文本内容 value 应该为： hello world responsibility hello combine google java php python java java java

        String[] words = value.toString().split(" ");
        for (String word : words) {
            // 将单词作为key 每个单词出现一次 1 作为value
            context.write(new Text(word), new Text("1"));
        }
    }
}

Reduce 文件

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;


public class WcReducer extends Reducer {

    @Override
    protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException {
        // 此时 Hadoop已经将Key相同的数据进行了合并 例： hello <1,1,1>   word <1>

        int count = 0; //总数
        for (Text value : values) {
            count+=Integer.parseInt(value.toString());
        }
        context.write(key, count);
    }
}

Mian方法

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

public class WordCount {
    public static void main(String[] args) throws Exception {

        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf,"wordcount");

        // 读取方式 一般为 TextInputFormat
        job.setInputFormatClass(TextInputFormat.class);
        //设置读取文件路径
        TextInputFormat.addInputPath(job,new Path("file:///C:\Users\Administrator\Desktop\Hadoop学习\words.txt"));
        //设置Map文件 以及 K2,V2 类型 对应Map文件里的K2,V2
        job.setMapperClass(WcMapper.class);
        job.setMapOutputKeyClass(Text.class); // K2
        job.setMapOutputValueClass(Text.class); // V2
        //设置Reducer文件 以及 K3,V3 类型 对应Reducer文件里的K3,V3
        job.setReducerClass(WcReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Integer.class);
        // 设置输出方式 一般为TextOutputFormat 
        job.setOutputFormatClass(TextOutputFormat.class);
        //设置输出路径
        TextOutputFormat.setOutputPath(job,new Path("file:///C:\Users\Administrator\Desktop\Hadoop学习\wordCount"));
        //等待任务结束
        boolean flag = job.waitForCompletion(true);
        System.exit(flag?0:1);
    }
}

结果

part-r-00000 为最后输出的数据

得到预期结果