Hi， 大罗 > 在 2020年8月28日，11:57，大罗 <tianyu1...@163.com> 写道： > > > 你好，我想提两个问题，关于flink-sql整合hbase，问题列表如下： > 问题一：flink-sql通过hive catalog整合hbase后的查询执行计划，可否使用rowkey做精确filter而不是整表扫描。 > > 问题二：flink-sql > 查询hive的hbase外部表，是否没有通过HBaseStorageHandler正确解析INPUTFORMAT和OUTPUTFORMAT >
问题一： HbaseTableSource 目前没有支持 SupportsFilterPushDown ，所以你写的filter条件没有下推到source中，做的是全表扫描。 问题二： Flink 支持 读取Hive的external表的，如果不能读就是有bug，可以贴下详细的异常栈吗？我可以看下了开个issue 祝好 Leonard > > > 两个问题的具体情景如下： > > 我所使用的版本如下： > > > > Hadoop 3.0.0+cdh6.3.2 > > HDFS 3.0.0+cdh6.3.2 > > HBase 2.1.0+cdh6.3.2 > > Hive 2.1.1+cdh6.3.2 > > Flink 1.11.1 > > > > 我在HBase里有两个表，分别为't1'和'econ_run_data_minutes'，其中, > 表't1'的行数为2行，而'econ_run_data_minutes'的行数超过1亿行。 > > > > 我在flink-sql里，例用hive catalog定义'myHbaseT1'映射't1', > 定义'myHbaseMin'映射'econ_run_data_minutes'，如下: > > > > CREATE TABLE myHbaseT1 ( > > rowkey string, > > f1 row (val VARCHAR, dqf VARCHAR), > > f2 row (val VARCHAR, dqf VARCHAR) > > ) WITH ( > > 'connector' = 'hbase-1.4', > > 'table-name' = 't1', > > 'zookeeper.quorum' = > 'dev-hadoop-node-c:2181,dev-hadoop-node-d:2181,dev-hadoop-node-e:2181' > > ); > > > > > > CREATE TABLE myHbaseMin ( > > rowkey string, > > m1 row (val VARCHAR, dqf VARCHAR, ts TIMESTAMP(3)), > > PRIMARY KEY (rowkey) NOT ENFORCED > > ) WITH ( > > 'connector' = 'hbase-1.4', > > 'table-name' = 'econ_run_data_minutes', > > 'zookeeper.quorum' = > 'dev-hadoop-node-c:2181,dev-hadoop-node-d:2181,dev-hadoop-node-e:2181' > > ); > > > > 然手，在flink-sql client里执行查询语句，"select * from myHbaseT1"，因为数据量只有两行，所以很快返回结果，如下： > > Flink SQL> select * from myHbaseT1 ; > > +-----+----------------------+----------------------+----------------------+ > > | +/- | rowkey | f1 | f2 | > > +-----+----------------------+----------------------+----------------------+ > > | + | IDIN200F.d1.P-159... | null,null | 100,0 | > > | + | row1 | 200,0 | null,null | > > +-----+----------------------+----------------------+----------------------+ > > Received a total of 2 rows > > > > Flink SQL> > > 以上测试，证明我的flink-sql环境通过hive catalog整合hbase查询是没有问题的！ > > > > 接着，我在flink-sql client里执行查询语句， > > "select * from myhbasemin where rowkey >= 'IDIN200F.d1.P-1598493600000' and > rowkey <= 'IDIN200F.d1.P-1598497200000'"， > > 这个查询结果其实只是返回不到100行的数据，但却要耗时半个小时左右，原因是因为flink做了整表扫描，执行计划如下以及核心日志如下： > > > > Source: TableSourceScan(table=[[myhive, dw, myhbasemin]], fields=[rowkey, > m1]) -> Calc(select=[rowkey, m1], where=[((rowkey >= > _UTF-16LE'IDIN200F.d1.P-1598493600000') AND (rowkey <= > _UTF-16LE'IDIN200F.d1.P-1598497200000'))]) -> SinkConversionToTuple2 -> > Sink: SQL Client Stream Collect Sink > > > > > > 2020-08-28 11:20:24,501 INFO > org.apache.flink.connector.hbase.source.AbstractTableInputFormat [] - > opening split > (this=org.apache.flink.connector.hbase.source.HBaseRowDataInputFormat@5c382732)[1|[dev-hadoop-node-c:16020]|BSBMS.d1.cu3.061c.U-15935022|BSBMS.d1.cu3.147c.U-159092] > > 2020-08-28 11:20:49,086 INFO > org.apache.flink.connector.hbase.source.AbstractTableInputFormat [] - > Closing split (scanned 4410553 rows) > > > > 我设想中的情景，flink-sql应该抓住这个rowkey，在hbase里面使用StartRow, > EndRow做精确扫描，而不是整表扫描后，在flink的task manager里面做filter ！ > > 所以，问题一：flink-sql通过hive catalog整合hbase后的查询执行计划，可否使用rowkey做精确filter而不是整表扫描。 > > > > 在问题一的基础思考之上，我想是否可以通过hive的external table整合hbase，然后在flink-sql里查询hive的externa > table，实现快速精确查询，实验过程如下： > > > > 首先，在hive里定义external table整合hbase： > > > > CREATE EXTERNAL TABLE `hive_hbase_t1`( > > `m_key` string COMMENT '', > > `f1_val` string COMMENT '', > > `f1_dqf` int COMMENT '', > > `f2_val` string COMMENT '', > > `f2_dqf` int COMMENT '', > > `ts` timestamp COMMENT '') > > ROW FORMAT SERDE > > 'org.apache.hadoop.hive.hbase.HBaseSerDe' > > STORED BY > > 'org.apache.hadoop.hive.hbase.HBaseStorageHandler' > > WITH SERDEPROPERTIES ( > > 'hbase.columns.mapping'=':key,f1:val,f1:dqf,f2:val,f2:dqf, :timestamp', > > 'serialization.format'='1') > > > > TBLPROPERTIES ( > > 'hbase.table.name'='t1' > > ); > > > > 然后，在hive里查询，"select * from hive_hbase_t1": > > > > hive> select * from hive_hbase_t1 ; > > IDIN200F.d1.P-1593484146001 NULL NULL 100 0 2020-06-30 > 10:56:07.983 > > row1 200 0 NULL NULL 2020-06-30 10:54:56.558 > > hive> > > 以上测试，证明我的环境，hive是可以通过external table整合hbase进行查询。 > > > > 接着，我尝试在flink-sql里，查询，"select * from hive_hbase_t1": > > Flink SQL> select * from hive_hbase_t1; > > Caused by: java.lang.NullPointerException > > at java.lang.Class.forName0(Native Method) > > at java.lang.Class.forName(Class.java:348) > > at > org.apache.flink.connectors.hive.read.HiveTableInputFormat.createInputSplits(HiveTableInputFormat.java:305) > > ... 24 more > > > > End of exception on server side>] > > > > Flink SQL> > > > > 经过调试，我发现，当org.apache.flink.connectors.hive.read.HiveTableInputFormat进行序列化分区的时候，找不到StorageDescriptor的inputFormat, > > 报错的代码在方法的org.apache.flink.connectors.hive.read.HiveTableInputFormat.createInputSplits： > > format = (InputFormat)Class.forName(sd.getInputFormat(), true, > Thread.currentThread().getContextClassLoader()).newInstance(); > > > > 而实际上，我是有引入"org.apache.hive:hive-hbase-handler:2.1.1"， > > 但很明显，flink-sql并没有利用'hive_hbase_t1'里的定义"STORED BY > 'org.apache.hadoop.hive.hbase.HBaseStorageHandler' ", > > 从而找不到正确的 INPUTFORMAT > 'org.apache.hadoop.hive.hbase.HiveHBaseTableInputFormat' 和 OUTPUTFORMAT > 'org.apache.hadoop.hive.hbase.HiveHBaseTableOutputFormat' > > > > 所以，我的问题二：flink-sql > 查询hive的hbase外部表，是否没有通过HBaseStorageHandler正确解析INPUTFORMAT和OUTPUTFORMAT > > > > > > > -- > Sent from: http://apache-flink.147419.n8.nabble.com/
