目录

1 数据源(学生成绩.csv)

2 hadoop平台上传数据源

3 idea代码

3.1 工程框架

3.2 导入依赖

3.3 系统主入口(menu)

3.4 六个mapreduce

3.4.1 计算每门成绩的最高分、最低分、平均分(Mma)

3.4.2 计算每个学生的总分及平均成绩并进行排序(Sas)

3.4.3 统计所有学生的信息(Si)

3.4.4 统计每门课程中相同分数分布情况(Css)

3.4.5 统计各性别的人数及他们的姓名(Snn)

3.4.6 统计每门课程信息(Ci)

4 运行

5 改进

---

         本文只是用来分享代码，如果想要学习MapReduce如何去写的请转至下面的参考博客，该篇博客以“”统计每门课程中相同分数分布情况”为模板，从问题分析入手，一步步创建一个mapper、reducer和main(driver)从而组成一整个的MapReduce。

【手把手 脑把脑】教会你使用idea基于MapReduce的统计数据分析(从问题分析到代码编写)_扎哇太枣糕的博客-CSDN博客

不想跟着博客一步步操作的也可以选择直接下载项目文件，并在自己的idea上运行，数据源依旧是以下的学生成绩。

Hadoop-MapReduce项目代码ZIP压缩包+面向小白(注释详细清晰)-Hadoop文档类资源-CSDN下载

1 数据源(学生成绩.csv)

💥 旧坑勿踩：可以复制下面数据，粘贴到txt里把文件拓展格式改为csv，在上传至Hadoop平台之前一定要确保文件的编码方式为utf-8(否则中文会乱码)，具体操作为使用记事本打开学生成绩.csv文件，看右下角的编码方式，如果不是utf-8则可以将文件另存为时修改其编码方式。

💥一定一定一定不要为了元数据的好看就在第一行为数据加字段名，看是好看了，到时候运行不出来结果就很难受，不要问我怎么知道的，一个下午的血淋淋的教训。

英语,李沐,85,男,20
数学,李沐,54,男,20
音乐,李沐,54,男,20
体育,李沐,34,男,20
语文,李媛,81,女,20
音乐,李媛,85,女,20
体育,李媛,89,女,20
语文,马珂,75,女,19
英语,马珂,85,女,19
音乐,马珂,75,女,19
体育,马珂,65,女,19
语文,潘琴,42,女,20
英语,潘琴,48,女,20
音乐,潘琴,48,女,20
体育,潘琴,78,女,20
英语,秦灿,75,男,19
数学,秦灿,89,男,19
音乐,秦灿,85,男,19
体育,秦灿,99,男,19
语文,王靓,85,女,21
英语,王靓,85,女,21
数学,王靓,48,女,21
音乐,王靓,86,女,21
音乐,王靓,85,女,21
体育,王靓,96,女,21
体育,王靓,87,女,21
英语,吴起,85,男,20
数学,吴起,85,男,20
英语,张翔,96,男,20
数学,张翔,85,男,20
音乐,张翔,85,男,20
体育,张翔,87,男,20
语文,郑虎,85,男,20
数学,郑虎,85,男,20
音乐,郑虎,88,男,20
体育,郑虎,68,男,20
语文,周伟,76,男,19
英语,周伟,85,男,19
数学,周伟,76,男,19
音乐,周伟,99,男,19
体育,周伟,90,男,19
数学,朱鸿,90,男,21
音乐,朱鸿,80,男,21
体育,朱鸿,81,男,21

2 hadoop平台上传数据源

        Hadoop平台上传数据，其实也可以理解为向HDFS里存储数据，前提是Hadoop的集群必须搭建好，这里就默认大家都已经搭建完成并可以正常运行。这里可以如下图双击hadoop下的sbin目录下的start-all.cmd启动集群。

        集群启动成功后，在源数据的存储路径下打开DOS窗口，可以在该目录的文件路径框下输入cmd打开，或者直接在桌面打开DOS窗口再cd进源数据的存储路径。按照下图使用命令创建目录并将源数据(学生成绩.csv)上传至hadoop平台

3 idea代码

3.1 工程框架

新建一个maven工程，建立如下工程框架 ：

3.2 导入依赖

        MapReduce需要四个核心依赖，hadoop-client、hadoop-hdfs、hadoop-common、hadoop-mapreduce-client-core，依赖复制粘贴进自己的项目一定要记得刷新依赖，避免依赖还没导入成功就运行导致报错。

<dependencies>
	<dependency>
		<groupId>org.apache.hadoop</groupId>
		<artifactId>hadoop-client</artifactId>
		<version>2.7.3</version>
	</dependency>
	<dependency>
		<groupId>org.apache.hadoop</groupId>
		<artifactId>hadoop-hdfs</artifactId>
		<version>2.7.3</version>
	</dependency>
	<dependency>
		<groupId>org.apache.hadoop</groupId>
		<artifactId>hadoop-common</artifactId>
		<version>2.7.3</version>
	</dependency>
	<dependency>
		<groupId>org.apache.hadoop</groupId>
		<artifactId>hadoop-mapreduce-client-core</artifactId>
		<version>2.7.3</version>
	</dependency>
</dependencies>

3.3 系统主入口(menu)

//这里的导包是完成跨package调用其它包里的类
import couerse_info.CiMain;
import course_score_same.CssMain;
import max_min_avg.MmaMain;
import sex_number_name.SnnMain;
import student_info.SiMain;
import sum_avg_sort.SasMain;

import java.lang.reflect.Method;
import java.util.Scanner;

