import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.io.compress.Compression;
import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
import org.apache.hadoop.hbase.util.Bytes;

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

/**
 * @ProjectName: hbase-demo
 * @Package: PACKAGE_NAME
 * @ClassName: PutExample
 * @Description:
 * @Author: yehui.mao
 * @CreateDate: 2019/7/24 12:12
 * @UpdateUser: yehui.mao
 */
public class PutExample {
    private static final String TABLE_NAME = "test";

    public static Configuration conf = null;
    public HTable table = null;
    public HBaseAdmin admin = null;
    public static Connection conn;

    static {
        try {
            conf = HBaseConfiguration.create();
            conf.set("hbase.zookeeper.quorum", "127.0.0.1");
            conf.set("hbase.zookeeper.property.clientPort", "2181");
            conf.set("hbase.master", "127.0.0.1:60000");
            conn = ConnectionFactory.createConnection(conf);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * program entrance
     *
     * @param args
     * @throws IOException
     */
    public static void main(String[] args) throws IOException {
        createTable();
        addRecord(TABLE_NAME, "1", "cf1", "name", "robbin");
        addRecord(TABLE_NAME, "2", "cf1", "name", "ruby");
        removeRecord(TABLE_NAME, "2");
        getRecord(TABLE_NAME, "123", 3);
        scanTable(TABLE_NAME);
        deleteTable();
    }

    /**
     * create table
     *
     * @throws IOException
     */
    public static void createTable() {
        try {
            Admin admin = conn.getAdmin();

            if (!admin.tableExists(TableName.valueOf(TABLE_NAME))) {
                TableDescriptorBuilder tableDescBuilder = TableDescriptorBuilder.newBuilder(TableName.valueOf(TABLE_NAME));
                ColumnFamilyDescriptorBuilder columnDescBuilder = ColumnFamilyDescriptorBuilder
                        .newBuilder(Bytes.toBytes("cf1")).setBlocksize(32 * 1024)
                        .setCompressionType(Compression.Algorithm.NONE).setDataBlockEncoding(DataBlockEncoding.NONE);
                tableDescBuilder.setColumnFamily(columnDescBuilder.build());
                admin.createTable(tableDescBuilder.build());
            } else {
                System.out.println("table exists");
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * delete table
     */
    public static void deleteTable() {
        try {
            Admin admin = conn.getAdmin();
            admin.disableTable(TableName.valueOf(TABLE_NAME));
            admin.deleteTable(TableName.valueOf(TABLE_NAME));
            System.out.println("delete table: " + TABLE_NAME + " done.");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * insert data
     */
    public static void addRecord(String tableName, String rowKey, String family, String qualifier, String value) {
        try {
            Table table = conn.getTable(TableName.valueOf(tableName));
            Put put = new Put(Bytes.toBytes(rowKey));
            put.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier), Bytes.toBytes(value));

            try {
                table.put(put);
            } catch (IOException e) {
                e.printStackTrace();
            }

            System.out.println("insert recored " + rowKey + " to table " + tableName + " ok.");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * delete record
     */
    public static void removeRecord(String tableName, String rowKey) {
        try {
            Table table = conn.getTable(TableName.valueOf(tableName));
            List list = new ArrayList();
            Delete del = new Delete(rowKey.getBytes());
            list.add(del);
            table.delete(list);
            System.out.println("del recored " + rowKey + " ok.");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * query record
     */
    public static void getRecord(String tableName, String rowKey, int versions) {
        try {
            Table table = conn.getTable(TableName.valueOf(tableName));
            Get get = new Get(rowKey.getBytes());
            get.readVersions(versions);
            if (get.isCheckExistenceOnly()) {
                Result rs = table.get(get);
                for (Cell cell : rs.listCells()) {
                    System.out.println(Bytes.toString(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength()));
                    System.out.println(Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength()));
                    System.out.println(Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength()));
                    System.out.println(Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()));
                    System.out.println(cell.getTimestamp());
                    System.out.println("------");
                }
            } else {
                System.out.println("not found data according to rowkey");
            }

        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void getRecord(String tableName, String rowKey) {
        getRecord(tableName, rowKey, 1);
    }

    /**
     * scan table
     *
     * @param tableName
     */
    public static void scanTable(String tableName) {
        try {
            Table table = conn.getTable(TableName.valueOf(tableName));
            Scan s = new Scan();

            ResultScanner rs = table.getScanner(s);
            for (Result r : rs) {
                for (Cell cell : r.listCells()) {
                    System.out.println(Bytes.toString(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength()));
                    System.out.println(Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength()));
                    System.out.println(Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength()));
                    System.out.println(Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()));
                    System.out.println(cell.getTimestamp());
                    System.out.println("------");
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

}