标签：java hadoop --- apache job org 数据处理 import class

package org;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
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.File;
import java.io.IOException;
import java.net.URI;

public class WordCountApp {

    public static class mymapper extends Mapper
            <LongWritable,Text,Text,LongWritable>{
        //
        LongWritable one=new LongWritable(1);

        @Override
        protected void map(LongWritable key, Text value, Context context)
                throws IOException, InterruptedException {

            //先读出一行
            String lines=value.toString();
            //拆分成单词，放入字符串数组
            String [] words=lines.split(" ");
            //把字符数组中的每一个单词  --> word，1
            for (String word:words
                 ) {
                context.write(new Text(word),one);
            }
        }
    }

    public static class myreduce extends Reducer
            <Text,LongWritable,Text,LongWritable>{
        @Override
        protected void reduce(Text key, Iterable<LongWritable> values, Context context)throws IOException, InterruptedException {
            //统计count
            long sum=0;
            // 把values 中的值，累加
            for (LongWritable value:values
                 ) {
                sum+=value.get();
            }
            //把结果输出
            context.write(key,new LongWritable(sum));
        }
    }
    /**
     * 主方法
     * Driver,封装了MapReduce作业的所有信息
     */
    public static void main(String[] args)throws  IOException, ClassNotFoundException, InterruptedException {

        //创建配置项
        Configuration configuration = new Configuration();

        // hadoop jar 包  wordcount（类名） 数据输入地址args[0]  输出地址args[1]
        //准备工作，清理已经存在的输出目录
        Path outputpath = new Path(args[1]);
        //链接到HDFS
        //FileSystem.get(new URI("192.168.1.1"),configuration,"hadoop");
        FileSystem fileSystem = FileSystem.get(configuration);
        //0、如果 输出目录存在，--删除
        if (fileSystem.exists(outputpath)) {
            fileSystem.delete(outputpath, true);
            System.out.println("输出目录存在，但是已经被删除了！");
        }
        //1、创建一个job
        Job job = Job.getInstance(configuration, "wordcount");
        //2.设置处理哪个类  告诉job要处理的是哪个类
        job.setJarByClass(WordCountApp.class);
        //3.作业要处理的数据的路径
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        //4map处理的相关参数（固定套路） 1.找到自己处理的类  2。设置输出的key   3.设置输出value
        job.setMapperClass(mymapper.class);

        //固定
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(LongWritable.class);

        //5设置reduce先关参数
        job.setReducerClass(myreduce.class);
        //
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(LongWritable.class);
        //6.设置combiner  逻辑跟reduce一样
        job.setCombinerClass(myreduce.class);
        //7.作业处理完之后数据输出的路径
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        //8.程序执行后的处理
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }

}

标签：java,hadoop,---,apache,job,org,数据处理,import,class
来源： https://blog.csdn.net/weixin_46476341/article/details/113803789

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。