public class menu {
    public static void main(String[] args) {
        try {
            Scanner scanner = new Scanner(System.in);
            while(true){
                System.out.println("=========基于MapReduce的学生成绩分析=========");
                System.out.println("1、计算每门成绩的最高分、最低分、平均分");
                System.out.println("2、计算每个学生的总分及平均成绩并进行排序");
                System.out.println("3、统计所有学生的信息");
                System.out.println("4、统计每门课程中相同分数分布情况");
                System.out.println("5、统计各性别的人数及他们的姓名");
                System.out.println("6、统计每门课程信息");
                System.out.println("7、退出");
                System.out.print("请输入你想要选择的功能：");
                int option = scanner.nextInt();
                Method method = null;
                switch(option){
                    case 1:
                        method = MmaMain.class.getMethod("main", String[].class);
                        method.invoke(null, (Object) new String[] {});
                        break;
                    case 2:
                        method = SasMain.class.getMethod("main", String[].class);
                        method.invoke(null, (Object) new String[] {});
                        break;
                    case 3:
                        method = SiMain.class.getMethod("main", String[].class);
                        method.invoke(null, (Object) new String[] {});
                        break;
                    case 4:
                        method = CssMain.class.getMethod("main", String[].class);
                        method.invoke(null, (Object) new String[] {});
                        break;
                    case 5:
                        method = SnnMain.class.getMethod("main", String[].class);
                        method.invoke(null, (Object) new String[] {});
                        break;
                    case 6:
                        method = CiMain.class.getMethod("main", String[].class);
                        method.invoke(null, (Object) new String[] {});
                        break;
                    case 7:
                        System.exit(1);
                        break;
                    default:
                        System.out.println("输入正确的功能按键！！");
                        break;
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

3.4 六个mapreduce

3.4.1 计算每门成绩的最高分、最低分、平均分(Mma)

package max_min_avg;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;
/*
stu[0]:课程名称
stu[1]:学生姓名
stu[2]:成绩
stu[3]:性别
stu[4]:年龄
该功能实现的计算出每门课程中的最高分、最低分、平均分
 */

public class MmaMapper extends Mapper<LongWritable,Text,Text,Text> {
    @Override
    protected void map(LongWritable key1,Text value1,Context context)throws IOException,InterruptedException{
        //将文件的每一行传递过来，使用split分割后利用字符数组进行接收
        String[] splits = value1.toString().split(",");

        //向Reducer传递参数-> Key：课程 Value：成绩
        context.write(new Text(splits[0]),new Text(splits[2]));
    }
}

package max_min_avg;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

public class MmaReducer extends Reducer<Text, Text,Text, Text> {

    @Override
    protected void reduce(Text key,Iterable<Text> value,Context context)throws IOException,InterruptedException{
        //Arraylist集合储存所有的成绩数据，借用collections的方法求最大值最小值
        List<Integer> list = new ArrayList<>();
        for(Text v: value){
            list.add(Integer.valueOf(v.toString()));
        }
        //求max及min
        int maxScore = Collections.max(list);
        int minScore = Collections.min(list);
        // 求平均成绩
        int sum = 0;
        for(int score: list){
            sum += score;
        }
        double avg = sum  / list.size();
        System.out.println("*****************************************");
        String result = "的最高分:"+maxScore+"    最低分:"+minScore+"    平均分:"+avg;
        System.out.println(key.toString()+result);

        context.write(key,new Text(result));
    }
}

package max_min_avg;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
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 java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;

public class MmaMain {
        public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException, URISyntaxException {
            //创建job和“统计相同课程相同分数的人数”任务入口
            Configuration conf = new Configuration();
            Job job = Job.getInstance(conf);
            job.setJarByClass(MmaMain.class);

            //设置Mapper和Reducer的入口
            job.setMapperClass(MmaMapper.class);
            job.setReducerClass(MmaReducer.class);

            //设置Mapper的输入输出类型
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(Text.class);

            //设置Reducer的输入输出类型
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(Text.class);

            //指定输入输出路径
            String inputPath = "hdfs://localhost:9000/mapreduce/input/学生成绩.csv";
            String outputPath = "hdfs://localhost:9000/mapreduce/output/最大值最小值平均值.txt";
            FileInputFormat.setInputPaths(job,new Path(inputPath));
            FileOutputFormat.setOutputPath(job,new Path(outputPath));

            //输出路径存在的话就删除，不然就只能手动删除，否则会报该文件已存在的异常
            FileSystem fileSystem = FileSystem.get(new URI(outputPath), conf);
            if (fileSystem.exists(new Path(outputPath))) {
                fileSystem.delete(new Path(outputPath), true);
            }

            //执行job
            job.waitForCompletion(true);
        }
}

3.4.2 计算每个学生的总分及平均成绩并进行排序(Sas)

package sum_avg_sort;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

/*
stu[0]:课程名称
stu[1]:学生姓名
stu[2]:成绩
stu[3]:性别
stu[4]:年龄
该功能实现：统计每个学生总分平均分并对成绩进行排序
 */
public class SasMapper extends Mapper<LongWritable, Text,Text,Text> {
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        //将文件的每一行传递过来，使用split分割后利用字符数组进行接收
        String[] stu = value.toString().split(",");
        //向Reducer传递参数-> Key：学生姓名 Value：成绩
        context.write(new Text(stu[1]),new Text(stu[2]));
    }
}

package sum_avg_sort;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

public class SasReducer extends Reducer<Text,Text,Text,Text> {
    @Override
    protected void reduce(Text key, Iterable<Text> values,Context context) throws IOException, InterruptedException {
        System.out.println("*********************************************************************");
        //定义一个ArrayList集合接收该学生的各项成绩
        List<Integer> scores = new ArrayList<>();
        for(Text value:values){
            scores.add(Integer.valueOf(value.toString()));
        }
        //对该学生的成绩进行求总分、平均分
        int num = 0, sum = 0;
        for(Integer score:scores){
            sum = sum + score.intValue();
            num = num + 1;
        }
        float avg = sum / num;
        //成绩排序
        Collections.sort(scores);
        //使用一个字符串拼接排好序的所有成绩
        String sort = "的总分:"+sum+" 平均分:"+avg+" 该生的成绩从低到高排序是:";
        for(Integer score:scores){
            sort = sort + score + "  ";
        }
        System.out.println(key.toString()+sort);
        //输出
        context.write(key,new Text(sort));
    }
}

package sum_avg_sort;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
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 java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;

public class SasMain {
    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException, URISyntaxException {
        //创建一个job和任务的入口
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);
        job.setJarByClass(SasMain.class);

        //设置mapper和reducer的入口
        job.setMapperClass(SasMapper.class);
        job.setReducerClass(SasReducer.class);

        //设置mapper输出类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

        //设置reducer的输出类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        //指定输入输出路径
        String inputPath = "hdfs://localhost:9000/mapreduce/input/学生成绩.csv";
        String outputPath = "hdfs://localhost:9000/mapreduce/output/每个学生总分平均分排序.txt";
        FileInputFormat.setInputPaths(job,new Path(inputPath));
        FileOutputFormat.setOutputPath(job,new Path(outputPath));

        //输出路径存在的话就删除，不然就只能手动删除，否则会报该文件已存在的异常
        FileSystem fileSystem = FileSystem.get(new URI(outputPath), conf);
        if (fileSystem.exists(new Path(outputPath))) {
            fileSystem.delete(new Path(outputPath), true);
        }

        //执行job
        job.waitForCompletion(true);
    }
}

3.4.3 统计所有学生的信息(Si)

package student_info;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;
/*
stu[0]:课程名称
stu[1]:学生姓名
stu[2]:成绩
stu[3]:性别
stu[4]:年龄
该功能实现：统计所有学生课程考试信息
 */
public class SiMapper extends Mapper<LongWritable,Text,Text,Text> {
    @Override
    protected void map(LongWritable Key1, Text value1,Context context) throws IOException, InterruptedException {
        //将文件的每一行传递过来，使用split分割后利用字符数组进行接收
        String[] splits= value1.toString().split(",");
        //拼接姓名+性别+年龄
        String name = splits[1];
        String sex = splits[3];
        String age = splits[4];
        String stu_info = name+"-"+sex+"-"+age;
        //拼接课程+成绩
        String course = splits[0];
        String score = splits[2];
        String course_info = course+"-"+score;
        //向Reducer传递参数-> Key：姓名+性别+年龄  Value：课程+成绩
        context.write(new Text(stu_info),new Text(course_info));
    }
}

package student_info;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class SiReducer extends Reducer<Text, Text,Text, Text> {

    @Override
    protected void reduce(Text key,Iterable<Text> values,Context context)throws IOException,InterruptedException{
        //拼接学生各科考试成绩信息
        String scoreInfo = "";
        for(Text value:values){
            scoreInfo = scoreInfo + value+"   ";
        }
        System.out.println("********************************************************");
        System.out.println(key.toString()+"\n"+scoreInfo);
        context.write(key,new Text(scoreInfo));
    }
}

package student_info;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
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 java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;

public class SiMain {
    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException, URISyntaxException {
        //创建job和“统计相同课程相同分数的人数”任务入口
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);
        job.setJarByClass(SiMain.class);

        //设置Mapper和Reducer的入口
        job.setMapperClass(SiMapper.class);
        job.setReducerClass(SiReducer.class);

        //设置Mapper的输入输出类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

        //设置Reducer的输入输出类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        //指定输入输出路径
        String inputPath = "hdfs://localhost:9000/mapreduce/input/学生成绩.csv";
        String outputPath = "hdfs://localhost:9000/mapreduce/output/学生信息.txt";
        FileInputFormat.setInputPaths(job,new Path(inputPath));
        FileOutputFormat.setOutputPath(job,new Path(outputPath));

        //输出路径存在的话就删除，不然就只能手动删除，否则会报该文件已存在的异常
        FileSystem fileSystem = FileSystem.get(new URI(outputPath), conf);
        if (fileSystem.exists(new Path(outputPath))) {
            fileSystem.delete(new Path(outputPath), true);
        }

        //执行job
        job.waitForCompletion(true);
    }
}

3.4.4 统计每门课程中相同分数分布情况(Css)

package course_score_same;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

/*
stu[0]:课程名称
stu[1]:学生姓名
stu[2]:成绩
stu[3]:性别
stu[4]:年龄
该功能实现：统计该课程中成绩相同的学生姓名
 */
public class CssMapper extends Mapper<LongWritable, Text,Text,Text> {
    @Override
    protected void map(LongWritable key, Text value,Context context) throws IOException, InterruptedException {
        //将文件的每一行传递过来，使用split分割后利用字符数组进行接收
        String[] stu = value.toString().split(",");
        //拼接字符串：课程和成绩
        String sc = stu[0]+"\t"+stu[2];
        //向Reducer传递参数-> Key：课程+成绩 Value：学生名
        context.write(new Text(sc),new Text(stu[1]));
    }
}

package course_score_same;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

public class CssReducer extends Reducer <Text,Text,Text,Text>{
    @Override
    protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        //创建StringBuffer用来接收该课程中成绩相同的学生的姓名
        StringBuffer sb = new StringBuffer();
        //num变量用来计数
        int num = 0;
        //遍历values参数，将所有的value拼接进sb，并统计学生数量
        for(Text value:values){
            sb.append(value.toString()).append(",");
            num++;
        }
        //如果num=1，则表明该课程的这个成绩只有一个学生，否则就输出
        if(num>1){
            String names = "一共有" + num + "名学生,他们的名字是：" +sb.toString();
            System.out.println("*************************************************");
            System.out.println(key.toString() + names);
            context.write(key,new Text(names));
        }
    }
}

package course_score_same;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
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 java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;

public class CssMain {
    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException, URISyntaxException {
        //创建job和“统计相同课程相同分数的人数”任务入口
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);
        job.setJarByClass(CssMain.class);

