Drill Storage Plugin介绍

Drill是一个交互式SQL查询引擎，官方默认支持的数据源有hive、hbase、kafka、kudu、mongo、opentsdb、jdbc等，其中jdbc storage plugin可以覆盖所有支持jdbc协议的数据源，如：mysql、oracle等关系型数据库。所有数据源的接入都是通过drill的storage plugin实现的，理论上Drill通过storage plugin机制可以支持对任何数据源进行异构查询。

Drill作为一个SQL查询引擎，它跟传统数据库有着很多相似之处，主要包括SQL Parser、SQL Validator、Query Optimizer、Data Flow Operators等几部分。如下图所示，SQL Parser阶段会把SQL语句解析为SQL查询语法树，这个阶段Storage Plugin没有介入。 从SQL Valiator阶段Storage Plugin开始介入，在这个阶段会通过Storage Plugin获取Schema信息对SQL进行校验，如判断表、字段是否存在等。Query Optimzer阶段会把SqlNode转换为PhysicalPlan，在这个过程中会通过Storage Plugin获取Planner Rule对SQL进行优化。Data Flow Operators阶段是对目标数据源进行数据读取，这部分操作是通过Storage Plugin的RecordReader实现的。

图 1

Drill Storage Plugin加载机制

DrillBit为drill的主类，drill启动时会自动加载所有有效的Storage Plugin，加载时序图如图2所示。

图 2

Plugin的注册主要是在类StoragePluginRegistryImpl中完成，插件注册主要分为以下几步。

第一步是加载classpath下所有drill-module.conf文件，这个文件配置了需要扫描的包路径，在这个包路径下接口StoragePlugin所有实现类都会被加载；第二步是校验，首先校验的是接口StoragePlugin的实现类的构造方法是否符合标准要求，构造方法参数必须为3个，且三个参数的类型必须分别为StoragePluginConfig，DrillbitContext、String。其次是校验plugin的配置是否有效，加载plugin配置，如果是首次启动，会读取classpath下bootstrap-storage-plugins.json文件,每个plugin都会对应一个这样的json文件。这个json文件最终会反序列为StoragePluginConfig实现类对象。非首次启动bootstrap-storage-plugins.json文件不会被加载。drill会以本地/tmp/drill/sys.storage_plugins目录下配置文件为准，集群模式配置信息保存在zookeeper /drill/sys.storage_plugins目录下。第三步是通过发射的方式进行插件实例化并注册。整个Plugin的注册流程如图 3所示

图 3

Drill查询流程分析

在正式介绍Drill Storage Plugin开发实战之前我们先了解下Drill的查询流程，这样有助于对Storage Plugin进行深入理解，而不是简单的依葫芦画瓢。Drill是分布式的，并且节点之间是对等的，所有drill节点都可以对外提供服务，当节点接收到sql查询请求之后，在UserWorker中会拉起一个Foreman线程来单独处理这个请求，Foreman会完成sql到物理执行计划的转换，并根据物理执行计划切分成可并行执行的Fragment，Foreman根据一定的算法把Fragment分发到本机或者其他drill节点进行执行，执行完之后会在接收初始请求的Drill节点中进行结果合并，然后返回给客户端。如图4 所示。

图 4

一条SQL到物理执行计划，会经过SqlNode（sql节点解析树）、RelNode（关系表达式）、DrillRel（drill关系表达式）、Prel（物理关系表达式）、PhysicalPlan（物理执行计划）几个步骤的转换。如图 5所示

图 5

SqlNode、RelNode、DrillRel、Prel都是树形结构，以一条简单查询druid数据源的SQL为例，SQL->SqlNode->RelNode这两个阶段只会存在节点之间的转换，不会考虑目标数据源之间的差异进行SQL优化和改写之类的动作。RelNode->DrilRel节点会进行逻辑执行计划的优化，示例中对filter进行了下推操作。DrillRel->Prel节点会进行物理执行计划的优化。 各阶段文本化之后如下所示。

原始SQL

select * from hbase.staff t1 where row_key='10000'

RelNode(关系表达式节点树)，有3个节点分别为LogicalProject、LogicalFilter、EnumerableTableScan

LogicalProject(row_key=[$0], f1=[$1]) LogicalFilter(condition=[=($0, '10000')]) EnumerableTableScan(table=[[hbase, staff]])

