mapreduce的自定义InputFormat的代码

浅浅的花香味﹌ 2022-12-15 12:56 145阅读 0赞

需求：Excel，一行一行的读取，传给Mapper

实现：

package application.mapreduce;
    
    import jxl.Cell;
    import jxl.Sheet;
    import jxl.Workbook;
    import lombok.SneakyThrows;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.NullWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.*;
    import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    import org.apache.hadoop.mapreduce.lib.input.FileSplit;
    
    import java.io.File;
    import java.io.IOException;
    import java.util.Arrays;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2020, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2020-10-17 4:57
     * @Version: 1.0
     * @Description:
     */
    public class ExcelFileInputFormat extends FileInputFormat<NullWritable, Text> {
        @Override
        protected boolean isSplitable(JobContext context, Path filename) {
            return false;
        }
    
        @Override
        public RecordReader<NullWritable, Text> createRecordReader(InputSplit inputSplit, TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
            ExcelRecordReader excelRecordReader = new ExcelRecordReader();
            excelRecordReader.initialize(inputSplit, taskAttemptContext);
            return excelRecordReader;
        }
    }
    
    
    class ExcelRecordReader extends RecordReader<NullWritable,Text>{
    
        private FileSplit fileSplit = null;
        private TaskAttemptContext context = null;
    
        private int rowIndex = 0;
        private int rowCount = 0;
    
        private Workbook workbook = null;
        private Sheet sheet = null;
    
        private Text text = new Text();
    
        @SneakyThrows
        @Override
        public void initialize(InputSplit inputSplit, TaskAttemptContext context) throws IOException, InterruptedException {
            this.fileSplit = (FileSplit) inputSplit;
            this.context = context;
    
            workbook = Workbook.getWorkbook(new File("D:/datas/bigdata.xls"));
            sheet = workbook.getSheet(0);
            rowCount = sheet.getRows();
        }
    
        @Override
        public boolean nextKeyValue() throws IOException, InterruptedException {
            Cell[] row = sheet.getRow(rowIndex);
            text.set(Arrays.asList(row).toString());
            if (rowIndex == rowCount - 1){
                rowIndex ++;
                return false;
            }
            else {
                rowIndex ++;
                return true;
            }
        }
    
        @Override
        public NullWritable getCurrentKey() throws IOException, InterruptedException {
            return NullWritable.get();
        }
    
        @Override
        public Text getCurrentValue() throws IOException, InterruptedException {
            return text;
        }
    
        @Override
        public float getProgress() throws IOException, InterruptedException {
            return 0;
        }
    
        @Override
        public void close() throws IOException {
            System.out.println("close");
            workbook.close();
        }
    }

package application.mapreduce;
    
    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.FileInputFormat;
    import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    import org.apache.hadoop.util.Tool;
    import org.apache.hadoop.util.ToolRunner;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2020, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2020-10-17 5:45
     * @Version: 1.0
     * @Description:
     */
    public class ExcelDriver extends Configured implements Tool {
        @Override
        public int run(String[] strings) throws Exception {
            Configuration configuration = new Configuration();
            Job job = Job.getInstance(configuration);
    
            job.setJarByClass(ExcelDriver.class);
    
            job.setMapperClass(ExcelMapper.class);
    
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(NullWritable.class);
    
            job.setNumReduceTasks(0);
    
            job.setInputFormatClass(ExcelFileInputFormat.class);
    
            FileInputFormat.setInputPaths(job,new Path(strings[0]));
            FileOutputFormat.setOutputPath(job,new Path(strings[1]));
    
            boolean res = job.waitForCompletion(true);
            return res ? 0 : 1;
        }
    
        public static void main(String[] args) throws Exception {
            ToolRunner.run(new ExcelDriver(),args);
        }
    }

## **其中，getProgress()函数可以不用写，即进度。** ##

## **close()函数会在nextKeyValue()返回false的时候进行调用，关闭所使用的流。** ##