Drill storage plugin实现原理分析

2018-09-12 本文已影响24人阿里云云栖号

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实现的。

Drill Storage Plugin加载机制

DrillBit为drill的主类，drill启动时会自动加载所有有效的Storage Plugin，加载时序图如图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所示

Drill查询流程分析

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

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

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

原始SQL

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

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

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

Drill Storage Plugin开发实战

经过前面的介绍，大家对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的初始配置。

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进行注册，加载插件配置，指定优化规则等

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, 数据详细信息如下

演示SQL 1

结果1

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