https://hadoop.apache.org
集群 全分布式模式
当有了上面的依赖之后，已经有了一个本地模式的Hadoop，即便不下载Hadoop，也是可以独立运行Hadoop程序的
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;
public class WordCount {
/*继承了Mapper父类，实现了其中的map方法。这个类承担了MapReduce中Map的角色 * 至于Map这个角色到底做了什么事情呢？这是完全由map方法决定的。 * map方法一般都是把数据做成k-v的形式供reduce使用*/ public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> { /*Mapper的四个泛型参数： * 1. 第一个参数表示map函数输入（指从原始数据中的输入，此处就是file01和02）键的类型 * 2. 第二个参数表示map函数输入（指从原始数据中的输入，此处就是file01和02）值的类型 * 3. 第三个参数表示Map函数输出（输出到reduce阶段）键的类型 * 4. 第四个参数表示Map函数输出（输出到reduce阶段）值的类型*/ //Text当作是Java中的String类型，IntWritable当作Java中的Integer类型 private final static IntWritable one = new IntWritable(1); //Integer one = new Integer(1); private Text word = new Text(); //String word = new String(""); public void map(Object key, Text value, Context context) throws IOException, InterruptedException { System.out.println(key.toString()); /*map方法的三个参数： * 1. key：键，即等于四个范型中的第一个 * 2. value：值，即等于四个范型中的第二个 * 3. context：上下文，这里的上下文指map-reduce这个上下文*/ StringTokenizer itr = new StringTokenizer(value.toString());//此处的value就是文件内容 //StringTokenizer主要用于分割字符串，默认情况下，空格，\t \n 等都是分割符。当然也可以自己指定分割符。 while (itr.hasMoreTokens()) { word.set(itr.nextToken()); //word类似于String，set方法就类似于给字符串赋值，即word = "Hello"; context.write(word, one); //map阶段的结果写出给reduce阶段 } } /*实际上，Map的结果是类似于下面： Hello 1 World 1 Bye 1 World 1 Hello 1 Hadoop 1 Goodbye 1 Hadoop 1 */ } /*继承了Reducer父类，这个类承担了MapReducer中reduce的角色 * 至于reduce这个角色到底做了什么事情呢？这是完全由reduce方法决定的。*/ public static class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable> { /*Reducer的四个泛型参数： * 1. 第一个参数表示reduce函数输入（此处的输入是来自于map阶段的键）键的类型 * 2. 第二个参数表示reduce函数输入（此处的输入是来自于map阶段的值）值的类型 * 3. 第三个参数表示reduce函数输出（输出到最终结果）键的类型 * 4. 第四个参数表示reduce函数输出（输出到最终结果）值的类型*/ private IntWritable result = new IntWritable(); public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { /* 1. key指map阶段输出的key * 2. values指所有map阶段输出相同key的值的集合，整个的结果类似： * Text key Iterable<IntWritable> values * Hello {1, 1} * World {1, 1} * Bye {1} * Hadoop {1, 1} * Goodbye {1} * */ int sum = 0; for (IntWritable val : values) { sum += val.get(); } result.set(sum); context.write(key, result); } } //当程序运行时，首先执行Map阶段，之后再来执行Reduce阶段 public static void main(String[] args) throws Exception { //Configuratio 用来做配置的，默认配置 Configuration conf = new Configuration(); //job任务。在hadoop上跑的一个任务/程序 Job job = Job.getInstance(conf, "word count"); // 打jar包，打jar包主类是（xxxx.class） job.setJarByClass(WordCount.class); //Mapper这个类是（） job.setMapperClass(TokenizerMapper.class); //Combiner是一个优化 job.setCombinerClass(IntSumReducer.class); //Reducer这个类是谁（） job.setReducerClass(IntSumReducer.class); //设置一下输出键的类型 job.setOutputKeyClass(Text.class); //设置一下输出值的类型 job.setOutputValueClass(IntWritable.class); //文件输入格式：指定文件的输入 FileInputFormat.addInputPath(job, new Path(args[0])); //文件输出格式：指定文件的输出 FileOutputFormat.setOutputPath(job, new Path(args[1])); //退出， 阻塞方法：等待任务完成。 System.exit(job.waitForCompletion(true) ? 0 : 1); }}
