package com.example.hbaseproject.dir01;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.NamespaceDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;

class Main{
    /************************* 获取 hbase 单线程连接的方法，其相当重，适合一个进程共用一个 *************************/
    public void fun01() throws IOException {
        Configuration conf = new Configuration();
        // 等于 hbase-site.xml 文件中的配置项
        conf.set("hbase.zookeeper.quorum", "ubuntu01,ubuntu02,ubuntu03");
        // 异步连接
        // CompletableFuture<AsyncConnection> asyncConn = ConnectionFactory.createAsyncConnection();
        // 默认同步连接，记得 close
        try(Connection conn = ConnectionFactory.createConnection(conf);){

        }
    }

    /************** hbase 是线程安全的，内部的Table和Admin不是线程安全的，也不建议缓存或池化二者 **************/
    public static Connection connection = null;
    static {
        Configuration conf = new Configuration();
        conf.set("hbase.zookeeper.quorum", "ubuntu01,ubuntu02,ubuntu03");
        try {
            // 无参的工厂方法会到 hbase-default.xml hbase-site.xml 中读取配置，那样无需在代码中配置
            connection = ConnectionFactory.createConnection(conf);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
    public static void closeConnection() throws IOException {
        if (connection != null) {
            connection.close();
        }
    }
    /***************************************** DDL *****************************************/
    // 链接的IO异常应当由方法调用者处理
    public static void createNameSpace() throws IOException {
        // 用于操作 DDL，轻量级链接，线程不安全，不推荐池化或缓存
        Admin admin = connection.getAdmin();

        NamespaceDescriptor.Builder builder = NamespaceDescriptor.create("namespace");
        builder.addConfiguration("key", "value");
        // 创建命名空间出现的问题，属于本方法自身的问题，不该抛出
        try {
            admin.createNamespace(builder.build());
        } catch (IOException e) {
            System.out.println("log : 命名空间已存在");
            e.printStackTrace();
        }

        admin.close();
    }
    public static boolean isTableExits(String namespace, String tableName) throws IOException {
        Admin admin = connection.getAdmin();
        boolean exists = false;
        try {
            exists = admin.tableExists(TableName.valueOf(namespace, tableName));
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        admin.close();
        return exists;
    }
    public static void createTable() throws IOException {
        Admin admin = connection.getAdmin();
        // 创建对列族的描述
        ColumnFamilyDescriptorBuilder columnFamilyDescriptorBuilder =
                ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes("columnFamily01"));
        columnFamilyDescriptorBuilder.setMaxVersions(3);
        ColumnFamilyDescriptor columnFamilyDescriptor = columnFamilyDescriptorBuilder.build();
        // 创建对表的描述
        // 这种方式得到的是一个新的表描述，没有已存在表的信息，不能用于修改
        TableDescriptorBuilder tableDescriptorBuilder = 
            TableDescriptorBuilder.newBuilder(TableName.valueOf("tablename"));
        tableDescriptorBuilder.setColumnFamily(columnFamilyDescriptor);
        TableDescriptor tableDescriptor = tableDescriptorBuilder.build();

        try {
            admin.createTable(tableDescriptor);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }

        admin.close();
    }
    public static void modifyTable() throws IOException {
        Admin admin = connection.getAdmin();
        // 通过这个方法获得已存在的表的描述
        TableDescriptor existingTable = admin.getDescriptor(TableName.valueOf("namespace", "ExistingTable"));
        TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(existingTable);

        // 新的列族
        ColumnFamilyDescriptorBuilder columnFamilyDescriptorBuilder = 
            ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes("newColumnFamily"));
        ColumnFamilyDescriptor newColumnFamilyDescriptor = columnFamilyDescriptorBuilder.build();

        // 注意：没有在这一行中指定的列族都会被设置为初始值。如果想原有的值不变，需要从 existingTable 获取旧的列并插入
        tableDescriptorBuilder.modifyColumnFamily(newColumnFamilyDescriptor);
        admin.close();
    }
    
    // TableName tableName = TableName.valueOf("namespace", "table");
    // admin.disable(tableName);
    // admin.deleteTable(tableName);
}

