package org.example.charcount;


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

import java.io.IOException;

// 用于完成Map阶段
// 再MapReduce中，要求被处理的数据能够被序列化
// MApReduce提供了一套单独的序列化机制
// KEYIN-输入的键的类型。如果不指定，那么默认情况下，表示行的字节偏移量
// VALUEIN-输入值得类型。如果不指定，那么默认情况下，表示的读取到的一行数据
// KEYOUT-输出的键的类型。当前案例中，输出的键表示的是字符
// VALUEOUT-输出的值的类型。当前案例，输出的值表示的是次数
public class CharCountMapper extends Mapper<LongWritable,Text, Text,LongWritable> {
    private final LongWritable once = new LongWritable(1);
    // 覆盖map方法，将处理逻辑写到这个方法中
    // key：键。表示的是行的字节偏移量
    // value：值。表示读取到的一行数据
    // context：配置参数

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        // 将一行数据中的字符拆分出来
        char[] cs = value.toString().toCharArray();
        // 假设数据是hello，那么拆分出来的数组中包含的就是{'h','r','l','l','o'}
        // 可以写出h:1 e:1 l:2 o:1
        // 可以写出h:1 e:1 l:1 l:1 o:1
        for (char c:cs){
            context.write(new Text(c+""),once);
        }
    }
}

reducer代码：

package org.example.charcount;

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

import java.io.IOException;

// KEYIN、VALUEIN输入的键的类型。
// Reducer的数据从Mapper来的
// 所以Mapper的输出就是Reducer的输入
// KEYOUT、VALUEOUT-输出的值的类型。当前案例中，要输出每一个字符对应的总次数
public class CharCountReducer extends Reducer<Text, LongWritable,Text,LongWritable> {
    // 覆盖reduce方法，将计算逻辑写到这个方法中
    // key：键。当前案例中，键是字符
    // values：值。当前案例中，值是次数的集合对应的迭代器
    // context：配置参数

    @Override
    protected void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {
        // key='a'
        // value={1,1,1,1,1,1,1...}
        // 定义变量来记录总次数
        int sum=0;
        for (LongWritable value:values){
            sum+=value.get();
        }
        context.write(key,new LongWritable(sum));
    }
}

Driver代码：

package org.example.charcount;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

public class CharCountDriver {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        // 构建环境变量
        Configuration conf = new Configuration();
        // 构建任务
        Job job = Job.getInstance(conf);

        // 设置入口类
        job.setJarByClass(CharCountDriver.class);
        // 设置Mapper类
        job.setMapperClass(CharCountMapper.class);
        // 设置Reducer类
        job.setReducerClass(CharCountReducer.class);
        // 设置Mapper的输出类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(LongWritable.class);
        // 设置Reduce的输出类型
        job.setOutputKeyClass(Text.class);
        job.setOutputKeyClass(LongWritable.class);

        // 设置输入路径
        FileInputFormat.addInputPath(job, new Path("hdfs://hadoop01:9000/txt/characters.txt"));
        // 设置输出路径-要求输出路径必须不存在
        FileOutputFormat.setOutputPath(job,new Path("hdfs://hadoop01:9000/result/char_count"));
        // 提交任务
        job.waitForCompletion(true);
    }
}

现在运行我们会发现会报错：因为Hadoop对Windows系统的兼容性不强，所以在Windows中运行Hadoop程序的时候需要添加一些其他的配置

3、配置windows的hadoop

这些软件压缩包后面将放在我的百度云中，现在你们可以从网上下载，也可以私聊我要。

bin.7z解压的时候会遇到相同名字，我们选择全部替换。

配置之后，需要双击winutils.exe，如果出现一个黑色窗口一闪而过，那么没有任何问题；

如果双击winutils.exe之后报错，那么将msvcr120.dll文件放到C:\\Windows\\System32目录下，然后再次双击winutils.exe工具，查看是否报错

然后配置环境变量：

新建HADOOP_HOME

修改Path

新建HADOOP_USER_NAME

4、运行程序

①、如果运行程序的时候，出现了null/bin/winutils.exe错误，那么解决方案：

②、如果运行程序的时候，出现了NativeIO$Windows，那么说明Hadoop和Windows系统兼容性不够强，在运行程序的时候，检查出错，解决方案如下：

③、如果上述两种方案依然不能解决问题，那么需要在当前工程下建好对应的包，然后将jar目录下的NativaIO.java拷贝到这个包下

④、重新运行程序的时候你需要注意删除hdfs的result目录

成功后我们可以看到result文件下有char_count文件，下载到本地可以看到里面的结果：

