Flink 数据写入 HBase 表教程：完整代码示例

在 Flink 中将数据写入 HBase 表，需要使用 HBase 的 sink 函数以及相关的 HBase 连接参数和表结构信息。以下是一个示例代码：

import org.apache.flink.addons.hbase.HBaseTableSchema;
import org.apache.flink.addons.hbase.TableInputFormat;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.io.jdbc.JDBCInputFormat;
import org.apache.flink.api.java.io.jdbc.JDBCOutputFormat;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011;
import org.apache.flink.types.Row;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapred.TableOutputFormat;
import org.apache.hadoop.hbase.util.Bytes;

import java.util.Properties;

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

        // 设置 Kafka 消费者属性
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "localhost:9092");
        properties.setProperty("group.id", "test");

        // 创建 Kafka 数据源
        FlinkKafkaConsumer011<String> consumer = new FlinkKafkaConsumer011<>( "test", new SimpleStringSchema(), properties);

        // 将 Kafka 数据源添加到执行环境
        DataStream<String> stream = env.addSource(consumer);

        // 解析传入的消息并提取必要字段
        DataStream<Tuple2<String, String>> dataStream = stream.map(new MapFunction<String, Tuple2<String, String>>() {
            @Override
            public Tuple2<String, String> map(String value) throws Exception {
                String[] fields = value.split(",");
                return new Tuple2<>(fields[0], fields[1]);
            }
        });

        // 定义 HBase 连接参数
        org.apache.hadoop.conf.Configuration hConf = org.apache.hadoop.hbase.HBaseConfiguration.create();
        hConf.set(TableOutputFormat.OUTPUT_TABLE, "test");
        hConf.set("hbase.zookeeper.quorum", "localhost");
        hConf.set("hbase.zookeeper.property.clientPort", "2181");

        // 定义 HBase 表结构
        HBaseTableSchema schema = new HBaseTableSchema();
        schema.addColumn("cf", "col1", TypeInformation.of(String.class));
        schema.addColumn("cf", "col2", TypeInformation.of(String.class));

        // 创建 HBase sink 函数
        TableOutputFormat<Put> outputFormat = new TableOutputFormat<>();
        outputFormat.configure(hConf);
        DataStream<Tuple2<String, String>> hbaseStream = dataStream.map(new MapFunction<Tuple2<String, String>, Tuple2<ImmutableBytesWritable, Put>>() {
            @Override
            public Tuple2<ImmutableBytesWritable, Put> map(Tuple2<String, String> value) throws Exception {
                Put put = new Put(Bytes.toBytes(value.f0));
                put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("col1"), Bytes.toBytes(value.f0));
                put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("col2"), Bytes.toBytes(value.f1));
                return new Tuple2<>(new ImmutableBytesWritable(Bytes.toBytes(value.f0)), put);
            }
        });
        hbaseStream.writeUsingOutputFormat(outputFormat);

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

在上述示例中，首先使用 Kafka 数据源接收数据流，然后使用 MapFunction 解析数据并提取必要的字段。接下来，定义 HBase 连接参数和表结构信息，并使用 TableOutputFormat 创建一个 HBase sink 函数。将数据流映射到一个 Tuple2 格式的数据流中，并使用 MapFunction 将其转换为一个 Put 对象。最后，将 Put 对象写入 HBase 表中。

需要注意的是，如果 HBase 表已经存在，则可以使用 hbaseStream.addSink(new HBaseSinkFunction<>(hConf, schema, new HBasePutMapper())); 来将数据写入表中。其中，HBasePutMapper 是一个实现了 MapFunction 接口的类，用于将 Tuple2 转换为 Put 对象。在 HBaseSinkFunction 构造函数中，需要传入 HBase 连接参数、表结构信息和 Put Mapper。