@Override
    public DataStreamSink<Tuple2<Boolean, Row>> consumeDataStream(DataStream<Tuple2<Boolean, Row>> dataStream) {
        HbaseOutputFormat.HbaseOutputFormatBuilder builder = HbaseOutputFormat.buildHbaseOutputFormat();

        HbaseOutputFormat outputFormat = builder
                .setHost(this.zookeeperQuorum)
                .setZkParent(this.parent)
                .setTable(this.tableName)
                .setRowkey(rowkey)
                .setColumnNames(fieldNames)
                .setColumnNameFamily(columnNameFamily)
                .setKerberosAuthEnable(kerberosAuthEnable)
                .setRegionserverKeytabFile(regionserverKeytabFile)
                .setRegionserverPrincipal(regionserverPrincipal)
                .setSecurityKrb5Conf(securityKrb5Conf)
                .setZookeeperSaslClient(zookeeperSaslClient)
                .setClientPrincipal(clientPrincipal)
                .setClientKeytabFile(clientKeytabFile)
                .setBatchSize(Integer.parseInt(batchSize))
                .setBatchWaitInterval(Long.parseLong(batchWaitInterval))
                .setDirtyManager(DirtyDataManager.newInstance(dirtyProperties))
                .finish();

        RichSinkFunction richSinkFunction = new OutputFormatSinkFunction(outputFormat);
        DataStreamSink dataStreamSink = dataStream.addSink(richSinkFunction).name(registerTabName);

        if (parallelism > 0) {
            dataStreamSink.setParallelism(parallelism);
        }

        return dataStreamSink;
    }
Date: 2018/09/14
Company: www.dtstack.com

@author sishu.yss
hbase/hbase-sink/src/main/java/com/dtstack/flink/sql/sink/hbase/HbaseSink.java