第六次

172.22.0.3 master
172.22.0.4 slave1
172.22.0.5 slave2

<dependency>
  <groupId>org.apache.hbase</groupId>
  <artifactId>hbase-client</artifactId>
  <version>1.2.0</version>
</dependency>

hbase(main):002:0> create 'tab1','f1','f2'

package cn.edu.nnu;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.HBaseAdmin;

public class HbaseDemo {
    public static void main(String[] args) {
        Configuration conf = null;
        Connection conn = null;
        HBaseAdmin admin = null;
        try {
            conf = HBaseConfiguration.create();
            // 1. 对于第二个参数，如果在本机的 hosts 中已经配置了 172.22.0.3 master 等等项
            //    则可以直接用主机名即可，否则需要填写主机的 IP，例如 `172.22.0.3:2181`
            // 2. 如果在 `/hbase/conf/hbase-site.xml` 中 clientPort 写入了 2181 的端口号
            //    则按照下面来写即可，否则还需要在主机名后加上端口号，例如 `master:2181`
            conf.set("hbase.zookeeper.quorum", "master,slave1,slave2");
            conn = ConnectionFactory.createConnection(conf);
            admin = (HBaseAdmin) conn.getAdmin();
            if (admin.tableExists("tab1")) {
                System.out.println("tab1 exists!");
            } else {
                System.out.println("tab1 noexists!");
            }
            System.out.println("ok");
        } catch (Exception ex) {
            System.out.println("fail");
        } finally {
            try {
                assert conn != null;
                conn.close();
            } catch (Exception ex) {
                ex.printStackTrace();
            }
        }
    }
}

private static boolean existTable(String tableName) throws IOException {
    if (admin.tableExists(tableName)) {
        System.out.println(tableName + "已存在");
        return true;
    } else {
        System.out.println(tableName + "不存在");
        return false;
    }
}

private static void createTable(String tabName, String... cfs) throws IOException {
    if (existTable(tabName)) {
        return;
    }
    HTableDescriptor tab = new HTableDescriptor(
            TableName.valueOf(tabName));
    for (String cf : cfs) {
        HColumnDescriptor colDesc = new HColumnDescriptor(cf); // 创建列
        tab.addFamily(colDesc); // 添加列族
    }
    admin.createTable(tab); // 创建表
    System.out.println("Create Success!");
}
// main
createTable("tab2", "cf1", "cf2");

private static void putTest(String tabName, String rowKey, Map<String, List> data)
  throws IOException {
    if (!existTable(tabName)) {  // 判断表是否存在
        return;
    }
    Table tab = conn.getTable(TableName.valueOf(tabName)); // 获取表对象
    Put put = new Put(rowKey.getBytes(StandardCharsets.UTF_8));
    for (Map.Entry<String, List> entry : data.entrySet()) { // 逐个获取 data 中的数据
        String cf = entry.getKey();
        List val = entry.getValue();
        for (int i = 0; i < val.size(); i += 2) {
            put.addColumn(  // 添加列
                    cf.getBytes(StandardCharsets.UTF_8),  // 列名
                    ((String) val.get(i)).getBytes(StandardCharsets.UTF_8),  // 列项
                    ((String) val.get(i + 1)).getBytes(StandardCharsets.UTF_8)  // 数据
            );
        }
    }
    tab.put(put);  // 放入数据
}

public static void main(String[] args) {
    try {
        // ... 初始化
        Map<String, List> data = new HashMap<>(); // 创建 Map 对象
        data.put("cf1", Arrays.asList("name", "lc", "age", "20"));  // cf1 列族的项
        data.put("cf2", Arrays.asList("java", "100", "hadoop", "50"));  // cf2 列族的项
        putTest("tab2", "rk001", data);  // 静态成员方法添加数据
        System.out.println("ok");
    } catch (Exception ex) {
        System.out.println("fail");
    } finally {
        // ... 后续处理
    }
}

// main
getTest("tab2", "rk001", "cf1:name");
// get 输出 lc
private static void getTest(String tabName, String rowKey, String cfName)
  throws IOException {
    Table tab = conn.getTable(TableName.valueOf(tabName));
    Get get = new Get(rowKey.getBytes(StandardCharsets.UTF_8));
    Result result = tab.get(get);
    Cell[] cells = result.rawCells();
    for (int i = 0; i < cells.length; i++) {
        Cell cell = cells[i];
        String name = new String(CellUtil.cloneFamily(cell)) + ":"
      + new String(CellUtil.cloneQualifier(cell));
        if (name.equals(cfName)) { // 只读取一条
            String val = new String(CellUtil.cloneValue(cell));
            System.out.println(val);
            break;
        }
    }
}

// main
scanTest("tab2");
// scan
private static void scanTest(String tabName) throws IOException {
    Table tab = conn.getTable(TableName.valueOf(tabName));
    Scan scan = new Scan();
    ResultScanner scanner = tab.getScanner(scan); // 读取全部
    while (true) {
        Result result = scanner.next();
        if (result == null) {
            break;
        }
        Cell[] cells = result.rawCells();
        for (int i = 0; i < cells.length; i++) {
            Cell cell = cells[i];
            String rk = new String(CellUtil.cloneRow(cell));
            String name = new String(CellUtil.cloneFamily(cell)) + ":"
        + new String(CellUtil.cloneQualifier(cell));
            String val = new String(CellUtil.cloneValue(cell));
            System.out.println(rk + "\t" + name + "\t" + val);
        }
    }
}

// 输出
rk001  cf1:age    20
rk001  cf1:name  lc
rk001  cf2:hadoop  50
rk001  cf2:java  100

create 'music','info'

<dependency>
  <groupId>org.apache.hbase</groupId>
  <artifactId>hbase-server</artifactId>
  <version>1.2.0</version>
