黑马程序员技术交流社区

标题: 学习经历 [打印本页]

---

作者: 小弹壳    时间: 2019-12-24 16:42
标题: 学习经历
不知觉一个月过去了，感觉没什么说的，来份我的收获，直接来份手写代码
    package cn.itcast.demo1;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
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.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class JobMain extends Configured implements Tool {
    @Override
    public int run(String[] args) throws Exception {
        //1.获取Job对象
        Job job  =Job.getInstance(super.getConf(),"E:\\input\\common_friends_step_out");

        //2.设置job任务
            //1.设置输出类和路径
             job.setInputFormatClass(TextInputFormat.class);
             TextInputFormat.addInputPath(job,new Path("file:///E:\\input\\common_friends_step_out"));

             //2.设置Mapper类和数据类型
             job.setMapperClass(MyOutputFormatMapper.class);
             job.setMapOutputKeyClass(Text.class);
             job.setMapOutputValueClass(NullWritable.class);


             //3 4 5 6

             //7.设置输出类和输出路径

             //8设置输出类和路径
             job.setOutputFormatClass(MyOutputFormat.class);
             TextOutputFormat.setOutputPath(job,new Path("file:///E:\\input\\common_friends_step2_out"));
        //3.等待Job任务结束
        boolean b1 = job.waitForCompletion(true);

        return b1? 0:1;
    }

    public static void main(String[] args) throws Exception {
        Configuration configuration = new Configuration();

        //启动Job任务
        int run = ToolRunner.run(configuration, new JobMain(), args);

        System.exit(run);
    }
}

package cn.itcast.demo1;


import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

public  class MyOutputFormat extends FileOutputFormat<Text, NullWritable> {


    @Override
    public RecordWriter<Text ,NullWritable> getRecordWriter(TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
        //1.获取目标文件输出流
        FileSystem fileSystem = FileSystem.get(taskAttemptContext.getConfiguration());
        FSDataOutputStream goodCommentOutputStream = fileSystem.create(new Path("file:///E:\\input\\good_comments\\good_comments.txt"));
        FSDataOutputStream badCommentOutputStream = fileSystem.create(new Path("file:///E:\\input\\bad_comments\\bad_comments.txt"));

        //2.
        MyRecordWriter myRecordWriter = new MyRecordWriter(goodCommentOutputStream, badCommentOutputStream);

        return myRecordWriter;
    }
}

package cn.itcast.demo1;

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 MyOutputFormatMapper 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 cn.itcast.demo1;

import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

import java.io.IOException;
import java.util.Spliterator;

public class MyRecordWriter extends RecordWriter<Text, NullWritable> {
    private FSDataOutputStream goodCommentsOutputStream;
    private FSDataOutputStream badCommentsOutputStream;

    public MyRecordWriter() {
    }

    public MyRecordWriter(FSDataOutputStream goodCommentsOutputStream, FSDataOutputStream badCommentsOutputStream) {
        this.goodCommentsOutputStream = goodCommentsOutputStream;
        this.badCommentsOutputStream = badCommentsOutputStream;
    }

    @Override
    public void write(Text text, NullWritable nullWritable) throws IOException, InterruptedException {
        //1.从文本数据中获得第9个字段
        String[] split = text.toString().split("\t");
        String numStr = split[9];
        //2:根据字段的值,判断评论的类型,然后将对应的数据写入不同的文件夹文件中
              if(Integer.parseInt(numStr) <= 1){
           //好评或者中评
             goodCommentsOutputStream.write(text.toString().getBytes());
              goodCommentsOutputStream.write("\r\n".getBytes());
           }else{
                //差评
                  badCommentsOutputStream.write(text.toString().getBytes());
                  badCommentsOutputStream.write("\r\n".getBytes());
       }
    }

    @Override
    public void close(TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
            IOUtils.closeStream(goodCommentsOutputStream);
            IOUtils.closeStream(badCommentsOutputStream);
    }
}

以上就是我的收获是不是很厉害，继续努力

---

欢迎光临 黑马程序员技术交流社区 (http://bbs.itheima.com/)

黑马程序员IT技术论坛 X3.2