        //设置Mapper和Reducer的入口
        job.setMapperClass(CssMapper.class);
        job.setReducerClass(CssReducer.class);

        //设置Mapper的输入输出类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

        //设置Reducer的输入输出类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        //指定输入输出路径
        String inputPath = "hdfs://localhost:9000/mapreduce/input/学生成绩.csv";
        String outputPath = "hdfs://localhost:9000/mapreduce/output/该课程中成绩相同的学生姓名.txt";
        FileInputFormat.setInputPaths(job,new Path(inputPath));
        FileOutputFormat.setOutputPath(job,new Path(outputPath));

        //输出路径存在的话就删除，不然就只能手动删除，否则会报该文件已存在的异常
        FileSystem fileSystem = FileSystem.get(new URI(outputPath), conf);
        if (fileSystem.exists(new Path(outputPath))) {
            fileSystem.delete(new Path(outputPath), true);
        }

        //执行job
        job.waitForCompletion(true);
    }
}

3.4.5 统计各性别的人数及他们的姓名(Snn)

package sex_number_name;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

/*
stu[0]:课程名称
stu[1]:学生姓名
stu[2]:成绩
stu[3]:性别
stu[4]:年龄
该功能实现：各性别人数及他们的姓名
 */
public class SnnMapper extends Mapper<LongWritable, Text,Text,Text> {
    @Override
    protected void map(LongWritable key, Text value,Context context) throws IOException, InterruptedException {
        //将文件的每一行传递过来，使用split分割后利用字符数组进行接收
        String[] stu = value.toString().split(",");
        //向Reducer传递参数-> Key：性别 Value：姓名
        context.write(new Text(stu[3]),new Text(stu[1]));
    }
}

package sex_number_name;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;

public class SnnReducer extends Reducer<Text,Text,Text,Text> {
    @Override
    protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        //创建集合来去除重复值(HashSet不允许重复值的存在，故可用来去重)
        List<String> names= new ArrayList<>();
        for (Text value:values){
            names.add(value.toString());
        }
        HashSet<String> singleNames = new HashSet(names);
        //创建StringBuffer用来接收同性别学生的姓名
        StringBuffer sb = new StringBuffer();
        //拼接学生姓名以及统计人数
        int num = 0;
        for(String singleName:singleNames){
            sb.append(singleName.toString()).append(",");
            num++;
        }
        //输出
        String result = "生一共有" + num + "名,他们的名字是：" +sb.toString();
        System.out.println("********************************************");
        System.out.println(key.toString() + result);
        context.write(key,new Text(result));
    }
}

package sex_number_name;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
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 java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;

public class SnnMain {
    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException, URISyntaxException {
        //创建job和“统计相同课程相同分数的人数”任务入口
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);
        job.setJarByClass(SnnMain.class);

        //设置Mapper和Reducer的入口
        job.setMapperClass(SnnMapper.class);
        job.setReducerClass(SnnReducer.class);

        //设置Mapper的输入输出类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

        //设置Reducer的输入输出类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        //指定输入输出路径
        String inputPath = "hdfs://localhost:9000/mapreduce/input/学生成绩.csv";
        String outputPath = "hdfs://localhost:9000/mapreduce/output/各性别人数及他们的姓名.txt";
        FileInputFormat.setInputPaths(job,new Path(inputPath));
        FileOutputFormat.setOutputPath(job,new Path(outputPath));

        //输出路径存在的话就删除，不然就只能手动删除，否则会报该文件已存在的异常
        FileSystem fileSystem = FileSystem.get(new URI(outputPath), conf);
        if (fileSystem.exists(new Path(outputPath))) {
            fileSystem.delete(new Path(outputPath), true);
        }

        //执行job
        job.waitForCompletion(true);
    }
}

3.4.6 统计每门课程信息(Ci)

package couerse_info;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;
/*
stu[0]:课程名称
stu[1]:学生姓名
stu[2]:成绩
stu[3]:性别
stu[4]:年龄
该功能实现的是:通过指定信息查找学生课程考试信息
 */

public class CiMapper extends Mapper<LongWritable,Text,Text,Text> {
    @Override
    protected void map(LongWritable Key1, Text value1,Context context) throws IOException, InterruptedException {
        //将文件的每一行传递过来，使用split分割后利用字符数组进行接收
        String[] splits= value1.toString().split(",");
        //拼接字符串：学生名和成绩
        String course = splits[0];
        String name = splits[1];
        String score = splits[2];
        String course_info = name + "：" + score;
        //向Reducer传递参数-> Key：课程 Value：学生名+成绩
        context.write(new Text(course),new Text(course_info));
    }
}

package couerse_info;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class ciReducer extends Reducer<Text, Text,Text, Text> {
    @Override
    protected void reduce(Text key,Iterable<Text> values,Context context)throws IOException,InterruptedException{
       //拼接课程的学生姓名和成绩
        String courseInfo = "\n";
        for(Text Info:values){
            courseInfo = courseInfo + Info + "   ";
        }
        System.out.println(key.toString()+"："+courseInfo);
        System.out.println("***********************************************************************************************************************");
        context.write(key,new Text(courseInfo));
    }
}

package couerse_info;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
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 java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;

public class CiMain {
    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException, URISyntaxException {
        //创建job和“统计相同课程相同分数的人数”任务入口
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);
        job.setJarByClass(CiMain.class);

        //设置Mapper和Reducer的入口
        job.setMapperClass(CiMapper.class);
        job.setReducerClass(ciReducer.class);

        //设置Mapper的输入输出类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

        //设置Reducer的输入输出类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        //指定输入输出路径
        String inputPath = "hdfs://localhost:9000/mapreduce/input/学生成绩.csv";
        String outputPath = "hdfs://localhost:9000/mapreduce/output/课程信息.txt";
        FileInputFormat.setInputPaths(job,new Path(inputPath));
        FileOutputFormat.setOutputPath(job,new Path(outputPath));

        //输出路径存在的话就删除，不然就只能手动删除，否则会报该文件已存在的异常
        FileSystem fileSystem = FileSystem.get(new URI(outputPath), conf);
        if (fileSystem.exists(new Path(outputPath))) {
            fileSystem.delete(new Path(outputPath), true);
        }

        //执行job
        job.waitForCompletion(true);
    }
}

4 运行

5 改进

        至此一个完整的基于mapreduce的学生成绩分析系统就算是基本完成了，当然完成的功能还是十分的基础。如果想要追求进阶操作，可以尝试使用多重处理，即把一个甚至多个mapreduce处理得到的结果当做是一个数据集，对该结果继续进行mapreduce分析。如果有意愿还可以再进一步分析，反正越分析越详细，这可能就是你课设比别人突出的部分，是一个大大的加分项。 

