// HBase配置
// 需从HBase的配置文件目录中分别下载core-site.xml,hdfs-site.xml,hbase-site.xml
Configuration hbaseConf = HBaseConfiguration.create();
hbaseConf.addResource(new Path("./conf/core-site.xml"));
hbaseConf.addResource(new Path("./conf/hdfs-site.xml"));
hbaseConf.addResource(new Path("./conf/hbase-site.xml"));

// 构建客户端
// 当3个配置文件存在项目根目录下时，可以不指定config
HBaseClient client = new EveHBase.Builder()
        .config(hbaseConf)
        .build();

// HBase配置
// ……

// 安全认证配置
// 需从集群管理界面下载对应用户的kerberos文件
SecurityConf securityConf = new SecurityConf(
        "test",
        "./kerberos/user.keytab",
        "./kerberos/krb5.conf");

// 构建安全模式客户端
HBaseClient client = new EveHBase.Builder()
        .config(hbaseConf)
        .enableSafeSupport(securityConf)
        .build();

ExecutorServiceAdapter adapter = new ExecutorServiceAdapter() {
            @Override
            public ExecutorService generateExecutorService() {
                return Executors.newFixedThreadPool(4);
            }
        };
HBaseClient client = new EveHBase.Builder()
        .config(hbaseConf)
        .pool(adapter)
        .build();

EveTable eveTable = new EveTable.Builder()
        .table("tb_test")
        .familys("family01")
        .encoding(DataBlockEncoding.FAST_DIFF)
        .compression(Compression.Algorithm.SNAPPY)
        .build();

client.create(eveTable);

client.disableAndDelete("tableName");

String[] splitKeys = {"aaa", "ddd", "hhh", "vvv"};
client.multiSplit("tb_test", splitKeys);

client.createIndex(
    "tb_test", 
    "family01",
    "phone",
    "tb_test_phone_idx");

public class Person {

    @RowKey
    public String rowkey;

    // 列簇, 对应family 
    @ColumnFamily
    public Info f;

    public class Info {

        // 列簇中的字段, 对应qualifier
        public String address;

        public String phone;

        public String name;

        public String age;
        
    }
    
}

String[] qualifiers = {"address", "phone", "name", "age", "lac"};
EveGet eveGet = new EveGet.Builder()
        .table("tb_test")
        .addRowkey("ccc_21312312")
        .addRowkey("hhh_24242348")
        .select("f", qualifiers)
        .build();

try {
    List<Person> personList = client.get(eveGet, Person.class);
    System.out.println("person = " + personList);
} catch (IOException e) {
    e.printStackTrace();
}

client.getAsync(eveGet, new ResultCallback<Person>() {
    @Override
    public void onSuccessful(List<Person> results) {
        for (Person result : results) {
            System.out.println("result = " + result);
        }
    }

    @Override
    public void onFailed(Exception e) {
        System.out.println("e = " + e);
        e.printStackTrace();
    }
});

// 传入Result.class将返回HBase原生的结果
List<Result> personList = client.get(eveGet, Result.class);

String[] qualifiers = {"address", "phone", "name", "age", "lac"};

EveScan eveScan = new EveScan.Builder()
        .table("tb_test")
        .startRow("aaa")
        .endRow("hhh")
//      .filter(...)
        .select("f", qualifiers)
        .build();
try {
    List<Person> personList = client.scan(eveScan, Person.class);
    System.out.println("personList = " + personList);
} catch (IOException e) {
    e.printStackTrace();
}

client.scanAsync(eveScan, new ResultCallback<Person>() {
    @Override
    public void onSuccessful(List<Person> results) {
        for (Person result : results) {
            System.out.println("result = " + result);
        }
    }

    @Override
    public void onFailed(Exception e) {

    }
});

List<Put> putList = new ArrayList<>();
for (int i = 0; i < 5; i++) {
    Put put = new Put(Bytes.toBytes("ggg_132131" + i));
    put.addColumn(Bytes.toBytes("f"), Bytes.toBytes("phone"), Bytes.toBytes("1891234567"+ i));
    putList.add(put);
}
try {
    client.put("tb_test", putList);
} catch (IOException e) {
    e.printStackTrace();
}

PutBuffer buffer = client.createPutBuffer("test_table", 1000, 5000);
for (int i = 0; i < 100_000; i++) {
    Put put = new Put(Bytes.toBytes("ggg_132131" + i));
    put.addColumn(Bytes.toBytes("f"), Bytes.toBytes("phone"), Bytes.toBytes("1891234567" + i));
    
    buffer.put(put);
}
buffer.flush();

client.putAsync("tb_test", putList, new PutCallback() {
    @Override
    public void onSuccessful() {
        System.out.println("GetTest.onSuccessful 提交Put成功!");
    }

    @Override
    public void onFailed(Exception e, List<Put> puts) {
        System.out.println("GetTest.onFailed 提交Put失败!");
        System.out.println("失败的 puts = " + puts);
    }
});

client.putAsync("tb_test", putList, null);