flink 自定义hbasesink 批量写入数据到hbase

如果你想使用 Flink 批量将数据写入 HBase，可以自定义一个 HBaseSinkFunction。下面是一个简单的示例：

public class HBaseBatchSinkFunction extends RichSinkFunction<List<Tuple2<String, String>>> {
    private transient Connection connection;
    private transient BufferedMutator bufferedMutator;

    @Override
    public void open(Configuration parameters) throws Exception {
        Configuration config = HBaseConfiguration.create();
        config.set("hbase.zookeeper.quorum", "localhost");
        config.set("hbase.zookeeper.property.clientPort", "2181");
        config.set("zookeeper.znode.parent", "/hbase");
        config.set("hbase.client.write.buffer", "10000000");
        config.set("hbase.client.retries.number", "3");
        connection = ConnectionFactory.createConnection(config);
        TableName tableName = TableName.valueOf("my_table");
        BufferedMutatorParams params = new BufferedMutatorParams(tableName);
        params.writeBufferSize(1024 * 1024);
        bufferedMutator = connection.getBufferedMutator(params);
    }

    @Override
    public void invoke(List<Tuple2<String, String>> values, Context context) throws Exception {
        List<Put> puts = new ArrayList<>();
        for (Tuple2<String, String> value : values) {
            Put put = new Put(Bytes.toBytes(value.f0));
            put.addColumn(Bytes.toBytes("my_cf"), Bytes.toBytes("my_col"), Bytes.toBytes(value.f1));
            puts.add(put);
        }
        bufferedMutator.mutate(puts);
    }

    @Override
    public void close() throws Exception {
        if (bufferedMutator != null) {
            bufferedMutator.flush();
            bufferedMutator.close();
        }
        if (connection != null) {
            connection.close();
        }
    }
}

在这个自定义的 HBaseSinkFunction 中，我们使用 BufferedMutator 批量写入数据。在 open() 方法中，我们获取 HBase 连接和缓冲器。在 invoke() 方法中，我们将数据转换为 Put 对象，并添加到缓冲器中。最后，在 close() 方法中，我们刷新缓冲器并关闭连接。

在你的 Flink 程序中，你可以使用这个自定义的 HBaseSinkFunction，例如：

DataStream<Tuple2<String, String>> dataStream = ...;
dataStream.addSink(new HBaseBatchSinkFunction());

这样，你就可以批量将数据写入 HBase 了。