Hadoop大数据技术课程设计说明

《Hadoop大数据技术》课程设计任务书

一、设计时间及地点

1、时间：2021-2022年第一学期第15-16周。上午：8:00-11:30，下午：2:00-5:30。设计周的最后两天为验收时间，每个小组要求对课程设计任务提交设计报告。
2、地点：机房10#A301,机房10#A302,机房10#A303，以及安排的相关机房

二、设计目的和要求

（一）目的
本课程设计的目的是培养应用Hadoop大数据平台技术的相关工具以及思想解决实际问题的能力，掌握使用课程所学相关知识，提高调查研究、查阅技术资料以及编写技术文献的能力
（二）任务
在学好《Hadoop大数据技术》课程的基础上，搜集、研究和学习解决问题的相关知识，综合运用所学知识解决对应实际问题。

三、设计题目和要求

课程设计以小组方式进行，每个小组成员不超过3人，小组成员必须明确分工，保证组员的工作量符合课程考核要求。课程设计题目必须围绕Hadoop大数据平台相关技术选题，可以参考如下7类题目，但不局限于以下题目，可以采用自拟题目。
1 部署高可用的Hadoop平台
2 基于HDFS的应用开发
3 基于MapReduce的数据分析或应用开发
4 基于Hive的数据分析
5 基于Zookeeper的分布式协调服务
6 基于Flume的高可靠分布式日志采集系统
7 基于大数据的数据处理流程
参考选题：
1、部署高可用的Hadoop平台
功能描述：基于Ambari、CDM、TDM等工具完成多个节点的Hadoop平台搭建，并基于管理工具进行相关组件的运维与基本测试。
2、基于HDFS的应用开发
功能描述：搭建伪分布Hadoop集群，并基于HDFS实现相关应用开发。应用主题可以为实现网盘功能 ，包括上传，下载，删除等功能，也可自行拓展权限管理等功能。
3、基于MapReduce的数据分析
功能描述：搭建伪分布Hadoop集群，并基于MapReduce完成相关数据分析。可包含推荐系统、多维度的数据分析、PageRank算法等。
4、基于Hive的数据分析
功能描述：搭建伪分布Hadoop集群，并基于Hive完成相关数据而分析。功能点可包括内外部表的使用，分区分桶表的使用，多个维度的HQL分析等。
5、基于Zookeeper的分布式协调服务
功能描述： 搭建Zookeeper集群，并基于Zookeeper实现分布式锁或服务动态上下线的功能。
6、基于Flume的高可靠分布式日志采集系统
功能描述：基于Flume完成分布式的日志采集，并完成日志采集系统的高可靠，或是多路分流。
7、基于大数据的数据处理流程
功能描述：功能不限，结合大数据的数据处理流程，能正确的使用大数据的各个组件完成的大数据的数据处理分析过程。

四、设计成果的编制

（一）课程设计过程产生的相关代码、结果以及结果分析。
（二）编写课程设计报告，内容包括：
设计报告是对课程设计阶段所进行工作的总结，必须独立撰写一份课程设计报告，课程结束前需将报告打印好交指导老师评分。报告必须包括如下几个部分：
1．封面（见附一）
2．课题分析：对课题要解决的问题进行描述、开发语言与开发环境。
3．功能分析：对课题需要完成的功能模块进行分析。
4．实现技术：描述完成课题使用的方法，问题处理的具体实现过程。
5．设计实现：功能的具体实现，以及使用的工具的配置，脚本等。
6．结果与分析：展示设计实现功能、分析运行结果。
7．学习体会：包括设计、软件部署与编码调试过程中遇到的问题及解决办法；课程设计中的不足以及改进设想；设计中收获、体会等。
五、评分标准及成绩评定
1．平时纪律、预习及上机考核（占20%）
2．课程设计工作量、小组任务分工、系统完成情况（占50%）
3．设计报告（占30%）。
4．等级划分：优≥90分、良≥80分、中≥70分、及格≥60分和不及格＜60分。
5．与他人雷同或抄写复制他人程序及报告者，成绩按不及格处理。

六、设计指导教师及分组情况

在整个设计过程中，参与设计的学生上机安排在实验机房，机房为10#A302,机房10#A303，或安排的相关机房，未安排在机房上机的时间，同学们选择进入图书馆进行课程设计，要遵守纪律，按时考勤。
上机安排如下：
第15周 周一到周五的 下午6-9节

七、课程设计说明：

本课程设计分为规定动作+自主设计动作两个部分：
规定动作：完成LSN实验教学平台中《hadoop大数据技术》课程的实验–19级hadoop大数据技术课程设计案例中的前2个案例。
自主设计动作：可参考如下案例,每个主题限选2组
1、部署高可用的Hadoop平台
功能描述：基于Ambari、CDM、TDM等工具完成多个节点的Hadoop平台搭建，并基于管理工具进行相关组件的运维与基本测试。
可参考: https://www.bilibili.com/video/BV1mJ411s7vP

2、基于HDFS的应用开发
功能描述：搭建伪分布Hadoop集群，并基于HDFS实现相关应用开发。应用主题可以为实现网盘功能 ，包括上传，下载，删除等功能，也可自行拓展权限管理等功能。
可参考：
《hadoop大数据技术》课程的实验–19级hadoop大数据技术课程设计案例中的第3个案例
https://blog.csdn.net/weixin_41983824/article/details/84586194

3、基于MapReduce的数据分析（本主题依据数据集判断是否为同一主题）
功能描述：搭建伪分布Hadoop集群，并基于MapReduce完成相关数据分析。可包含推荐系统、多维度的数据分析、PageRank算法等。
可参考：
https://gitee.com/somefusion/LogAnalyzeHelper
https://www.cnblogs.com/cairsha/p/10033947.html
https://www.cnblogs.com/MoooJL/p/13583325.html
https://www.cnblogs.com/zimo-jing/p/8835667.html

4、基于Hive的数据分析
功能描述：搭建伪分布Hadoop集群，并基于Hive完成相关数据而分析。功能点可包括内外部表的使用，分区分桶表的使用，多个维度的HQL分析等。
参考：
https://gitee.com/master_empty/hive/tree/master
https://blog.csdn.net/ysy_1_2/article/details/106466263

5、基于Zookeeper的分布式协调服务
功能描述： 搭建Zookeeper集群，并基于Zookeeper实现分布式锁或服务动态上下线的功能。
参考：
https://blog.csdn.net/liyiming2017/category_8119571.html

6、基于Flume的高可靠分布式日志采集系统
功能描述：基于Flume完成分布式的日志采集，并完成日志采集系统的高可靠，或是多路分流。

7、基于大数据的数据处理流程
功能描述：功能不限，结合大数据的数据处理流程，能正确的使用大数据的各个组件完成的大数据的数据处理分析过程。

八、课程设计选题说明：

1.选题过程中，要注意主题的明确
可以修改为：
基于MapReduce实现图书馆数据的分析
基于MapReduce实现天气数据的分析
基于MapReduce实现天气推荐系统
等

2.工作量体现

每个人基于LSN完成规定动作
每天要记录当天完成的工作任务，体现在课程设计中
每位同学的工作量要饱满

问题集

1. mapreduce的环境怎么搭建，以及Pom文件怎么写

答：先安装好hadoop环境，并开启hdfs，yarn，通过jps确定5个服务都在
这时，就可以测试执行mapreduce程序了

# 基于hadoop的提供的MR程序，进行词频统计的案例
hadoop jar /app/hadoop/hadoop/share/hadoop/mapreduce/hadoop-mapreduce-examples-2.7.3.jar wordcount /demoinput /output
# /demoinput是输入文件的路径
# /output 是输出文件的路径

 # 基于hadoop的提供的MR程序，进行蒙特卡洛就π的案例
hadoop jar /app/hadoop/hadoop/share/hadoop/mapreduce/hadoop-mapreduce-examples-2.7.3.jar pi 5 5

经过上面的例子，我们会发现，执行MR程序，只需要编写一个MR代码，然后通过hadoop jar 去执行就好了。

