一個mapreduce同時加載讀取多個文件的代碼部分

方法一:

 

　　a.第一步:在job中加載兩個文件所在的位置

 

         FileInputFormat.setInputPaths(job, new Path[] { new Path("hdfs://192.168.9.13:8020/gradeMarking"),
　　　　　　new Path("hdfs://192.168.9.13:8020/implyCount") });

 

　　b.第二步:在Mapper類中重寫setup方法,使用context對象獲取該文件所在的文件名(如果是經過處理后的數據文件,因為文件名一樣part-r-00000,所以要獲取其所在的文件夾名)
        @Override
　　　　protected void setup(Mapper<Text, Text, Text, Text>.Context context) throws IOException, InterruptedException {
　　　　FileSplit fs = (FileSplit) context.getInputSplit();
　　　　parentName = fs.getPath().getParent().getName();
　　　　}

方法二:

　　a.第一步:在job中將文件加載到本地

　　　　job.addCacheFile(new URI("hdfs://192.168.9.13:8020/meanwhileFind(同現)_data/part-r-00000"));
　　b.第二步:在Mapper函數中重寫setup函數,用字符緩沖流進行讀取
　　　

@Override
        protected void setup(Mapper<LongWritable, Text, Text, Sort>.Context context)
                throws IOException, InterruptedException {
            @SuppressWarnings("resource")
            BufferedReader br = new BufferedReader(new FileReader("part-r-00000"));
            String str = null;
            while ((str = br.readLine()) != null) {
                String[] datas = str.split("\t");
                String[] sp = datas[0].split("-");
                if (!map.containsKey(sp[0])) {
                    HashMap<String, Double> mapInner = new HashMap<>();
                    mapInner.put(sp[1], Double.parseDouble(datas[1]));
                    map.put(sp[0], mapInner);
                } else {
                    @SuppressWarnings("rawtypes")
                    HashMap mapInner = map.get(sp[0]);
                    mapInner.put(sp[1], Double.parseDouble(datas[1]));
                }
            }
        }