hadoop——hbase实验

2023-07-05

实验报告

报告标题：MR-Hbase-Hive

学号：20190423

姓名：金玉洲

日期：2022-12-11

一、实验环境

操作系统：

windows10

centos7

相关软件（含版本号）：

hadoop-2.6.0-cdh5.7.0

hbase-1.2.0-cdh5.7.0

hive-1.1.0-cdh5.7.0

mysql-connector-java-5.1.24-bin

zookeeper-3.4.5-cdh5.7.0

其它工具：

Vmware WorkStation 16 Pro

FinalShell 1.0.0.0

IntelliJ IDEA 2022.1

二、实验内容及其完成情况

（针对上述实验内容逐一详述实验过程）

（1） MapReduce 编程题 1：

上传气象数据代码:

public static void main( String[] args ) throws Exception {
    System.setProperty("HADOOP_USER_NAME","jyz");
    config=new Configuration();
    fileSystem=FileSystem.get(new URI(HDFS),config);
    put("D:\\NNU\\大三上\\hadoop\\weather.txt","/demo");
}
private static void put(String src,String dest) throws IOException{
    fileSystem.copyFromLocalFile(new Path(src),new Path(dest));
    System.out.println("ok");
}

MR程序代码：

WeatherMapper：

public class WeatherMapper extends Mapper<LongWritable, Text, Text,LongWritable> {
    @Override
    protected void map(LongWritable ke,Text value,Context context) throws IOException, InterruptedException {
        String []str=value.toString().split(",");

        try{
            context.write(new Text(str[0]),new LongWritable(Long.parseLong(str[5])));
        }
        catch (Exception e){

        }

    }
}

WeatherReducer：

public class WeatherReducer extends Reducer<Text, LongWritable, Text,DoubleWritable> {
    @Override
    protected void reduce(Text key,Iterable<LongWritable> values,Context context) throws IOException, InterruptedException {
        long sum=0;
        long cnt=0;
        for (LongWritable value:values){
            sum+=value.get();
            cnt++;
        }
        context.write(key,new DoubleWritable(sum*1.0/cnt));
    }
}

下载的代码：

public static void main( String[] args ) throws Exception {
    System.setProperty("HADOOP_USER_NAME","jyz");
    config=new Configuration();
    fileSystem=FileSystem.get(new URI(HDFS),config);
    get("/out4/part-r-00000","D:/hadoopfile/weatherresult.txt");
}
private static void get(String src, String dest) throws IOException {
    fileSystem.copyToLocalFile(false,new Path(src),new Path(dest),true);
    System.out.println("ok");
}

WeatherResultApp：

public class WeatherResultApp {
    static class Weather{
        public String date;
        public Double temp;
    }
    public static void main(String[] args) throws IOException {
        File file=new File("D:\\NNU\\大三上\\hadoop\\实验\\实验二\\weatherresult.txt");
        if(!file.exists()){
            System.out.println("file not found");
            return;
        }
        InputStream is=null;
        Reader reader=null;
        BufferedReader br=null;
        List<WeatherResultApp.Weather> weathers=new ArrayList<>();
        try {
            is=new FileInputStream(file);
            reader=new InputStreamReader(is);
            br=new BufferedReader(reader);
            String line=null;
            while((line=br.readLine())!=null){
                String []tmp=line.split("\t");
                WeatherResultApp.Weather weather=new WeatherResultApp.Weather();
                weather.date=tmp[0];
                weather.temp=Double.parseDouble(tmp[1]);
                weathers.add(weather);
            }
        }
        catch (Exception ex){
            System.out.println(ex);
        }
        finally {
            br.close();
        }
        Map<String, Double> map = new TreeMap<String, Double>(
                new Comparator<String>() {
                    public int compare(String obj1, String obj2) {
                        return obj1.compareTo(obj2);
                    }
                });
        Map<String,Integer> daycount=new HashMap<>();
        for(int i=0;i<weathers.size();i++){
            String month=weathers.get(i).date.substring(0,6);
            double temp=weathers.get(i).temp;
            if(map.get(month)!=null){
                temp+=map.get(month);//温度相加
            }
            map.put(month,temp);
            if(daycount.get(month)==null){
                daycount.put(month,1);
            }
            else {
                daycount.put(month,daycount.get(month)+1);//计算天数
            }
        }
        for (Map.Entry<String, Double> entry : map.entrySet()) {
            String mapKey = entry.getKey();
            Double mapValue = entry.getValue()/daycount.get(mapKey);
            System.out.println(mapKey + "平均温度：" + mapValue);
        }
    }
}

