Flink CDC中除了hive有没有那些引擎支持查询hbase的啊？

除了Hive，Flink CDC还支持从HBase中查询数据。Flink提供了HBase源插件，可以从HBase中读取增量数据，并将其转换为Flink的数据格式。用户可以通过编写Flink应用程序代码，将HBase数据查询到Flink中，进行进一步的处理和分析。

phoenix，此回答整理自钉群“Flink CDC 社区”

在 Flink CDC 中，除了支持 Hive 外，还可以通过 Flink 的 Table API 和 SQL 来查询 HBase 数据。下面是一个示例代码：

import org.apache.flink.api.common.typeinfo.Types;
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.table.api.*;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.descriptors.HBase;
import org.apache.flink.table.descriptors.Schema;
import org.apache.flink.table.descriptors.ConnectorDescriptor;
import org.apache.flink.table.descriptors.Json;

public class HBaseQueryExample {

    public static void main(String[] args) throws Exception {
        // 创建 Execution Environment
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, settings);

        // 定义 HBase 的连接器描述符和格式化器描述符
        ConnectorDescriptor hbaseConnector = new HBase()
            .version("1.4.3")
            .table("table-name")
            .zookeeperQuorum("zk-host")
            .zookeeperPort("zk-port");

        FormatDescriptor hbaseFormat = new Json().failOnMissingField(true);

        // 定义 HBase 表的 Schema
        Schema hbaseSchema = new Schema()
            .field("rowkey", DataTypes.STRING())
            .field("columnFamily", DataTypes.STRING())
            .field("columnQualifier", DataTypes.STRING())
            .field("value", DataTypes.STRING());

        // 创建 HBase 表
        tEnv.connect(hbaseConnector).withFormat(hbaseFormat).withSchema(hbaseSchema).createTemporaryTable("hbase_table");

        // 注册 HBase 表
        tEnv.registerTable("hbase_table");

        // 执行查询
        Table queryResult = tEnv.sqlQuery("SELECT * FROM hbase_table");

        // 转换结果为 DataStream
        DataStream<Tuple2<Boolean, Row>> resultStream = tEnv.toRetractStream(queryResult, Types.ROW(Types.STRING, Types.STRING, Types.STRING, Types.STRING));

        // 输出查询结果
        resultStream.print();

        // 提交作业并执行
        env.execute("HBase Query Job");
    }
}