那么怎么搭建Mapreduce工程呢，流程分为，创建maven工程，添加pom依赖，编写Mapper，Reducer，主方法类，然后打包后，上传到hadoop 集群，再通过hadoop jar XXX，即可执行

具体步骤较为琐碎，可以参考lsn中实验
实验3 分析和编写WordCount程序
http://172.16.16.164:8000/courses/10/assignments/62

总结

同学们有问题的话，可以留言，看到会及时回复

第一章 大数据概述

大数据的定义

大数据是指无法在一定时间内用常规软件工具对其 内容进行抓取、管理和处理的数据集合，比如：与人类社会活动有关的网络数据。

大数据产生方式的变革

• 运营式系统阶段
• 用户原创内容阶段
• 感知式系统阶段

大数据对科学研究的影响

• 第一种范式：实验科学
• 第二种范式：理论科学
• 第三种范式：计算科学
• 第四种范式：数据密集型科学

大数据对思维方式的影响

• 全样而非采样
• 效率而非准确
• 相关而非因果

大数据的概念（4V+xV）

volume：数据量大

variety：类型繁多

velocity：速度快，时效高

valueless：价值密度低

variability：可变性

veracity：真实性

数量与类型

• 结构化数据 structured data

  

• 半结构化数据 semi-structured data

  

• 非结构化数据 unstructured data

  

速度与价值

• 从数据的生成到消耗，时间窗口非常小，可用于生成决策的时间非常少

• 1秒定律：这一点也是和传统的数据挖掘技术有着本质的不同

• 价值密度低，商业价值高

  以视频为例，连续不间断监控过程中，可能有用的数据仅仅有一两秒，但是具有很高的商业价值

大数据计算模式

模式	解决问题	代表产品
批处理计算	针对大规模数据的批量处理	MapReduce、Spark等
流计算	针对流数据的实时计算	Storm、S4、Flume、Streams、Puma、DStream、Super Mario、银河流数据处理平台等
图计算	针对大规模图结构数据的处理	Pregel、GraphX、Giraph、PowerGraph、Hama、GoldenOrb等
查询分析计算	大规模数据的存储管理和查询分析	Dremel、Hive、Cassandra、Impala等

第二章 Linux使用与Hadoop介绍

Linux基础

• Linux泛指一类操作系统，具体的版本有： Ubuntu、CentOS、Debian、Red Hat、OpenSUSE、UOS(统信)、Deepin(深度)、优麒麟（Ubuntu Kylin）等

• Linux是一个多用户、多任务的操作系统。

• Linux支持带卓面(GNOME)的鼠标操作方式（GUI）,也支持不带卓面的命令行操作方式(CMD)。

系统目录结构

基本操作

目录文件：

• pwd ，查看当前路径

• cd 或 cd ~ 进入用户主目录， ~表示用户主目录

• mkdir mydir 在当前目录下创建子目录mydir

• rmr -R mydata 删除目录（R表示递归删除其下子目录）

• rm filename 删除文件

• touch filename 创建一个文件（内容为空）

压缩包：

• tar：解包（x）、压包（c）。

  z表示gzip格式

  tar -xzvf 待解压文件名 –C 目标路径

  tar -czvf 目标压缩文件名 文件或目录

查看文件：

• cat 由第一行开始显示文件内容

• tac 从最后一行开始显示， tac 是 cat 的倒著写！

• nl 显示的时候，输出行号！

• more 一页一页的显示文件内容

• less 与 more 类似，可以往前翻页！

• head 只看头几行

• tail 只看末尾几行

文本编辑器

vi/vim：

Linux与Windows的换行符：

• Linux/unix：\n
• Windows：\r\n

文件格式互转：

• unix2dos：将具有unix风格的格式文件转化为具有window下的格式文件。
• dos2unix：将具有windows风格的格式文件转化为unix下的格式文件。

Hadoop概述

Hadoop定义

Apache开源软件基金会开发的运行于大规模普通服务器上的大数据存储、计算、分析的分布式存储系统和分布式运算框架。

Hadoop特点

• 扩容能力（Scalable）：能可靠地（reliably）存储和处理千兆字节（PB）数据；
• 成本低（Economical）：可以通过普通机器组成的服务器群来分发以及处理数据。这些服务器群总计可达数千个节点；
• 高效率（Efficient）：通过分发数据，Hadoop 可以在数据所在的节点上并行地（parallel）处理它们，这使得处理非常的快速；
• 可靠性（Reliable）：Hadoop 能自动地维护数据的多份副本，并且在任务失败后能自动地重新部署（redeploy）计算任务。

Hadoop2.0组成

• 分布式文件系统：HDFS
• 资源分配系统：YARN
• 分布式运算框架：MapReduce

Hadoop与Google三大论文

Google	Apache
BigTable	HBase
MapReduce	MapReduce
GFS	HDFS

Hadoop集群的安装与使用

运行模式

• 单机模式（Standalone,独立或本地模式：安装简单，运行时只启动单个进程，仅调试用途；
• 伪分布模式（Pseudo-Distributed）：在单节点上同时启动 name node、data node、secondary name node、resource manager 、node manager 等5个进程，模拟分布式运行的各个节点；
• 完全分布式模式（Fully-Distributed）：正常的Hadoop集群，由多个各司其职的节点构成。

安装步骤

1. 配置主机名、关防火墙、编辑hosts文件；

2. 配置免密登录；

   Hadoop 运行过程中需要管理远端 Hadoop 守护进程，启动后，Name Node 是通过 SSH（Secure Shell）来无密码登录启动和停止各个 Data Node 上的各种守护进程的。同理，Data Node 上也能使用 SSH 无密码登录到 Name Node。

3. Hadoop解压安装；

4. 编辑Hadoop的配置文件，etc/hadoop下；（主要配置文件）

   Hadoop目录

   

   • 编辑 hadoop-env.sh、yarn-env.sh；（脚本）

   • 编辑 core-site.xml、hdfs-site.xml、mapred-site.xml、yarn-site.xml；（xml文件）

     core-site.xml主要内容：

     fs.default.name，机器名称的9000端口（hdfs://master:9000）

     hadoop.tmp.dir，集群数据目录（home/scau/hadoopdata）

     集群默认端口

     fs.defaultFS，9000；

     HDFS，Name Node，50070（或9870），dfs.namenode.http-address，http 服务的端口；

     YARN，Resource Manager，8088，yarn.resourcemanager.webapp.address，http 服务的端口。

   • 编辑 **slaves（workers）**文件。

     由主机列表组成，每台一行，用于说明数据节点

5. 复制hadoop工程文件夹到其他节点；

6. 格式化hdfs；

   Hdfs namenode -format

7. 启动hadoop。

   sbin/start-all.sh

   ​ 在 Hadoop Master 的终端执行 jps 命令，在打印结果中会看到 4 个进程，分别是 Resource Manager、Jps、 Name Node 和 Secondary Name Node，如果出现了这 4 个进程表示主节点进程启动；

   ​ 在 Hadoop Slave 的终端执行 jps 命令，在打印结果中会看到 3 个进程，分别是 Node Manager、Data Node 和 Jps，如果出现了这 3 个进程表示从节点进程启动。

集群使用基本命令

• 创建文件夹

  hdfs dfs -mkdir test
  

• 复制文件到hdfs

  hdfs dfs -put /resource /direct
  

• 显示文件（夹）统计信息

  hdfs dfs -ls test
  

• 显示文件内容，管道head（默认十行）

  hdfs dfs -cat file | head
  

• 复制文件

  hdfs dfs -cp /resource /direct
  

• 复制到本地

  hdfs dfs -get /resource
  

• 删除

  hdfs dfs -rm -r dir
  hdfs dfs -rm file
  

第三章 HDFS技术

HDFS定义

• HDFS是一个使用Java实现的、分布式的、可横向扩展的文件系统；
• 是Hadoop的核心组件；

设计目标

• 基于廉价的普通硬件，可以容忍硬件出错；

  • 系统中的某一台或几台服务器出现故障时，系统仍可用且数据完整。