上传weather.txt文件

图表 1 上传weather.txt文件

数据文件上传成功

图表 2 数据文件上传成功

将jar上传到hadoop

图表 3 将jar上传到hadoop

虚拟机中运行得出结果

图表 4 虚拟机中运行得出结果

IDEA下载结果到本地

图表 5 IDEA下载结果到本地

查看Weatherresult.txt

图表 6 查看Weatherresult.txt

成功计算每月平均温度

图表 7 成功计算每月平均温度

（2） MapReduce 编程题 2：

上传四大名著代码:

public static void main( String[] args ) throws Exception {
    System.setProperty("HADOOP_USER_NAME","jyz");
    config=new Configuration();
    fileSystem=FileSystem.get(new URI(HDFS),config);
    writeAllFilesOfDir("D:/NNU/大三上/hadoop/实验/hadoop实验/book","/book/four-book"); //book四个文件整合
}
private static void writeAllFilesOfDir(String dir,String dest) throws IOException {
    File path=new File(dir);
    if(path.exists() && path.isDirectory()){
        File[] files=path.listFiles();
        System.out.println(files[0].getAbsolutePath()+":"+files[0].length());
        byte[] bts=new byte[4096];
        int size=0;
        FSDataOutputStream outputStream=fileSystem.create(new Path(dest));
        for(int i=0;i<files.length;i++){
            if(files[i].isFile()){
                FileInputStream fis=null;
                try{
                    fis=new FileInputStream(files[i]);
                    while((size=fis.read(bts))>0){
                        outputStream.write(bts,0,size);
                    }
                }
                catch (Exception ex){
                    System.out.println(ex);
                }
                finally {
                    {
                        try{
                            fis.close();
                        }
                        catch (Exception ex){
                            System.out.println(ex);
                        }

                    }
                }
            }
        }
    }
    else {
        System.out.println("未找到该文件夹");
    }
}

MR程序代码：

WordCountMapper：

public class WordCountMapper extends Mapper<LongWritable,Text, Text,LongWritable> {
    @Override
    protected void map(LongWritable ke,Text value,Context context) throws IOException, InterruptedException {
        LongWritable one=new LongWritable(1);
        JiebaSegmenter seg=new JiebaSegmenter();
        List<SegToken> tokenList=seg.process(value.toString(),JiebaSegmenter.SegMode.SEARCH);
        for (SegToken segToken:tokenList){
            if(segToken.word.matches("[\u4e00-\u9fa5]{2,}")){
                context.write(new Text(segToken.word),one);
            }
        }
    }
}

WordCountReducer：

public class WordCountReducer extends Reducer<Text, LongWritable, Text,LongWritable> {
    @Override
    protected void reduce(Text key,Iterable<LongWritable> values,Context context) throws IOException, InterruptedException {
        long sum=0;
        for (LongWritable value:values){
            sum+=1;
        }
        context.write(key,new LongWritable(sum));
    }
}

下载的代码：

public static void main( String[] args ) throws Exception {
    System.setProperty("HADOOP_USER_NAME","jyz");
    config=new Configuration();
    fileSystem=FileSystem.get(new URI(HDFS),config);
    get("/out5/part-r-00000","D:/hadoopfile/wordcountresult.txt");
}
private static void get(String src, String dest) throws IOException {
    fileSystem.copyToLocalFile(false,new Path(src),new Path(dest),true);
    System.out.println("ok");
}

WordCountResultApp：

public class WordCountResultApp {
    static class Word{
        public String key;
        public Integer count;
    }

    public static void main(String[] args) throws IOException {
        File file=new File("D:\\NNU\\大三上\\hadoop\\实验\\实验二\\wordcountresult.txt");
        if(!file.exists()){
            System.out.println("file not found");
            return;
        }
        InputStream is=null;
        Reader reader=null;
        BufferedReader br=null;
        List<Word> words=new ArrayList<>();
        try {
            is=new FileInputStream(file);
            reader=new InputStreamReader(is);
            br=new BufferedReader(reader);
            String line=null;
            while((line=br.readLine())!=null){
                String []tmp=line.split("\t");
                Word word=new Word();
                word.key=tmp[0];
                word.count=Integer.parseInt(tmp[1]);
                words.add(word);
            }
        }
        catch (Exception ex){
        }
        finally {
            br.close();
        }
        words.sort((a,b)->b.count-a.count);
        for(int i=0;i<20;i++){
            System.out.println(words.get(i).key+":"+words.get(i).count);
        }
    }
}

