package it.dawn.YARNPra.wc_hdfs;

import java.io.IOException;

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

/**
 * @author Dawn
 * @date 2019年5月1日23:09:08
 * @version 1.0
 * 
 * 思路？
 * wordcount单词计数（数据来源hdfs上）
 * <单词,1>
 * 
 * 数据传输
 * 
 * KEYIN:数据的起始偏移量0~10 11~20 21~30 
 * VALUEIN:数据
 * 
 * KEYOUT：mapper输出到reduce阶段 k的类型
 * VALUEOUT：mapper输出到reduce阶段v的类型
 * <hello,1><hunter,1><henshuai,1>
 */
public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable>{

	//key 起始偏移量 value 数据  context 上下文 
	@Override
	protected void map(LongWritable key, Text value, Context context)
			throws IOException, InterruptedException {
		//1.读取数据
		String line=value.toString();
		
		//2.切割 hello hunter
		String[] words=line.split(" ");
		
		//3.循环的写到下一个阶段<hello,1><hunter,1>
		for(String word: words) {
			//4.输出到reducer阶段
			context.write(new Text(word), new IntWritable(1));
		}
	}
	

}

package it.dawn.YARNPra.wc_hdfs;

import java.io.IOException;

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

/**
 * @author Dawn
 * @date 2019年5月1日23:15:33
 * @version 1.0
 * 
 * 汇总 <hello,4> <hunter,1> <henshuai,2>
 */
public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable>{

	@Override
	protected void reduce(Text k3, Iterable<IntWritable> v3,
			Context context) throws IOException, InterruptedException {
		//1.统计单词出现的次数
		int sum=0;
		
		//2 累加求和
		for(IntWritable v :v3) {
			//拿到累加值
			sum+=v.get();
		}
		
		//3 输出结果
		context.write(k3, new IntWritable(sum));
	}
	

}

package it.dawn.YARNPra.wc_hdfs;

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


import it.dawn.YARNPra.wc_hdfs.WordCountMapper;
import it.dawn.YARNPra.wc_hdfs.WordCountReducer;
/**
 * @author Dawn
 * @date 2019年5月2日14:28:27
 * @version 1.0
 * 输入和输出路径都是hdfs上的路径
 */
public class WordCountDriver_hdfs {
	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
		//1.获取job信息
		Configuration conf=new Configuration();
		Job job=Job.getInstance();
		
		//2.获取jar包
		job.setJarByClass(WordCountDriver_hdfs.class);
		
		//3.获取自定义的mapper与reducer类
		job.setMapperClass(WordCountMapper.class);
		job.setReducerClass(WordCountReducer.class);
		
		//4.设置map输出的数据类型
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(IntWritable.class);
		
		//5.设置reduce输出的数据类型（最终的数据类型）
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		
		//6.设置输入存在的路径与处理后的结果路径(注意包的导入 是org.apache.hadoop.mapreduce.lib.下的包)
		FileInputFormat.setInputPaths(job, new Path("/dawn/wordcount.txt"));
		FileOutputFormat.setOutputPath(job, new Path("/output/wc"));
		
		//7.提交任务
		boolean rs=job.waitForCompletion(true);
		System.out.println(rs?0:1);
		
	}
}