DrillRel（Drill关系表达式节点树),转换为drill中关系表达式节点

DrillScreenRel DrillFilterRel(condition=[=($0, '10000')]) DrillScanRel(table=[[hbase, staff]], groupscan=[HBaseGroupScan [HBaseScanSpec=HBaseScanSpec [ tableName=staff, startRow=null, stopRow=null, filter=null], columns=[`row_key`, `f1`, `**`]]])

Prel（物理关系表达式节点树），这一步应用了物理优化规则，把filter下推到scan里面了

ScreenPrel ProjectPrel(row_key=[$0], f1=[$1]) ScanPrel(groupscan=[HBaseGroupScan [HBaseScanSpec=HBaseScanSpec [ tableName=staff, startRow=10000, stopRow=10000\x00, filter=null], columns=[`row_key`, `f1`, `**`]]])

经过前面的介绍，大家对Drill Storage Plugin作用与原理应该已经有一个比较全面的了解。接下来以hbase为例详细介绍drill storage plugin开发流程。 Hbase是一个分布式列存数据库，默认是不支持SQL查询的。为了实现在Drill中用SQL对Hbase进行异构查询，需要实现一个Hbase的storage plugin。 下面以Hbase storage plugin为例介绍storage plugin的开发流程。

1、在目录contrib新建mvn模块，如： stroage-hbase

2、在新建的模块resource目录新建两个文件drill-module.conf和bootstrap-storage-plugins.json。drill-module.conf定义plugin所在的包路径，在plugin加载的时候会用到。bootstrap-storage-plugins.json文件是一些必要连接Hbase的配置。Drill首次启动时会用这个文件作为Plugin的初始配置。

{ "type": "hbase", "config": { "hbase.zookeeper.quorum": "172.168.1.100", "hbase.zookeeper.property.clientPort": "2181" }, "size.calculator.enabled": false, "enabled": true}

3、修改UserBitShared.proto文件，在CoreOpertorType对象里面新增一行HBASE_SUB_SCAN = 33，33这个数字需根据自身实际情况进行递增。修改proto文件之后需要重新编译，具体参考protocol模块下的readme.txt

4、修改distribution模块下的bin.xml, 新增org.apache.drill.contrib:drill-hbase-storage

5、代码实现，部分代码剖析如下

HbaseStoragePlugin: 相当于plugin的总入口，对scheme进行注册，加载插件配置，指定优化规则等