将四大名著合并起来上传

图表 8 将四大名著合并起来上传

中文分词只对两个词以上的进行

图表 9 中文分词只对两个词以上的进行

虚拟机中运行

图表 10 虚拟机中运行

IDEA中下载wordcountresult.txt

图表 11 IDEA中下载wordcountresult.txt

查看wordcountresult.txt

图表 12 查看wordcountresult.txt

IDEA中查看最终结果

图表 13 IDEA中查看最终结果

(3) Hbase 集群的安装配置（包括 Zookeeper 的分布式配置）

配置zookeeper环境变量

图表 14 配置zookeeper环境变量

配置临时文件夹路径

图表 15 配置临时文件夹路径

配置hbase的环境变量

图表 16 配置hbase的环境变量

配置hbase-env.sh

图表 17 配置hbase-env.sh

配置hbase-site.xml

图表 18 配置hbase-site.xml

配置分布式修改zoo.cfg

图表 19 配置分布式修改zoo.cfg

Zookeeper分布式启动成功

图表 20 Zookeeper分布式启动成功

分布式配置hbase-site.xml

图表 21 分布式配置hbase-site.xml

配置regionservers

图表 22 配置regionservers

将hbase拷贝给另外两台虚拟机

图表 23 将hbase拷贝给另外两台虚拟机

启动hdfs

图表 24 启动hdfs

启动成功

图表 25 启动成功

Start-hbase.sh启动成功

图表 26 Start-hbase.sh启动成功

Slave1的jps情况

图表 27 Slave1的jps情况

Slave2的jps情况

图表 28 Slave2的jps情况

60010端口有三台

图表 29 60010端口有三台

(4) Hbase 编程题

完整代码：

public class StudentApp {
    private static Configuration config;
    private static Connection connection;
    private static HBaseAdmin admin;

    public static void main(String[] args) throws Exception {
        config= HBaseConfiguration.create();
        config.set("hbase.zookeeper.quorum","master:2181,slave1:2181,slave2:2181");
        connection= ConnectionFactory.createConnection(config);
        admin=(HBaseAdmin) connection.getAdmin();
        //1.建student表
//        createTable("student","info","scores");
        //2.建course表
//        createTable("course","info");
        //3.添加学生数据
//        addStudentData();
        //4.添加课程数据
//        addCourseData();
        //5.添加成绩信息
//        addScores();
        //6.根据课程名计算该课程所有学生成绩总分和均分
        calByCourseName("Java");
    }

    private static String getCourseKey(String courseName) throws IOException {
        Table table=connection.getTable(TableName.valueOf("course"));
        Scan scan=new Scan();
        Filter filter=new ValueFilter(CompareFilter.CompareOp.EQUAL,
                new BinaryComparator(courseName.getBytes()));
//        scan.setFilter(filter);
        Filter filter2=new ColumnPrefixFilter("name".getBytes());
//        scan.setFilter(filter2);
        FilterList list=new FilterList();
        list.addFilter(filter);
        list.addFilter(filter2);
        scan.setFilter(list);
        ResultScanner result= table.getScanner(scan);
        Result rs=null;
        String courseKey="null";
        while((rs=result.next())!=null) {
            Cell[] cells = rs.rawCells();
            for (int i = 0; i < cells.length; i++) {
                Cell cell = cells[i];
                String rk = new String(CellUtil.cloneRow(cell));
                String cf = new String(CellUtil.cloneFamily(cell));
                String cl = new String(CellUtil.cloneQualifier(cell));
                String val = new String(CellUtil.cloneValue(cell));
//                System.out.printf("行键【%s】\t%s:%s\t%s\n", rk, cf, cl, val);
                courseKey=rk;
                break;
            }
        }
        table.close();
        return courseKey;
    }
    private static void calByCourseName(String courseName) throws IOException {
        String courseKey=getCourseKey(courseName);
        if(courseKey=="null"){
            System.out.println("课程不存在！");
            return;
        }
        Table table=connection.getTable(TableName.valueOf("student"));
        Scan scan=new Scan();
//        Filter filter=new FamilyFilter(CompareFilter.CompareOp.EQUAL,
//                new BinaryComparator(courseKey.getBytes()));
//        scan.setFilter(filter);
        Filter filter2=new ColumnPrefixFilter(("C"+courseKey).getBytes());
//        scan.setFilter(filter2);
        FilterList list=new FilterList();
//        list.addFilter(filter);
        list.addFilter(filter2);
        scan.setFilter(list);
        ResultScanner result= table.getScanner(scan);
        Result rs=null;
        double avg=0;
        int count=0;
        while((rs=result.next())!=null) {
            Cell[] cells = rs.rawCells();
            for (int i = 0; i < cells.length; i++) {
                Cell cell = cells[i];
                String rk = new String(CellUtil.cloneRow(cell));
                String cf = new String(CellUtil.cloneFamily(cell));
                String cl = new String(CellUtil.cloneQualifier(cell));
                String val = new String(CellUtil.cloneValue(cell));
                avg+=Double.parseDouble(val);
//                System.out.printf("行键【%s】\t%s:%s\t%s\n", rk, cf, cl, val);
            }
            count++;
        }
        System.out.println("课程："+courseName+"   总分："+avg+"  平均分："+avg/count);
        table.close();
    }
    private static void addScores() throws IOException {
        Table table=connection.getTable(TableName.valueOf("student"));
        Random random=new Random();
        for(int i=1;i<=100;i++) {
            String rowKey="00"+i;
            rowKey=rowKey.substring(rowKey.length()-3);
            Put put = null;
            int count=random.nextInt(3)+2;
            for(int j=1;j<=count;j++) {
                String col="C00"+j;
                int score=random.nextInt(50)+50;
                put = new Put(rowKey.getBytes());
                put.addColumn("scores".getBytes(), col.getBytes(),(score+"").getBytes() );
                table.put(put);
            }
        }
        table.close();
        System.out.println("数据添加成功！");
    }

