flink读取kafka数据自定义组合成hbase中rowkey，列簇

可以使用 Flink 的 Kafka Consumer 和 HBase Sink 来实现这个功能。具体步骤如下：

1. 创建 Flink Kafka Consumer，读取 Kafka 中的数据。

FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>("topic", new SimpleStringSchema(), properties);
DataStream<String> stream = env.addSource(kafkaConsumer);

2. 对数据进行处理，自定义组合成 HBase 中的 RowKey 和列簇。

DataStream<Tuple2<String, String>> hbaseStream = stream.map(new MapFunction<String, Tuple2<String, String>>() {
    @Override
    public Tuple2<String, String> map(String value) throws Exception {
        // 自定义生成 RowKey 和列簇
        String rowkey = generateRowkey(value);
        String columnFamily = generateColumnFamily(value);
        return new Tuple2<>(rowkey, columnFamily);
    }
});

3. 创建 HBase Sink，将数据写入 HBase 中。

Configuration hbaseConf = HBaseConfiguration.create();
hbaseConf.set("hbase.zookeeper.quorum", "localhost");
hbaseConf.set("hbase.zookeeper.property.clientPort", "2181");

DataStreamSink<Tuple2<String, String>> sink = hbaseStream.addSink(new HBaseSink(hbaseConf, "tableName"));

其中，`generateRowkey` 和 `generateColumnFamily` 方法需要根据业务逻辑自行实现。HBaseSink 实现可参考以下代码：

public class HBaseSink extends RichSinkFunction<Tuple2<String, String>> {

    private Configuration hbaseConf;
    private String tableName;
    private transient Connection connection;
    private transient Table table;

    public HBaseSink(Configuration hbaseConf, String tableName) {
        this.hbaseConf = hbaseConf;
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        connection = ConnectionFactory.createConnection(hbaseConf);
        table = connection.getTable(TableName.valueOf(tableName));
    }

    @Override
    public void close() throws Exception {
        table.close();
        connection.close();
    }

    @Override
    public void invoke(Tuple2<String, String> value, Context context) throws Exception {
        Put put = new Put(Bytes.toBytes(value.f0));
        byte[] cfBytes = Bytes.toBytes(value.f1);
        put.addColumn(cfBytes, Bytes.toBytes("qualifier"), Bytes.toBytes("value"));
        table.put(put);
    }
}