Apache Hadoop（HDFS/YARN/MR）

Apache HBase

Apache Hive

Apache Oozie

Apache Spark

Apache Sqoop

Apache Zookeeper

Apache Parquet

Apache Phoenix（*CDH中需要额外安装）

package com.shujia;

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/students.txt"));
        int index = 0;
        //字符缓冲输出流
        BufferedWriter bw = new BufferedWriter(new FileWriter("data/blocks2/block---" + index));

        //现在是假设一行数据是1m,没128m数据，就生成一个block块，不到128m也会生成一个block块
        //每次读到128*1.1约等于140行的数据，就写入128行，剩下的12行计入下一次block块中去存储
        //定义一个集合，用于存储，读取的内容
        ArrayList<String> row = new ArrayList<>();

        String line = null;

        //定义一个变量，记录读取的行数
        int offset = 0;
        //定义一个变量，记录读取的是哪一个block块
        int rowNum = 0;


        while ((line = br.readLine()) != null) {
            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/blocks2/block---" + index));
            }
        }

        //把剩余的数据写到一个block块中
        for(int i = row.size()-offset;i<row.size();i++){
            String s = row.get(i);
            bw.write(s);
            bw.newLine();
            bw.flush();
        }

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

package com.shujia;

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<String, Integer> 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/parts2/part---" + offset));
//遍历HashMap
Set<Map.Entry<String, Integer>> entries = map.entrySet();
for (Map.Entry<String, Integer> 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();
}

}
}

package com.shujia;

import java.io.File;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/*
        Map(通过线程池的方式，简单来说，模拟hadoop中一个block块生成一个map任务，一个map任务相当于一个线程)
        (将切分出来的bolck块中，统计每个班级的人数)
        4423毫秒
 */
public class Map {
    public static void main(String[] args) {

        long start = System.currentTimeMillis();
        //创建一个线程池
        ExecutorService executorService = Executors.newFixedThreadPool(10000);
        File file = new File("data/blocks2");
        //定义一个文件编号，从0开始
        int offset = 0;
        File[] files = file.listFiles();
        for (File f : files) {
            MapTask mapTask = new MapTask(f, offset);
            executorService.submit(mapTask);
            offset++;
        }

        //关闭线程池
        executorService.shutdown();
        long end = System.currentTimeMillis();
        System.out.println("总耗时："+(end - start)+"毫秒");
    }
}

package com.shujia;

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

//306毫秒 537毫秒 == 843毫秒
//4423毫秒reduce总耗时：117551毫秒
/*
    模拟hadoop的切分map-reduce处理的方式，总耗时14毫秒
        将每个map任务的结果，再做一次总的聚合，统计出最终的班级人数

    当数据量过大的时候，TB以上的级别的时候
        1、一台服务器不够存
        2、可能一台够存，但是纯java程序是由JVM虚拟机调起的，内存有限，可能会导致，OOM内存溢出

    这时候，就必须使用分布式存储，将大文件进行切分，先局部做运算，这时候局部的数据少很多，然后再总的聚合，数据量少且块！
 */
public class Reduce {
    public static void main(String[] args) throws Exception {
        long start = System.currentTimeMillis();
        //将past目录封装成File对象
        File file = new File("data/parts2");
        //获取下面的所有文件对象数组
        File[] files = file.listFiles();
        //创建一个map集合，接收总的结果数据
        HashMap<String, Integer> 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-big/part-r-000000"));
        //遍历集合
        Set<Map.Entry<String, Integer>> entries = map.entrySet();
        for (Map.Entry<String, Integer> keyValue:entries){
            String clazz = keyValue.getKey();
            Integer number = keyValue.getValue();
            bw.write(clazz+":"+number);
            bw.newLine();
            bw.flush();
        }

        //释放资源
        bw.close();
        long end = System.currentTimeMillis();
        System.out.println("reduce总耗时："+(end-start)+"毫秒");
    }
}