    private static void addCourseData() throws IOException {
        Table table=connection.getTable(TableName.valueOf("course"));
        Random random=new Random();
        String []names={"Java","大数据","MySQL","数学"};
        String []teachers={"James","Mike",null,"张三"};
        String []periods={"64","48","24",null};

        for(int i=0;i<4;i++) {
            String rowKey="00"+(i+1);
            Put put = new Put(rowKey.getBytes());
            put.addColumn("info".getBytes(), "name".getBytes(), names[i].getBytes());
            table.put(put);
            if(teachers[i]!=null) {
                put = new Put(rowKey.getBytes());
                put.addColumn("info".getBytes(), "teacher".getBytes(), teachers[i].getBytes());
                table.put(put);
            }
            if(periods[i]!=null) {
                put = new Put(rowKey.getBytes());
                put.addColumn("info".getBytes(), "period".getBytes(), periods[i].getBytes());
                table.put(put);
            }
        }
        table.close();
        System.out.println("数据添加成功！");
    }

    private static void addStudentData() throws IOException {
        Table table=connection.getTable(TableName.valueOf("student"));
        Random random=new Random();
        for(int i=1;i<=100;i++) {
            String rowKey="00"+i;
            rowKey=rowKey.substring(rowKey.length()-3);
            Put put = new Put(rowKey.getBytes());
            put.addColumn("info".getBytes(), "name".getBytes(), ("stu"+rowKey).getBytes());
            table.put(put);
            put = new Put(rowKey.getBytes());
            int age=random.nextInt(3)+20;
            put.addColumn("info".getBytes(), "age".getBytes(), (""+age).getBytes());
            table.put(put);
        }
        table.close();
        System.out.println("数据添加成功！");
    }

    /**
     * 创建表
     * @param tableName         表名
     * @param columnFamilies    列族名（可以有多个）
     */
    private static void createTable(String tableName, String...columnFamilies) throws IOException {
        if(exists(tableName)){
            System.out.printf("表【%s】已经存在！\n",tableName);
            return;
        }
        if(columnFamilies.length<1){
            System.out.println("至少需要一个列族名！");
            return;
        }
        HTableDescriptor table=new HTableDescriptor(TableName.valueOf(tableName));
        HColumnDescriptor col=null;
        for (int i = 0; i < columnFamilies.length; i++) {
            col=new HColumnDescriptor(columnFamilies[i]);
            table.addFamily(col);
        }
        admin.createTable(table);
        System.out.printf("表【%s】创建成功，共有【%d】个列族，分别为：%s\n",
                tableName,columnFamilies.length, Arrays.toString(columnFamilies));


    }

    /**
     * 判断表是否存在
     * @param tableName 表名
     * @return
     */
    private static boolean exists(String tableName) throws IOException {
        return admin.tableExists(tableName);
    }
}