读取文件，统计每个班级的人数，并将结果写入到另一个文件中

1.java直接使用IO操作

import java.io.*;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;


public class Demo {
    public static void main(String[] args) throws Exception {
        //创建一个map集合，接收总的结果数据
        HashMap map = new HashMap<>();

        //读取文件，进行分割
        //创建缓冲字符输入流对象
        BufferedReader br = new BufferedReader(new FileReader("D://data/student.txt"));
        String line = null;
        while ((line = br.readLine()) != null) {
            //以冒号进行分割得到班级和人数
            String[] strings = line.split(",");
            String clazz = strings[4];
            //判断map集合中是否存在对应的key
            if (!map.containsKey(clazz)) {
                map.put(clazz, 1);
            } else {
                //如果存在，value值相加
                map.put(clazz, map.get(clazz) + 1);
            }
        }
        //关闭读取数据的通道
        br.close();

        //将结果写入到最终一个文件
        BufferedWriter bw = new BufferedWriter(new FileWriter("D://datas"));
        //遍历集合
        Set> entries = map.entrySet();
        for (Map.Entry keyValue : entries) {
            String clazz = keyValue.getKey();
            Integer number = keyValue.getValue();
            bw.write(clazz + ":" + number);
            bw.newLine();
            bw.flush();
        }

        //释放资源
        bw.close();
       
    }
}

 2.hadoop综合文件切分-仅仅将整个文件切分为block块

import java.io.*;
import java.util.ArrayList;

public class SplitFileBlock {
    public static void main(String[] args) throws Exception {
        //将数据读取进来
        //字符缓冲输入流
        BufferedReader br = new BufferedReader(new FileReader("data/bigstudents.txt"));
        //定义一个变量，记录读取的是哪一个block块
        int index = 0;
        //字符缓冲输出流
        BufferedWriter bw = new BufferedWriter(new FileWriter("data/blocks/block-" + index));
        //现在是假设一行数据是1m,没128m数据，就生成一个block块，不到128m也会生成一个block块
        //每次读到128*1.1约等于140行的数据，就写入128行，剩下的12行计入下一次block块中去存储
        //定义一个集合，用于存储，读取的内容
        ArrayList row = new ArrayList<>();
        String line = null;
        //定义一个变量，记录读取的行数
        int offset = 0;
        //偏移量
        int rowNum = 0;
        while ((line = br.readLine()) != null) {
            //row里面是全部数据
            row.add(line);
            offset++;
            //当我们的偏移量，128*1.1约等于140行的数据，就写入128行，剩下的12行计入下一次block块中去存储
            if (offset == 140) {
                //定位到该添加的数据
                rowNum = 128 * index;
                //循环128次，将集合存储的数据，写入到block块中
                for (int i = rowNum; i <= rowNum + 127; i++) {
                    String s = row.get(i);
                    bw.write(s);
                    bw.newLine();
                    bw.flush();
                }
                index++;
                //将offset设置为12
                offset = 12;
                bw = new BufferedWriter(new FileWriter("data/blocks/block-" + index));
            }
        }
        //把剩余的数据写到一个block块中
        for(int i = row.size()-offset;i 
3.hadoop切分-map切分-将每个block中的数据二次整合到新的part中
 
定义MapTask类
 
import java.io.*;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;

public class MapTask implements Runnable {
    private File file;
    public int offset;


    public MapTask(File file,int offset) {
        this.file = file;
        this.offset = offset;
    }

    @Override
    public void run() {
        //字符缓冲输入流
        try {
            BufferedReader br = new BufferedReader(new FileReader(file));
            //创建一个Map集合，使用HashMap
            HashMap map = new HashMap<>();
            String line = null;
            while ((line = br.readLine()) != null) {
                //用逗号进行切分
                String clazz = line.split(",")[4];
                //如果在map中没有该班级作为key，那我们就把这个班级作为key存放在集合，value设置为1
                if (!map.containsKey(clazz)) {
                    map.put(clazz, 1);
                } else {
                    //否则value值加1
                    map.put(clazz, map.get(clazz) + 1);
                }
            }
            //结束读取数据流程
            br.close();
            //将局部的map任务结果写入到文件中
            //创建字符缓冲输出流
            BufferedWriter bw = new BufferedWriter(new FileWriter("data/parts/part-" + offset));
            //遍历HashMap
            Set> entries = map.entrySet();
            for (Map.Entry keyValue : entries){
                String clazz = keyValue.getKey();
                Integer sumNumber = keyValue.getValue();
                //写入文件
                bw.write(clazz+":"+sumNumber);
                //换行
                bw.newLine();
                bw.flush();
            }
            //关闭写通道
            bw.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
 
定义Map类：
 
import java.io.File;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class Map {
    public static void main(String[] args) {
        //创建一个线程池
        ExecutorService executorService = Executors.newFixedThreadPool(10000);
        File file = new File("data/blocks");
        //定义一个part编号，从0开始
        int offset = 0;
        File[] files = file.listFiles();
        for (File f : files) {
            MapTask mapTask = new MapTask(f, offset);
            executorService.submit(mapTask);
            offset++;
        }
        //关闭线程池
        executorService.shutdown();
    }
}
 
4.定义reduce类
 
import java.io.*;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;

public class Reduce {
    public static void main(String[] args) throws Exception {
        //将past目录封装成File对象
        File file = new File("data/parts");
        //获取下面的所有文件对象数组
        File[] files = file.listFiles();
        //创建一个map集合，接收总的结果数据
        HashMap map = new HashMap<>();
        //遍历每个part文件
        for (File f : files) {
            //读取文件，进行分割
            //创建缓冲字符输入流对象
            BufferedReader br = new BufferedReader(new FileReader(f));
            String line = null;
            while ((line = br.readLine()) != null) {
                //以冒号进行分割得到班级和人数
                String[] strings = line.split(":");
                String clazz = strings[0];
                //包装类
                Integer sum = Integer.valueOf(strings[1]);
                //判断map集合中是否存在对应的key
                if (!map.containsKey(clazz)) {
                    map.put(clazz, sum);
                } else {
                    //如果存在，value值相加
                    map.put(clazz, map.get(clazz) + sum);
                }
            }
            //关闭读取数据的通道
            br.close();
        }
        //将结果写入到最终一个文件
        BufferedWriter bw = new BufferedWriter(new FileWriter("data/result/part"));
        //遍历集合
        Set> entries = map.entrySet();
        for (Map.Entry keyValue:entries){
            String clazz = keyValue.getKey();
            Integer number = keyValue.getValue();
            bw.write(clazz+":"+number);
            bw.newLine();
            bw.flush();
        }
        //释放资源
        bw.close();
    }
}