flink多数据插入hbase

在 Flink 中将多个数据插入 HBase，你可以使用 HBaseSinkFunction。下面是一个简单的示例代码，演示了如何使用 Flink 将多个数据插入 HBase：

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.hadoop.HadoopOutputFormat;
import org.apache.flink.streaming.connectors.hadoop.utils.HadoopUtils;
import org.apache.flink.streaming.connectors.hbase.HBaseSinkFunction;
import org.apache.flink.streaming.connectors.hbase.HBaseTableSchema;
import org.apache.flink.streaming.connectors.hbase.util.HBaseConfigurationUtil;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;

public class FlinkHBaseInsert {
    public static void main(String[] args) throws Exception {
        // 设置执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 创建数据源，这里假设数据源是一个 Tuple2<String, Integer> 类型的流
        DataStream<Tuple2<String, Integer>> input = env.fromElements(
                new Tuple2<>("key1", 10),
                new Tuple2<>("key2", 20),
                new Tuple2<>("key3", 30)
        );

        // 创建 HBaseTableSchema，定义表名、列簇名和列名
        HBaseTableSchema schema = new HBaseTableSchema();
        schema.setTableName("my_table");
        schema.addColumnFamily("cf1");
        schema.addColumn("cf1", "col1");

        // 创建 HBaseSinkFunction，用于将数据写入 HBase
        HBaseSinkFunction<Tuple2<String, Integer>> sinkFunction = new HBaseSinkFunction<>(
                HBaseConfigurationUtil.getHBaseConfiguration(),
                schema,
                (value) -> {
                    Put put = new Put(Bytes.toBytes(value.f0));  // 使用 Tuple2 的第一个字段作为 rowkey
                    put.addColumn(Bytes.toBytes("cf1"), Bytes.toBytes("col1"), Bytes.toBytes(value.f1.toString()));  // 插入数据到 cf1:col1 列
                    return put;
                }
        );

        // 将数据流写入 HBase
        input.map(new MapFunction<Tuple2<String, Integer>, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(Tuple2<String, Integer> value) throws Exception {
                return value;
            }
        }).addSink(sinkFunction);

        // 执行任务
        env.execute("Flink HBase Insert");
    }
}

在这个示例中，我们首先创建了一个数据流 `input`，其中包含了多个 Tuple2<String, Integer> 类型的数据。然后，我们创建了一个 `HBaseTableSchema`，定义了要插入的表名、列簇名和列名。接下来，我们创建了一个 `HBaseSinkFunction`，用于将数据写入 HBase。在 `HBaseSinkFunction` 中，我们使用 `Tuple2` 的第一个字段作为 rowkey，将其转换为 `Put` 对象，并设置要插入的列簇和列。

最后，我们通过调用 `addSink` 将数据流写入 HBase。在这个示例中，我们使用了 `map` 函数将输入流中的数据转换为 `Tuple2` 类型，然后将其传递给 `HBaseSinkFunction`。

请注意，你需要根据实际情况修改示例代码中的表名、列簇名、列名和数据类型，以及配置 HBase 的连接信息。确保你已经正确配置了 HBase 的环境和依赖项。