注意这是hadoop的hdfs可视化界面，端口9870，hadoop3.X的端口，hadoop2.X是50070

浏览器输入hadoop01:9870

我们把part-00000 下载到本地查看结果：

三、练习与IDEA插件

1、练习-单词统计

①、我们再做一个练习-单词统计，在txt文件夹创建一个word文件写入数据

②、在IDEA编写代码，我们还在那个MapReduce的Maven工程内新建一个wordcount的包

下面新建三个类：

WordCountMapper WordCountReducer WordCountDriver

package org.example.wordcount;

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

import java.io.IOException;

public class WordCountMapper extends Mapper<LongWritable, Text,Text, IntWritable> {
   private final IntWritable once = new IntWritable(1);
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        //拆分单词
       String[] arr= value.toString().split(" ");
        //写出单词
        for (String s:arr){
            context.write(new Text(s),once);
        }

    }
}

package org.example.wordcount;

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

import java.io.IOException;

public class WordCountReducer extends Reducer<Text, IntWritable,Text,IntWritable> {

    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        int sum=0;
        for(IntWritable value :values){
            sum += value.get();
        }
        context.write(key,new IntWritable(sum));
    }
}

package org.example.wordcount;

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.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

public class WordCountDriver {

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);

        job.setJarByClass(WordCountDriver.class);
        job.setMapperClass(WordCountMapper.class);
        job.setReducerClass(WordCountReducer.class);

        //如果Mapper和Reducer的输出类型一致，那么可以只设置一次
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        FileInputFormat.addInputPath(job,new Path("hdfs://hadoop01:9000/txt/words.txt"));
        FileOutputFormat.setOutputPath(job,new Path("hdfs://hadoop01:9000/result/word_count"));
        job.waitForCompletion(true);
    }
}

③、结果如下：

2、练习-IP去重

直接上代码:

package org.example.ip;

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

import java.io.IOException;

public class IPMapper extends Mapper<LongWritable, Text,Text, NullWritable> {

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        context.write(value,NullWritable.get());
    }
}

package org.example.ip;

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

import java.io.IOException;

public class IPReducer extends Reducer<Text, NullWritable,Text,NullWritable> {

    @Override
    protected void reduce(Text key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
        context.write(key,NullWritable.get());

    }
}

package org.example.ip;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.example.wordcount.WordCountDriver;
import org.example.wordcount.WordCountMapper;
import org.example.wordcount.WordCountReducer;

import java.io.IOException;

public class IPDriver {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);

        job.setJarByClass(IPDriver.class);
        job.setMapperClass(IPMapper.class);
        job.setReducerClass(IPReducer.class);


        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(NullWritable.class);

        FileInputFormat.addInputPath(job,new Path("hdfs://hadoop01:9000/txt/ip.txt"));
        FileOutputFormat.setOutputPath(job,new Path("hdfs://hadoop01:9000/result/ip"));
        job.waitForCompletion(true);
    }
}

3、IDEA插件

查看结果更加方便：

安装后重启之后：

然后连接我们的hdfs：

这时候我们可以进行上传下载文件，也可以打开刚才的练习结果：

四、组件

1、Writable-序列化

①、在MapReduce中，要求被处理的数据能够被序列化。MapReduce提供了单独的序列化机制 - MapReduce底层的序列化机制是基于AVRO实现的。

为了方便操作，在AVRO的基础上，MapReduce提供了更简单的序列化形式 - 只需要让被序列化的对象对应的类实现Writable接口，覆盖其中的write和readFields方法

②、MapReduce针对常见类型提供了基本的序列化类

Java类	MapReduce的序列化类型
Byte	ByteWritable
Short	ShortWritable
Int	IntWritable
Long	LongWritable
Float	FloatWritable
Double	DoubleWritable
Boolean	BooleanWritable
String	Text
Array	ArrayWritable
Map	MapWritable

③、注意

④、代码片段

package org.example.serialflow;

import org.apache.hadoop.io.Writable;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

public class Flow implements Writable {
    private int upFlow;
    private int downFlow;
    public int getDownFlow() {
        return downFlow;
    }
    public void setDownFlow(int downFlow) {
        this.downFlow = downFlow;
    }
    public int getUpFlow() {
        return upFlow;
    }
    public void setUpFlow(int upFlow) {
        this.upFlow = upFlow;
    }
    // 需要将有必要的属性依次序列化写出即可
    @Override
    public void write(DataOutput out) throws IOException {
        out.writeInt(getUpFlow());
        out.writeInt(getDownFlow());
    }

    @Override
    public void readFields(DataInput in) throws IOException {
        setUpFlow(in.readInt());
        setDownFlow(in.readInt());
    }
}