• 大数据集（大文件）；

  • HDFS适合存储大量文件，总存储量可达到PB、EB级;
  • HDFS适合存储大文件，单个文件一般在百MB级之上;
  • 文件数目适中。

• 简单的一致性模型；

  • HDFS应用程序需要一次写入，多次读取一个文件的访问模式；
  • 支持追加(append)操作，但无法更改已写入数据。

• 顺序的数据流访问；

  • HDFS适合用于处理批量数据，而不适合用于随机定位访问。

• 侧重高吞吐量的数据访问，可容忍数据访问的高延迟；

• 为把“计算”移动到“数据”提供基础和便利。

优劣势、特点

适合做什么：

• 处理超大文件：存储并管理PB级数据；
• 流式地访问数据：处理非结构化数据；注重数据处理的吞吐量（latency不敏感）；
• 运行于廉价的商用机器集群上。

不适合做什么：

• 低延迟数据访问；

• 存储小文件；（不建议使用）

• 大量的随机读；（不建议使用）

• 需要对文件的任意修改。（不支持）

相关概念

block 块

传统块存储介质中，块是读写的最小数据单位 (扇区)。windows上文件系统块，被称为簇。

• HDFS使用了块的概念，默认大小128M/256M字节；
• HDFS将一个文件分为一个或数个块来存储。
  • 每个块是一个独立的存储单位；
  • 以块为单位在集群服务器上分配存储。

与传统文件系统不同的是，如果实际数据没有达到块大小，则并不实际占用整个块磁盘空间。

HDFS采用抽象的块概念的好处：

• 支持大规模文件存储：大文件被分拆成若干个文件块并被分发到不同的节点上，因此文件的大小不会受到单个节点的存储容量的限制，可充分利用集群中所有的磁盘；
• 简化系统设计：容易计算出一个节点可以存储多少文件块；方便管理元数据，元数据不需要和文件块一起存储，可以由其他系统负责管理元数据；
• 适合数据备份：每个文件块都可以冗余存储到多个节点上，提高了系统的容错性和可用性。

meta data 元数据

元数据内容包括：

• 文件系统目录树信息：
  • 文件名、目录名；
  • 文件和目录的从属关系；
  • 文件和目录的大小，创建及最后访问时间；
  • 权限。
• 文件和块对应的关系：
  • 文件由哪些块组成。
• 块的存放位置：
  • 机器名，块ID。

HDFS对元数据和实际数据采取分别存储的方法：元数据存储在 Name Node 服务器上；实际数据储存在集群的 Data Node 中。

name node 命名节点

• Name Node 是用来管理文件系统命名空间的组件；

• 一个HDFS集群只有一台活跃的 Name Node ；

• Name Node 上存放了HDFS的元数据；

  • 记录了每个文件中各个块所在的数据节点的位置信息；
  • 一个HDFS集群只有一份元数据，可能有单点故障问题。

• 元数据工作时在 Name Node 内存中,以便快速查询。

  集群关闭时，元数据（FsImage、EditLog）持久化到磁盘中，启动集群时，需要将元数据装载到内存中。

secondary name node 次命名节点

第二名称节点是 HDFS 架构中的一个组成部 分，协助 Name Node 完成 FsImage 和edits文件合并工作，使得内存中的 FsImage 保持“最新” 。Secondary Name Node 一般是单独运行在一台机器上。

data node 数据节点

• 块的实际数据存放在 Data Node 上；
• 每个块会在本地文件系统产生两个文件，一个是实际的数据文件，另一个是块的附加信息文件，其中包括块数据的长度、校验和，以及时间戳；
• Data Node 通过心跳包(Heartbeat)与 Name Node 通讯；
• 客户端读取/写入数据的时候直接与 Data Node 通信；
• 集群运行中可以安全加入和退出一些机器。

HDFS原理

体系结构

• 主从结构；
• 通信协议：TCP/IP；
• Name node 单点故障

Name Node 的作用

• Name Node 是一个中心服务器，单一节点，负责管理文件系统的命名空间 namespace，以及客户端对文件的访问；
• 文件操作，Name Node 负责文件元数据的操作，Data Node 负责处理文件内容的读写请求，数据流不经过 Name Node ，只会询问它跟哪一个 Data Node 联系；
• 副本在 Data Node 上的存放信息由 Name Node 来控制，根据全局情况做出块放置决定，读取文件时 Name Node 尽量让用户先读取最近的副本，降低带块消耗和读取时延；
• Name Node 全权管理数据块的复制，它周期性地从集群中的每一个 Data Node 接收心跳信号（Heart beat）和块状态报告（Block report）。接收到心跳信号意味着该 Data Node 节点工作正常。块状态报告包含了一个该 Data Node 上所有数据块的列表；
• Data Node 启动后向 Name Node 注册，通过后，周期性向Name Node 上报所有块信息；
• 心跳是每3秒一次，心跳返回结果带有 Name Node 给该 Data Node 的命令如复制块数据到另一台机器，或删除某个数据块。如果超过10分钟没有收到某个 Data Node 的心跳，则认为该节点不可用，并copy其上的block到其他 Data Node 。

数据结构

Name Node 负责管理分布式文件系统的 Namespace 命名空间，保存了两个核心数据结构：FsImage 和 EditLog 。

• FsImage 用于维护文件系统树以及文件树中所有的文件和文件夹的元数据；（快照式保存）
• 操作日志文件 EditLog 中记录了所有针对文件的创建、删除、重命名等操作。（命令追加式）

Name Node 的启动和运行

启动时

1. 通过 FsImage 读取元数据，载入内存；
2. 执行 EditLog 中的记录，在内存中生成最新的元数据；
3. 清空 EditLog ，保存最新的元数据到 FsImage ；
4. 收集 Data Node 汇报的块的位置信息。

​ 在名称节点启动的时候，它会将 FsImage 文件中的内容加载到内存中，之后再执行 EditLog 文件中的各项操作，使得内存中的元数据和实际的同步，存在内存中的元数据支持客户端的读操作。

​ 一旦在内存中成功建立文件系统元数据的映射，则创建一个新的 FsImage 文件和一个空的 EditLog 文件。

​ 名称节点起来之后，HDFS 中的更新操作会重新写到 EditLog 文件中，因为 FsImage 文件一般都很大（GB级别的很常见），如果所有的更新操作都往 FsImage 文件中添加，这样会导致系统运行的十分缓慢，但是，如果往 EditLog 文件里面写就不会这样，因为EditLog 要小很多。每次执行写操作之后，且在向客户端发送成功代码之前，EditLog 文件都需要同步更新。

EditLog 文件过大引发的问题：

​ 只有在 Name Node 重启时，edit logs才会合并到 FsImage 文件中，从而得到一个文件系统的最新快照。

​ 在名称节点运行期间，HDFS的所有更新操作都是直接写到 EditLog 中，久而久之， EditLog 文件将会变得很大。

​ 虽然这对名称节点运行时候是没有什么明显影响的，但是，当名称节点重启的时候，名称节点需要先将 FsImage 里面的所有内容映像到内存中，然后再一条一条地执行 EditLog 中的记录，当 EditLog 文件非常大的时候，会导致名称节点启动操作非常慢，而在这段时间内HDFS系统处于安全模式，一直无法对外提供写操作，影响了用户的使用。

​ 解决办法：Secondary Name Node第二名称节点。主要工作是阶段性的合并FsImage和edits文件，以此来控制 edits 的文件大小在合理的范围。

运行时

1. 对文件创建和写操作，记录到 EditLog ；
2. 更新内存中的元数据；
3. 收集 Data Node 汇报的块的创建和复制信息。

副本存放策略

• 第一个副本：放置在上传文件的 Data Node；如果是集群外提交，则随机挑选一台磁盘不太满，CPU不太忙的节点；

• 第二个副本：放置在于第一个副本不同的机架的节点上；

• 第三个副本：与第一个副本相同机架的其他节点；

• 更多副本：随机节点。

数据操作过程

创建文件

