package cn.idcast.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellScanner;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hbase.thirdparty.org.eclipse.jetty.util.Scanner;
import org.junit.Before;
import org.junit.Test;

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

public class HbaseClientDML {
    private Connection connection=null;
    @Before
    public void getConnection() throws IOException {
        //构建连接对象
        Configuration configuration = HBaseConfiguration.create();
        configuration.set("hbase.zookeeper.quorum","node1:2181,node2:2181,node3:2181");
        connection = ConnectionFactory.createConnection(configuration);
    }
    //插入数据
    @Test
    public void testPut() throws IOException {
        TableName tb=TableName.valueOf("user_info");
        //获取一个指定表的table对象,执行DML操作
        Table table = connection.getTable(tb);
        //增加数据
        Put put = new Put(Bytes.toBytes("1"));
        put.addColumn(Bytes.toBytes("base_info"),Bytes.toBytes("username"),Bytes.toBytes("张三"));
        put.addColumn(Bytes.toBytes("base_info"),Bytes.toBytes("age"),Bytes.toBytes("19"));
        put.addColumn(Bytes.toBytes("extra_info"),Bytes.toBytes("address"),Bytes.toBytes("河北"));

        Put put2 = new Put(Bytes.toBytes("2"));
        put2.addColumn(Bytes.toBytes("base_info"),Bytes.toBytes("username"),Bytes.toBytes("李四"));
        put2.addColumn(Bytes.toBytes("base_info"),Bytes.toBytes("age"),Bytes.toBytes("19"));
        put2.addColumn(Bytes.toBytes("extra_info"),Bytes.toBytes("address"),Bytes.toBytes("邢台"));

        ArrayList<Put> puts = new ArrayList<>();
        puts.add(put);
        puts.add(put2);
        //插数据
        table.put(puts);
        connection.close();
        table.close();
    }
    //循环插入大量数据
    @Test
    public void testManyPut() throws IOException {
        TableName tb=TableName.valueOf("user_info");
        //获取一个指定表的table对象,执行DML操作
        Table table = connection.getTable(tb);
        ArrayList<Put> puts = new ArrayList<>();
        for(int i=0;i<1000;i++){
            //增加数据
            Put put = new Put(Bytes.toBytes(""+i));
            put.addColumn(Bytes.toBytes("base_info"),Bytes.toBytes("username"),Bytes.toBytes("张三"+i));
            put.addColumn(Bytes.toBytes("base_info"),Bytes.toBytes("age"),Bytes.toBytes((19+i)+""));
            put.addColumn(Bytes.toBytes("extra_info"),Bytes.toBytes("address"),Bytes.toBytes("河北"));
            puts.add(put);
        }
        table.put(puts);

        connection.close();
        table.close();
    }
    //删除数据
    @Test
    public void testDelete() throws IOException {
        TableName tb=TableName.valueOf("user_info");
        //获取一个指定表的table对象,执行DML操作
        Table table = connection.getTable(tb);

        Delete delete = new Delete(Bytes.toBytes(1));
        Delete delete2 = new Delete(Bytes.toBytes("2"));
        delete2.addColumn(Bytes.toBytes("extra_info"),Bytes.toBytes("address"));

        ArrayList<Delete> dels = new ArrayList<>();
        dels.add(delete);
        dels.add(delete2);
        table.delete(dels);

        table.close();
        connection.close();
    }
    @Test
    public void testGet() throws IOException {
        TableName tb=TableName.valueOf("user_info");
        //获取一个指定表的table对象,执行DML操作
        Table table = connection.getTable(tb);
        Get get = new Get("1".getBytes());
        Result result = table.get(get);
        byte[] row = result.getRow();
        CellScanner cellScanner = result.cellScanner();
        while(cellScanner.advance()){
            Cell cell = cellScanner.current();

            byte[] rowArray = cell.getRowArray();
            byte[] familyArray = cell.getFamilyArray();
            byte[] qualifierArray = cell.getQualifierArray();
            byte[] valueArray = cell.getValueArray();

            System.out.println("行键："+new String(rowArray,cell.getRowOffset(),cell.getRowLength()));
            System.out.println("列族名："+new String(familyArray,cell.getFamilyOffset(),cell.getFamilyLength()));
            System.out.println("列名："+new String(qualifierArray,cell.getQualifierOffset(),cell.getQualifierLength()));
            System.out.println("value："+new String(valueArray,cell.getValueOffset(),cell.getValueLength()));

        }
        table.close();
        connection.close();
    }
    //按行键查询数据
    @Test
    public void testScan() throws IOException {
        TableName tb=TableName.valueOf("user_info");
        //获取一个指定表的table对象,执行DML操作
        Table table = connection.getTable(tb);
        Scan scan = new Scan("10".getBytes(),"10000".getBytes());
        ResultScanner scanner = table.getScanner(scan);
        Iterator<Result> iterator = scanner.iterator();
        while(iterator.hasNext()){
            Result result = iterator.next();
            CellScanner cellScanner = result.cellScanner();
            while(cellScanner.advance()){
                Cell cell = cellScanner.current();

                byte[] rowArray = cell.getRowArray();
                byte[] familyArray = cell.getFamilyArray();
                byte[] qualifierArray = cell.getQualifierArray();
                byte[] valueArray = cell.getValueArray();

                System.out.println("行键："+new String(rowArray,cell.getRowOffset(),cell.getRowLength()));
                System.out.println("列族名："+new String(familyArray,cell.getFamilyOffset(),cell.getFamilyLength()));
                System.out.println("列名："+new String(qualifierArray,cell.getQualifierOffset(),cell.getQualifierLength()));
                System.out.println("value："+new String(valueArray,cell.getValueOffset(),cell.getValueLength()));

            }
            System.out.println("-----------------------------");
        }


    }
}