public class HBaseStoragePlugin extends AbstractStoragePlugin { private static final HBaseConnectionManager hbaseConnectionManager = HBaseConnectionManager.INSTANCE; private final HBaseStoragePluginConfig storeConfig; private final HBaseSchemaFactory schemaFactory; private final HBaseConnectionKey connectionKey; private final String name; //构造方法，参数必须是3个，且类型需要匹配 public HBaseStoragePlugin(HBaseStoragePluginConfig storeConfig, DrillbitContext context, String name) throws IOException { super(context, name); this.schemaFactory = new HBaseSchemaFactory(this, name); this.storeConfig = storeConfig; this.name = name; this.connectionKey = new HBaseConnectionKey(); } //注册schema  @Override public void registerSchemas(SchemaConfig schemaConfig, SchemaPlus parent) throws IOException { schemaFactory.registerSchemas(schemaConfig, parent); } //添加物理优化规则 @Override public Set<StoragePluginOptimizerRule> getPhysicalOptimizerRules(OptimizerRulesContext optimizerRulesContext) { return ImmutableSet.of(HBasePushFilterIntoScan.FILTER_ON_SCAN, HBasePushFilterIntoScan.FILTER_ON_PROJECT); }}

HbaseStoragePluginConfig: Plugin配置，参数与bootstrap-storage-plugins.json对应

HBaseSchemaFactory: Schema工厂，Schema相当于一个表元数据，包括表名、字段、以及字段类型等信息

public class HBaseSchemaFactory extends AbstractSchemaFactory { //注册schema，schema是有层级，查询时每层之间用.分隔 @Override public void registerSchemas(SchemaConfig schemaConfig, SchemaPlus parent) throws IOException { HBaseSchema schema = new HBaseSchema(getName()); SchemaPlus hPlus = parent.add(getName(), schema); schema.setHolder(hPlus); } class HBaseSchema extends AbstractSchema { HBaseSchema(String name) { super(Collections.emptyList(), name); } // hbase schema只有一层 @Override public AbstractSchema getSubSchema(String name) { return null; } @Override public Table getTable(String name) { HBaseScanSpec scanSpec = new HBaseScanSpec(name); try { return new DrillHBaseTable(getName(), plugin, scanSpec); } catch (Exception e) { // Calcite firstly looks for a table in the default schema, if the table was not found, // it looks in the root schema. // If the table does not exist, a query will fail at validation stage, // so the error should not be thrown here. logger.warn("Failure while loading table '{}' for database '{}'.", name, getName(), e.getCause()); return null; } } //调用hbase提供api，获取表信息 @Override public Set<String> getTableNames() { try(Admin admin = plugin.getConnection().getAdmin()) { HTableDescriptor[] tables = admin.listTables(); Set<String> tableNames = Sets.newHashSet(); for (HTableDescriptor table : tables) { tableNames.add(new String(table.getTableName().getNameAsString())); } return tableNames; } catch (Exception e) { logger.warn("Failure while loading table names for database '{}'.", getName(), e.getCause()); return Collections.emptySet(); } } @Override public String getTypeName() { return HBaseStoragePluginConfig.NAME; } }}public abstract class AbstractHBaseDrillTable extends DrillTable { protected HTableDescriptor tableDesc; public AbstractHBaseDrillTable(String storageEngineName, StoragePlugin plugin, Object selection) { super(storageEngineName, plugin, selection); } //字段类型转换，把hbase中的字段类型映射为SQL类型 @Override public RelDataType getRowType(RelDataTypeFactory typeFactory) { ArrayList<RelDataType> typeList = new ArrayList<>(); ArrayList<String> fieldNameList = new ArrayList<>(); fieldNameList.add(ROW_KEY); typeList.add(typeFactory.createSqlType(SqlTypeName.ANY)); Set<byte[]> families = tableDesc.getFamiliesKeys(); for (byte[] family : families) { fieldNameList.add(Bytes.toString(family)); //family映射为map结构 typeList.add(typeFactory.createMapType(typeFactory.createSqlType(SqlTypeName.VARCHAR), typeFactory.createSqlType(SqlTypeName.ANY))); } return typeFactory.createStructType(typeList, fieldNameList); }}

HbaseSubScan: 关系表达式的叶子节点，目标数据源能够识别的查询语言会在这里面定义

HbaseGroupScan: SubScan的一个超集

HbaseScanBatchCreator:根据节点泛型HbaseSubScan反射获取，获取HbaseSubScan参数并创建HbaseRecordReader对象

HbaseRecordReader:实现对目标数据源的进行记录读取,setup方法是在读取记录之前进行一些初始化工作， next方法中会调用hbase的api获取数据并放入OutputMutator对象中。

Rule: drill的优化规则，可用在逻辑计划、物理计划等优化阶段

实现一个Storage Plugin主要难点是在如何实现优化规则，where条件、聚合函数、分组、排序等是否可以下推都是由优化规则决定。下面以一个where条件下推为例介绍如何实现一个Rule。如图7所示，Filter经过下推转换为一颗 等价的查询树

图 6

Drill中优化规则很多，所有规则都是StoragePluginOptimizerRule类的子类，在进行逻辑计划和物理计划优化时并不是所有规则都会应用，只有匹配上的规则才会应用。匹配策略分两级，一级匹配比较粗略，只要查询节点树最小子树与规则类的构造放中操作类型class匹配就算匹配。如图7左边圈中部分和图8圈中部分所示。二级匹配是在matches方法，返回true才会执行onMatch方法进行关系表达式等价转换，这个方法默认是返回true，需要根据实际情况决定是否重写。在这个列子中我们进一步判断GroupScan是否是HbaseGroupScan实例，也就是说只有查询Hbase数据源的查询才会匹配这个规则。这里要说明一点的是，你在其中一个Storage Plugin中写的规则，对其他Storage Plugin来说都是可以使用的。

public abstract class HBasePushFilterIntoScan extends StoragePluginOptimizerRule { private HBasePushFilterIntoScan(RelOptRuleOperand operand, String description) { super(operand, description); } //FilterPrel.class、ScanPrel.class与图6圈中的部分匹配 public static final StoragePluginOptimizerRule FILTER_ON_SCAN =  new HBasePushFilterIntoScan(RelOptHelper.some( FilterPrel.class, RelOptHelper.any(ScanPrel.class)), "HBasePushFilterIntoScan:Filter_On_Scan") { @Override public void onMatch(RelOptRuleCall call) { final ScanPrel scan = (ScanPrel) call.rel(1); final FilterPrel filter = (FilterPrel) call.rel(0); final RexNode condition = filter.getCondition(); HBaseGroupScan groupScan = (HBaseGroupScan)scan.getGroupScan(); if (groupScan.isFilterPushedDown()) { /* * The rule can get triggered again due to the transformed "scan => filter" sequence * created by the earlier execution of this rule when we could not do a complete * conversion of Optiq Filter's condition to HBase Filter. In such cases, we rely upon * this flag to not do a re-processing of the rule on the already transformed call. */ return; } doPushFilterToScan(call, filter, null, scan, groupScan, condition); } //二级匹配 @Override public boolean matches(RelOptRuleCall call) { final ScanPrel scan = (ScanPrel) call.rel(1); //hbase数据源才会匹配 if (scan.getGroupScan() instanceof HBaseGroupScan) { return super.matches(call); } return false; } }; protected void doPushFilterToScan(final RelOptRuleCall call, final FilterPrel filter, final ProjectPrel project, final ScanPrel scan, final HBaseGroupScan groupScan, final RexNode condition) { final LogicalExpression conditionExp = DrillOptiq.toDrill(new DrillParseContext(PrelUtil.getPlannerSettings(call.getPlanner())), scan, condition); final HBaseFilterBuilder hbaseFilterBuilder = new HBaseFilterBuilder(groupScan, conditionExp); final HBaseScanSpec newScanSpec = hbaseFilterBuilder.parseTree(); if (newScanSpec == null) { return; //no filter pushdown ==> No transformation. } final HBaseGroupScan newGroupsScan = new HBaseGroupScan(groupScan.getUserName(), groupScan.getStoragePlugin(), newScanSpec, groupScan.getColumns()); newGroupsScan.setFilterPushedDown(true); //filter下推至scan中 final ScanPrel newScanPrel = ScanPrel.create(scan, filter.getTraitSet(), newGroupsScan, scan.getRowType()); // Depending on whether is a project in the middle, assign either scan or copy of project to childRel. final RelNode childRel = project == null ? newScanPrel : project.copy(project.getTraitSet(), ImmutableList.of(newScanPrel)); if (hbaseFilterBuilder.isAllExpressionsConverted()) { /* * Since we could convert the entire filter condition expression into an HBase filter, * we can eliminate the filter operator altogether. */ call.transformTo(childRel); } else { call.transformTo(filter.copy(filter.getTraitSet(), ImmutableList.of(childRel))); } }} 

效果演示

演示数据准备，表staff包含一个列簇f1, 数据详细信息如下

row_key f1:name f1:sex f1:age 10000 张三 男 18 10001 李四 男 28 10002 王五 男 38 演示SQL 1

select * from hbase.staff t1 where row_key='10000'

结果1

从上图结果可看出，同一个列簇f1是在同一个字段显示的，并且是一个json格式，列值都是经过编码的，这些结果的输出形式都是在HbaseRecordReader类中指定的，在类HbaseRecordReader中指定了row_key的输出类型为VarBinary, 列簇的输出类型为Map，Map中value为VarBinary类型。如果想要个列单独显示，SQL可以按以下方式书写。

演示SQL 2

select cast(row_key as varchar) row_key, cast(t1.f1.name as varchar) name, cast(t1.f1.sex as varchar) sex,cast(t1.f1.age as varchar) agefrom hbase.staff t1 limit 10

结果2

演示SQL 3

按列条件查询

select cast(row_key as varchar) row_key, cast(t1.f1.name as varchar) name, cast(t1.f1.sex as varchar) sex,cast(t1.f1.age as varchar) agefrom hbase.staff t1 where t1.f1.name='张三'

结果3

小结

本文对Drill SQL查询流程、Storage Plugin加载机制、以及Storage Plugin实现原理进行了分析。 希望对读者自己实现一个Storage Plugin有所帮助

作者：择云