• client
  • 客户端请求 Name Node 在命名空间中建立新的文件元信息；
  • 如果不能创建文件则 Name Node 会返回失败，文件已存在 / 资源不足；
  • 如创建成功，客户端得到此文件的写保护锁。
• Name Node
  • Name Node 检查集群和文件状态；
  • 创建写保护锁来保证只有一个客户端在操作该文件；
  • 建立该文件的元信息；
  • 把创建文件这个事件加入 EditLog ；
  • 为该文件分配块，以及块的冗余备份位置。

写文件

• 客户端写一个文件并不是直接写到 HDFS 上；

• HDFS 客户端接收用户数据，并把内容缓存在本地；

• 当本地缓存收集足够一个HDFS块大小的时候，客户端同 Name Node 通讯注册一个新的块；

• 注册块成功后，Name Node 给客户端返回一个 Data Node 的列表；

  列表中是该块需要存放的位置，包括冗余备份。

• 客户端向列表中的第一个 Data Node 写入块；

  • 当完成时，第一个 Data Node 向列表中的下个 Data Node 发送写操作，并把数据已收到的确认信息给客户端，同时发送确认信息给 Name Node ；

  • 之后的 Data Node 重复之上的步骤；

  • 当列表中所有 Data Node 都接收到数据并且由最后一个 Data Node 校验数据正确性完成后，返回确认信息给客户端。

    

• 收到所有 Data Node 的确认信息后，客户端删除本地缓存；

• 客户端继续发送下一个块，重复以上步骤；

• 当所有数据发送完成后，写操作完成。

读文件

• 客户端与 Name Node 通讯获取文件的块位置信息，其中包括了块的所有冗余备份的位置信息： Data Node 的列表；
• 客户端获取文件位置信息后直接同有文件块的 Data Node 通讯，读取文件；
• 如果第一个 Data Node 无法连接，客户端将自动联系下一个 Data Node；
• 如果块数据的校验值出错，则客户端需要向 Name Node 报告，并自动联系下一个 Data Node 。

第四章 MapReduce 原理和 YARN

MapReduce 原理

文件从HDFS读取出InputFormat格式（定义数据文件如何分割和读取），分割成许多的InputSplit（定义输入到单个Map任务的输入数据分块），转换成RecordReader（为InputSplit定义了数据记录转换成一个<k1,v1>的方法），提供给Mapper类

主要两阶段：

• mapper 阶段：输入<k1,v2>，输出<k2,v2>

  • 从HDFS读取输入文件的分片，解析成key、value对（行号和行内容）；

  • 对输入文件的每一行<k1,v1>，解析成多个<k2,v2>，

    每个<k1,v1>键值对调用一次map函数；

  • （可选）对同一节点上的Map输出数据进行归约（combiner）；

    对同一个Map端的输出进行合并，即合并相同 key 的值，以便减少partitioner数据通信开销；conf.setCombinerClass(Reduce. Class)。

    可以不用指定，如果指定合并类，其性质类似于本地执行的一个 Reducer，满足一定的条件才能够执行。

  • 对输出的<k2,v2>进行分区（partitioner，如果不指定默认一分区）；

  • shuffle：对不同分区的数据，按照k2进行排序、分组。相同k2的v2放到集合v2’={v21,v22,…}中；

  • mapper 的结果写入本地磁盘。

• reducer 阶段：

  • 对多个map任务的输出，按照不同的分区，通过网络复制到不同的reduce节点（取map端的结果数据）；
  • Reducer把所有由Mapper传来的分区子集合并排序成一个分区；
  • 一个Reduce任务处理一个分区数据；
  • 对输入的<k2,v2’s>做用户自定义Reduce处理，转换成新的<k3,v3>输出；
  • 把reduce的输出保存到文件中。文件名为 part-r-0000N（对应分区）。

MR v1&v2

v1工作流程

• JobTracker一直等待JobClient提交Job；

  JobTracker失败是单点故障。

• TaskTracker每隔3秒向JobTracker发送心跳询问有没有任务可做（slaves主动向master拉取pull任务）；

  TaskTracker由于崩溃或运行阻塞而心跳断开，JobTracker视为失败，重新把这些任务分配到其他TaskTracker上面运行。

• 如果有派发Task（Map、Reduce）执行。

v2概念

Resource Container

• 集群节点将自身内存、CPU、磁盘等资源封装一起的抽象类。

• 调度器RM根据应用资源需求和集群各个节点的Resource Container进行调度。

Resource Manager

• 处理 Client 的请求；
• 启动监控 ApplicationMaster ；
• 监控 NodeManager ；
• 把应用管理和资源调度分开；
• 资源分配与调度。

Node Manager

• 单个节点上的资源管理；
• 处理来自 ResourceManager 的命令；
• 处理来自 ApplicationMaster 的命令；
• 当收到任务时，启动 Container 完成（map或reduce）任务。

Application Master

• 数据切分；
• 为应用程序申请资源，并分配给内部任务；
• 任务监控与容错。

YRAN工作流程

1. Client 向 Resource Manager （RM）提交作业；

2. RM 的App Manager 选择一台节点 Node Manager （NM），安排其启动 Application Master （AM）；

3. 被选中的NM节点启动AM；

   开始重复4~6（申请资源，执行任务并监控任务），直到作业结束；

4. AM向RM申请执行任务（map、reduce）所需的资源（Resource Container的节点列表）；

5. AM要求申请到的资源（RC）对应的NM启动任务；

6. NM执行任务，定期向AM汇报任务的进展；

7. 所有任务结束，作业也结束，释放所有RC资源，AM关闭。

第五章 ZooKeeper

提供通用的分布式锁服务，用以协调分布式应用。

数据模型

• 层次化的目录结构；

• 每个 znode 节点有一个唯一的路径标识，包含数据和子节点；

  节点有两种类型：

  • 短暂的（ephemeral）：客户端会话结束时，zookeeper会将该 短暂znode删除，短暂znode不可以有子节点；
  • 持久的（persistent）：不依赖于客户端会话，只有当客户端明确 要删除该持久znode时才会被删除。

• 数据包含多个版本；

• 节点不支持部分读写，而是一次性完整读写。

角色

leader

负责进行投票的发起和决议，响应所有对ZK的状态变更的请求，更新系统状态。

learner

写请求转发给leader

• follower：在选举过程中参与投票；
• observer： observer 不参加投票过程，只同步 leader 的状态，observer 的目的是为了扩展系统，提高读取速度。

client

请求发起方。

应用场景

统一命名服务

配置管理

集群管理

共享锁

队列管理

服务器数量

奇数（3、5、7）；

最大允许宕机个数，一半以下。

命令：

• zkServer.sh start/stop
• zkServer.sh status

第六章 Hive

Hive定义概念

Hive 是一种数据仓库技术，用于查询和管理存储在分布式环境下的大数据集。

• 构建于 Hadoop 的 HDFS 和 MapReduce 上，用于管理和查询分析结构化/非结构化数据的数据仓库。
  • 使用HQL（类SQL语句）作为查询接口；
  • 使用HDFS作为底层存储；
  • 使用MapReduce作为执行层，即将HQL语句转译成MR Job，然后在Hadoop执行。

元数据

Hive 的元数据存储（MetaStore）

• 表的名字、表的列和分区及其属性，表的属性包括是否为外部表等，表的数据所在目录等；

• 可以是内置的dearby数据库；也可以是通用关系型数据库（如MySQL）。

内部表与外部表

• 相同点：需要指定元数据，都支持分区；

• 不同点：实际数据的存储方式不同；

  • 内部表。实际数据存储在数据仓库目录（默认/user/hive/warehouse下）。删除表时，表中的数据和元数据将会被同时删除。

    内部表建表指令

    create table pokes (foo int, bar string) row format delimited fields terminated by '\t';
    

  • 外部表。需要 external 关键字；实际数据存储在创建语句 location 指定的HDFS 路径中，不会移动到数据库目录中。如果删除一个外部表，仅会删除元数据，表中的数据不会被删除。

    外部表建表指令

    create external table ext_pokes (foo int, bar string) row format delimited fields terminated by '\t' location '/data/extpokes';
    