 package com.example.hbaseproject.dir01;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.ColumnValueFilter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.Arrays;

class Main{
    public static Connection connection = null;
    static {
        Configuration conf = new Configuration();
        conf.set("hbase.zookeeper.quorum", "ubuntu01,ubuntu02,ubuntu03");
        try {
            // 无参的工厂方法会到 hbase-default.xml hbase-site.xml 中读取配置，那样无需在代码中配置
            connection = ConnectionFactory.createConnection(conf);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
    public static void putCell(String namespace, String tableName, String rowKey,
                               String columnFamily, String columnName, String value) throws IOException {
        Table table = connection.getTable(TableName.valueOf(namespace, tableName));
        // 创建一个要插入的数据单元
        Put cell = new Put(Bytes.toBytes(rowKey));
        // 指定命名空间、列族、列名等
        cell.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(columnName), Bytes.toBytes(value));
        try {
            table.put(cell);
        } catch (IOException e) {
            e.printStackTrace();
        }
        table.close();
    }
    public static Cell[] getCell(String namespace, String tableName, String rowKey,
                               String columnFamily, String columnName) throws IOException {
        Table table = connection.getTable(TableName.valueOf(namespace, tableName));
        // 创建要查询的行对象
        Get get = new Get(Bytes.toBytes(rowKey));
        // 添加指定列
        get.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(columnName));
        // 指定读取所有的版本
        get.readAllVersions();

        Cell[] cells = null;
        try {
            Result result = table.get(get);
            // 获得cell数组
            cells = result.rawCells();

            Arrays.stream(cells).forEach(cell -> {
                // 真正的数据包裹在cell内部
                String value = new String(CellUtil.cloneValue(cell));
                System.out.println("value = " + value);
            });
        } catch (IOException e) {
            e.printStackTrace();
        }
        table.close();
        return cells;
    }

    public static void scanRows(String namespace, String tableName,
                                String startRow, String stopRow,
                                String columnFamily, String columnQualifier, String columnValue) throws IOException {
        Table table = connection.getTable(TableName.valueOf(namespace, tableName));
        // 得到 scan 对象，默认扫描整张表
        Scan scan = new Scan();

        // 过滤行
        scan.withStartRow(Bytes.toBytes(startRow) /*, true 默认包含当前这条*/);
        scan.withStopRow(Bytes.toBytes(stopRow) /*,false 默认不包含此条*/);

        // 过滤列
        FilterList filterList = new FilterList();
        // 1. 查询结果只保留指定列的数据
        ColumnValueFilter columnValueFilter = new ColumnValueFilter(Bytes.toBytes(columnFamily),
                Bytes.toBytes(columnQualifier), CompareOperator.EQUAL, Bytes.toBytes(columnValue));
        // 2. 查询结果保留匹配的整行数据，结果也会保留没有当前列的数据
        SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter(Bytes.toBytes(columnFamily),
                Bytes.toBytes(columnQualifier), CompareOperator.EQUAL, Bytes.toBytes(columnValue));
        // filterList.addFilter(columnValueFilter);
        filterList.addFilter(singleColumnValueFilter);

        scan.setFilter(filterList);

        try {
            // 读取多行数据
            // result 是一行数据，包含一个 cell 数组
            // ResultScanner 是多行数据，包含一个 result 数组
            ResultScanner resultScanner = table.getScanner(scan);
            resultScanner.forEach(result -> {
                Arrays.stream(result.rawCells()).forEach(cell -> {
                    System.out.printf("%s - %s:%s -%s",
                            new String(CellUtil.cloneRow(cell)),
                            new String(CellUtil.cloneFamily(cell)),
                            new String(CellUtil.cloneQualifier(cell)),
                            new String(CellUtil.cloneValue(cell)));
                });
                System.out.println();
            });
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    public static void deleteCell(String namespace, String tableName, String rowKey,
                                  String columnFamily, String columnQualifier) throws IOException {
        Table table = connection.getTable(TableName.valueOf(namespace, tableName));
        Delete delete = new Delete(Bytes.toBytes(rowKey));

        // 对应命令 delete，删除一个版本
        // delete.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(columnQualifier));
        // 对应命令 deleteall，删除所有版本
        delete.addColumns(Bytes.toBytes(columnFamily), Bytes.toBytes(columnQualifier));
        try {
            table.delete(delete);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        table.close();
    }
}