Flink可以通過HBaseInputFormat來讀取HBase表中的數據。以下是一個簡單的示例代碼：

import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.io.TextInputFormat;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.typeutils.TypeExtractor;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableInputFormat;

public class ReadHBaseData {

    public static void main(String[] args) throws Exception {
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        org.apache.hadoop.conf.Configuration configuration = HBaseConfiguration.create();
        configuration.set(TableInputFormat.INPUT_TABLE, "<your_hbase_table_name>");

        DataSet<Tuple2<ImmutableBytesWritable, Result>> hbaseData = env.createInput(new org.apache.hadoop.hbase.mapreduce.TableInputFormat<>(
                new TextInputFormat(),
                ImmutableBytesWritable.class,
                Result.class));

        DataSet<Tuple2<String, String>> mappedData = hbaseData.map(value -> {
            String key = new String(value.f0.get());
            String data = value.f1.toString();
            return new Tuple2<>(key, data);
        });

        mappedData.print();
    }
}

請確保在pom.xml文件中包含必要的Flink和HBase依賴項。您還需要將<your_hbase_table_name>替換為您要讀取的HBase表的名稱。