分区

Hive表中的一个分区对应表下的一个目录（类似于数据库中相应分区列的一个索引），所有分区的数据都存储在对应的子目录中。

htable包含ds、city两个分区，则相同日期、不同 城市的hdfs目录分别为：

/datawarehouse/htable/ds=20100301/city=GZ
/datawarehouse/htable/ds=20100301/city=BJ

如果按日期分区，yyyymmdd 或者year/month/day

分桶

桶对指定列进行哈希(hash)计算时，根据哈希值切分数据，每个桶对应一个文件。

将属性列user分散到32个桶中，哈希值为0 、10的分别对应的文件为：

/datawarehouse/htable/ds=20100301/city=GZ/part-00000
/datawarehouse/htable/ds=20100301/city=GZ/part-00010

• 索引和分区最大的区别就是索引不分割数据表，分区分割数据表；
• 分区和分桶最大的区别就是分桶随机分割数据表，分区是非随机分割数据表。

HQL

建表

数据上传

第七章 HBase

定义概述

• HBase 是一个分布式的、面向列的开源数据库，该技术来源于 Google 论文“Bigtable：一个结构化数据的分布式存储系统”；
• 高可靠性（HDFS提供高可靠底层存储支持）、高性能（MapReduce提供高性能计算能力）、面向列、可伸缩、实时读写；

数据库类型

• HBase是一种分布式、列式数据库；

  介于NoSQL和RDBMS之间，仅能通过主键(row key)和主键的 range 来检索数据，仅支持单行事务（可通过 hive 支持来实现多表join等复杂操作）。

表的特点

• 大：一个表可以有上亿行，上百万列；
• 面向列:面向列(族)的存储和权限控制，列（族）独立检索；
• 稀疏:对于为空(null)的列，并不占用存储空间，因此，表可以设计的非常稀疏。

表的基本概念

• Row key ：行键（主键）；

  • 用来检索记录的主键，访问

• Qualifier：列；

• Column family：列簇；

  • 创建表时必须声明，一个列簇可以包含多个列；

  • 列簇中的列可以自由定义，列中的数据都是以二进制形式存在，没有数据类型；

  • 列名都以列簇为前缀。

    < Column family : Qualifier >

    courses : history

    courses : math

• Timestamp：时间戳；

• Cell：单元格：

  • 由**{row key, column(= < family > + < qualifier >), version}**唯一确定一个单元，无数据类型，全部是字节码形式。

  • 每个 cell 都保存着同一份数据的多个版本（按倒序排列，读取时最先找到的是最近的版本）。版本通过时间戳来索引。

表的物理存储

• Table在行的方向上分割为多个 HRegion ；

  HRegion Server 内部管理了一系列 HRegion 对象；每个HRegion对应了Table中的一个Region；

  HMaster负责Table的schema操作(增删改)、HRegion的拆分和迁移；

• 当某个分区数据不断增大并达到阈值时，等分为两个新的 region 。

• HRegion 是 HBase 中分布式和负载均衡的最小单元；

  

  一个HRegion可以不可拆分地重复分布在不同的HRegion server上。

  

  HRegion虽然是分布式存储的最小单元，但并不是存储的最小单元。

  HRegion由一个或者多个 Store 组成，每个store保存一个 columns family 。

  每个Store又由一个 memStore 和0至多个 StoreFile 组成。StoreFile以HFile格式保存在HDFS上。

Region 的快速定位

在 HRegion 的记录前后分别有 StartKey 和 EndKey ，Client 通过 HMaster 快速定位 RowKey 在哪个 HRegion 。

写读操作

• 向 Region Server 提交请求并找到对应 region ;

• 获取系统时间作为时间戳；

• 先将更新写 WAL log；

• 再将更新写入 MemStore ；

• 判断 Memstore 是否需要刷新到磁盘，为 StoreFile 文件。

• 依次扫瞄 BlockCache、MemStore、StoreFile，将结果合并即可(Merge Read)。

数据紧缩

Compaction，紧缩

​ MemStore 每次Flush会创建新的HFile，而过多的HFile会引起读的性能问题，HBase采用Compaction 机制来解决这个问题。

minor compact

​ 是指选取一些小的、相邻的 StoreFile 将他们合并成一个更大的 StoreFile ，在这个过程中不会处理已经 Deleted 或 Expired 的 Cell。

​ 一次 Minor Compaction 的结果是更少并且更大的 StoreFile。

major compact

​ Major Compaction 是指将所有的 StoreFile 合并成一个 StoreFile，在这个过程中，标记为 Deleted 的 Cell 会被删除，已经 Expired 的 Cell 会被丢弃，已经超过最多版本数的Cell 会被丢弃。一次 Major Compaction 的结果是一个 HStore 只有一个 StoreFile 存在。

​ Major Compaction 可以手动或自动触发，然而由于它会引起很多的 IO 操作而引起性能问题，因而它一般会被安排在集群比较闲的时间。

shell操作

create,put,get,scan,delete,drop,disable

a. 建表 htest,包含列族 info，data; 
1. create 'htest',''info','data'  

b. 查看表的定义； 
1. describe 'htest'  

c. 修改 data 的 VERSIONS 为 5； 
1. disable 'htest'  
2. alter 'htest',NAME=>'data',VERSIONS=>'5'  

d. 分别插入 rowkey 为 rk01、rk02、rk03 的记录，并分别输入它们的 info 列族下 name、age、 gender 三个列的值；并输入 data 列族下不同课程如 java、hadoop、csharp 的数值； 

1. put 'htest','rk01','info:name','lyx'  
2. put 'htest','rk02','info:name','csy'  
3. put 'htest','rk03','info:name','dsb'  
4.   
5. put 'htest','rk01','info:age','21'  
6. put 'htest','rk02','info:age','20'  
7. put 'htest','rk03','info:age','19'  
8.   
9. put 'htest','rk01','info:gender','male'  
10. put 'htest','rk02','info:gender','female'  
11. put 'htest','rk03','info:gender','male&female'  
12.   
13. put 'htest','rk01','data:java','90'  
14. put 'htest','rk02','data:java','100'  
15. put 'htest','rk03','data:java','80'  
16.   
17. put 'htest','rk01','data:hadoop','95'  
18. put 'htest','rk02','data:hadoop','95  
19. put 'htest','rk03','data:hadoop','97'  
20.   
21. put 'htest','rk01','data:csharp','88'  
22. put 'htest','rk02','data:csharp','93'  
23. put 'htest','rk03','data:csharp','91'  


e. 输入 rk01 的 hadoop 课程成绩 5 次以上；每次不同值； 
1. put 'htest','rk01','data:hadoop','96'  
2. put 'htest','rk01','data:hadoop','97'  
3. put 'htest','rk01','data:hadoop','98'  
4. put 'htest','rk01','data:hadoop','99'  
5. put 'htest','rk01','data:hadoop','100'  


f. 用 get 方法查看 rk01 的成绩； 
1. get 'htest','rk01','data'  

g. 用 get 方法查看 rk01 的所有版本的成绩； 
1. get 'htest','rk01',COLUMNS=>'data',VERSIONS=>5  

h. 用 get 方法查看 rk01 的 3 个版本的成绩； 
1. get 'htest','rk01',COLUMNS=>'data',VERSIONS=>3  


i. 用 scan 方法查询（所有或者单个 rowkey, 如 rk01）；
scan 'htest',{STARTROW=>'rk01',ENDROW=>'rk03'}
 
j. 删除某个列，如 java 列； 
1. delete 'htest','rk01','data:java'  
2. delete 'htest','rk02','data:java'  
3. delete 'htest','rk03','data:java'  

k. 统计行数； 
1. count 'htest'  


l. 使用表的别名简化输入：a=get_table ‘htest’,然后执行操作 a.scan 或者 a.get ‘101’等； 
1. a=get_table 'htest'  
2. a.scan  
3. a.get 'rk01'  


m. 表的删除； 
1. disable 'htest'  
2. drop 'htest'