</dependency>
<dependency>
  <groupId>org.apache.hbase</groupId>
  <artifactId>hbase-hadoop-compat</artifactId>
  <version>1.2.0</version>
</dependency>

D:\ISO\HADOOP-DEMO
+---.idea
+---src
|   +---main
|   |   +---java
|   |       +---cn.edu.nnu
|   |       |   +---hbasemr
|   |       |       +---MovieNode.java    // 电影:评分:评论数
|   |       |       +---MovieRating.java  // 用于本机处理数据
|   |       |       +---MovieApp.java     // 将电影名存入 hbase
|   |       |       +---MovieMapper.java
|   |       |       +---MovieReducer.java
|   |       |       +---RatingApp.java    // 将评分统计入 hbase
|   |       |       +---RatingMapper.java
|   |       |       +---RatingReducer.java
|   |       +---demo
|   +---test
|       +---java
|           +---cn.edu.nnu
+---target

package cn.edu.nnu.hbasemr;

public class MovieNode {
    private final String movieName; // 电影名
    private double rate;            // 评分
    private long cnt;               // 评论数

    MovieNode(String movieName, Double rate, int cnt) {
        this.movieName = movieName;
        this.rate = rate;
        this.cnt = cnt;
    }

    MovieNode(String movieName) { this(movieName, 0., 0); }

    public void setCnt(long cnt) { this.cnt = cnt; }

    public void setRate(double rate) { this.rate = rate; }

    public double getRate() { return rate; }

    public long getCnt() { return cnt; }

    public String getMovieName() { return movieName; }
}

package cn.edu.nnu.hbasemr;

import jdk.nashorn.internal.runtime.regexp.joni.exception.ValueException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.BinaryComparator;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.Comparator;
import java.util.PriorityQueue;

public class MovieRating {
    static Configuration conf = null;
    static Connection conn = null;
    static HBaseAdmin admin = null;
    static PriorityQueue<MovieNode> queue = null;

    public static void main(String[] args) {
        try {
            init();  // 连接 hbase 初始化
            queue = new PriorityQueue<>(new Comparator<MovieNode>() {
                @Override
                public int compare(MovieNode o1, MovieNode o2) {
                    return Double.compare(o1.getRate(), o2.getRate());
                }
            });
            getValues("movies", "ratings");  // 将最终数据放到成员变量 queue 中
            printQueue();                    // 输出小根堆
        } catch (Exception ex) {
            ex.printStackTrace();
            System.out.println("fail");
        } finally {
            try {
                assert conn != null;
                conn.close();
            } catch (Exception ex) {
                ex.printStackTrace();
            }
        }
    }

    private static void getValues(String moviesTab, String ratingsTab) throws IOException {
        if (!existTable(moviesTab) || !existTable(ratingsTab)) {
            return;
        }
        Table mtab = conn.getTable(TableName.valueOf(moviesTab));
        Table rtab = conn.getTable(TableName.valueOf(ratingsTab));
        Scan mscan = new Scan();
        Scan rscan = new Scan();
        ResultScanner mscanner = mtab.getScanner(mscan);
        while (true) {
            Result mres = mscanner.next();  // 从 movies 表开始扫描
            if (mres == null) {
                break;
            }
            Cell[] cells = mres.rawCells();
            for (Cell cell : cells) {
                String movieNo = new String(CellUtil.cloneRow(cell));  // 电影编号
                String movieName = new String(CellUtil.cloneValue(cell)); // 电影名

                RowFilter rowFilter = new RowFilter( // 行过滤器（按电影编号）
                        CompareFilter.CompareOp.EQUAL,
            new BinaryComparator(Bytes.toBytes(movieNo)));
                rscan.setFilter(rowFilter);
                ResultScanner rscanner = rtab.getScanner(rscan); // 从 ratings 表读取
                MovieNode movieNode = new MovieNode(movieName); // 创建 MovieNode 对象

                while (true) {
                    Result rres = rscanner.next(); // 读取一个数据
                    if (rres == null) {
                        break;
                    }
                    Cell[] cells2 = rres.rawCells();
                    for (Cell cell2 : cells2) {
                        String rk = new String(CellUtil.cloneRow(cell2));
                        String qualifier = new String(CellUtil.cloneQualifier(cell2));
                        String val = new String(CellUtil.cloneValue(cell2));
                        if (qualifier.equalsIgnoreCase("avg")) {
                            try {  // 将字符串转为 double
                                movieNode.setRate(Double.parseDouble(val));
                            } catch (ValueException ve) {
                                ve.printStackTrace();
                            }
                        } else if (qualifier.equalsIgnoreCase("cnt")) {
                            try {  // 将字符串转为 int
                                movieNode.setCnt(Long.parseLong(val));
                            } catch (ValueException ve) {
                                ve.printStackTrace();
                            }
                        }
                    }
                }
                if (movieNode.getCnt() >= 20) {  // 如果评论数大于等于 20
                    queue.add(movieNode);  // 加入队列
                }
                if (queue.size() > 10) {  // 队列成员个数超过 10 个
                    queue.poll();  // 将评分最小的节点出队
                }
            }
        }
    }

    private static void printQueue() {
        for (MovieNode mn : queue) {
            System.out.println(mn.getMovieName() + "\t" + mn.getRate());
        }
    }

    private static void init() throws IOException {
        conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "master,slave1,slave2");
        conn = ConnectionFactory.createConnection(conf);
        admin = (HBaseAdmin) conn.getAdmin();
    }

    private static boolean existTable(String tableName) throws IOException {
        if (admin.tableExists(tableName)) {
            System.out.println(tableName + "已存在");
            return true;
        } else {
            System.out.println(tableName + "不存在");
            return false;
        }
    }
}