Flink1.8如何批量Sink到HBase

这篇文章主要为大家展示了“Flink1.8如何批量Sink到HBase”，内容简而易懂，条理清晰，希望能够帮助大家解决疑惑，下面让小编带领大家一起研究并学习一下“Flink1.8如何批量Sink到HBase”这篇文章吧。

实现背景：

消费Kafka数据写入HBase时，单条处理效率太低。需要批量插入hbase,这里自定义时间窗口countWindowAll 实现100条hbase插入一次Hbase

前面我就不写了直接上核心代码

/*每10秒一个处理窗口*/DataStream<List<Put>> putList = filterData.countWindowAll(Constants.windowCount).apply(new AllWindowFunction<String, List<Put>, GlobalWindow>() {    @Override    public void apply(GlobalWindow window, Iterable<String> message, Collector<List<Put>> out) throws Exception {        List<Put> putList=new ArrayList<Put>();        for (String value : message)        {            String rowKey=value.replace("::","_");            Put put = new Put(Bytes.toBytes(rowKey.toString()));            String[] column=value.split("::");            for (int i = 0; i < column.length; i++) {                put.addColumn(Bytes.toBytes(Constants.columnFamily),Bytes.toBytes(Constants.columnArray[i]),Bytes.toBytes(column[i]));            }            putList.add(put);        }        out.collect(putList);    }    }).setParallelism(4);

putList.addSink(new HBaseSinkFunction()).setParallelism(1);

这里sink需要继承Flink的RichSinkFunction接口，实现其中的三个比较重要的函数：

1.open()任务开始只调用一次

2.invoke(）每接收一条记录调用一次，多条记录调用多次

3.close()任务关闭只调用一次

写HBase自定义Sink为

HBaseSinkFunction extends RichSinkFunction<List<Put>>{@Overridepublic void open(Configuration parameters) throws Exception {    super.open(parameters);    HbaseUtils.connectHbase();    TableName table=TableName.valueOf(Constants.tableNameStr);    Admin admin = HbaseUtils.connection.getAdmin();    if(!admin.tableExists(table)){        HTableDescriptor tableDescriptor = new HTableDescriptor(Constants.tableNameStr);        tableDescriptor.addFamily(new HColumnDescriptor(Constants.columnFamily));        admin.createTable(tableDescriptor);    }}@Overridepublic void invoke(List<Put> putList, Context context) throws Exception {    Table table=HbaseUtils.connection.getTable(TableName.valueOf(Constants.tableNameStr));    table.put(putList);}@Overridepublic void close() throws Exception {    super.close();    HbaseUtils.closeHBaseConnect();}}

以上是“Flink1.8如何批量Sink到HBase”这篇文章的所有内容，感谢各位的阅读！相信大家都有了一定的了解，希望分享的内容对大家有所帮助，如果还想学习更多知识，欢迎关注行业资讯频道！

Flink1.8如何批量Sink到HBase

相关推荐