Apache Drill原理持续学习

DrillBit介绍

• DrillBit Drill的服务端控制进程,管理ServiceEngine,WorkerManager
• ServiceEngine 服务引擎,管理UserServer,Controller
• BootStrapContext 启动DrillBit的上下文,包括配置信息,度量注册
• DrillbitContext DrillBit工作时候的上下文
• Controller 不同DrillBit节点的通信
• ControllServer 不同节点间消息传输,连接等的RPC服务端
• DataServer 负责数据交互的RPC服务端

---

• UserServer 用户操作的服务端,会将工作交给UserWorker,它需要一个UserWorker
• UserWorker 用户操作的(工人), 通过WorkerBee构成

---

• WorkerManager 工人管理员,负责选择一个工人来工作

• WorkerBee 工蜂, 真正干活的

• Foreman 包工头,监工.由UserWorker创建出来. 因为UserWorker底层是WorkerBee,所以会将WorkerBee和Foreman关联起来

UserServer+UserWorker

UserServer处理RUN_QUERY_VALUE客户端的查询请求,会将任务分派给UserWorker处理, 由worker提交工作:

• 显然worker要在构造UserServer的时候也一起构造出来, 这样在收到任务的时候, 确保立即有工人接手这份工作；
• UserServer的构造在ServiceEngine,而服务引擎是由DrillBit创建的.
• UserWorker是由WorkerManager管理的, 而WorkerManager也是由DrillBit创建的.
  所以启动DrillBit服务后,参与计算的角色都已经准备好了.

• 1. UserServer通过RPC发送给客户端queryId, 表示客户端这一次的查询标识. 服务端已经接受了这次查询.
• 2. 但是服务端还没有开始执行这个查询任务, 后续如果客户端需要查询结果, 可以凭这个QueryId, 就可以向服务端要数据结果.
• 3. WorkerBee从名字上看是工作的蜜蜂, 工蜂一直默默无闻地工作. 它为母蜂Foreman服务.
• 现在我们由UserWorker创建了一个Foreman. 工蜂把它加进来.

public class UserWorker{
  private final WorkerBee bee;

  public QueryId submitWork(UserClientConnection connection, RunQuery query) {
    ThreadLocalRandom r = ThreadLocalRandom.current();
    // create a new queryid where the first four bytes are a growing time (each new value comes earlier in sequence).  Last 12 bytes are random.
    long time = (int) (System.currentTimeMillis()/1000);
    long p1 = ((Integer.MAX_VALUE - time) << 32) + r.nextInt();
    long p2 = r.nextLong();
    QueryId id = QueryId.newBuilder().setPart1(p1).setPart2(p2).build();
    incrementer.increment(connection.getSession());
    Foreman foreman = new Foreman(bee, bee.getContext(), connection, id, query);
    bee.addNewForeman(foreman);
    return id;

Foreman

• Foreman是由WorkerBee(工蜂)主动创建的；
• Foreman作为一个独立进程,不是自己启动,而是要由工人来启动；

• Foreman负责管理一次查询的所有fragments, Foreman会作为根节点/驱动节点；
• 设置Foreman, 但是并没有初始化任何的执行；
• Foreman的run方法根据RunQuery的类型执行不同的方法,比如SQL类型,则要负责将SQL语句通过Calcite解析成逻辑计划,生成物理计划,最后运行物理计划.

一次Query的生命周期

Drill 大神的这篇链接打不开了：http://tnachen.wordpress.com/2013/11/05/lifetime-of-a-query-in-drill-alpha-release/
只能看下别人翻译的文章；

Root fragment 会被提交给DrillBit上面的Worker manager 。中间fragment 保存在zookeeper中，所有的leaf fragment会直接通过BitCom（RPC层次的东西，协议是Protobuf ）发送给其他DrillBits。
　　 Worker Manager一旦接受到Root Fragment ，就会运行这个plan，并且包含一个Screen Store ，用来阻塞，并且等待返回的数据。如果该plan需要另外多个DrillBit，这些DrillBit组成一个wire，Worker Manager也同时会包含一个exchange operator，该exchange operator启动了一个Receiver用以等待wire中的数据。
　　在wire中，leaf fragment被发送给其他DrillBit并且执行。这些leaf fragment也会被转换成为由physical operator 组成的DAG。每一个Physical operator都会利用一个Pull 类型的消息机制，从树的底部开始，operator会从他的parent operator中pull 记录信息，而他的parent operator 则返回一个Outcome status消息。Operator被设计成能够处理每一个可能outcome status（STOP，OK，OK_WITH_NEW_SCHEMA,NONE）,因为Drill支持动态schema，也就是说Drill允许在同一个数据集中schema发生变化，所以Drill要能够处理当schema发生变化时的情况，可以参考columnnar storage（http://the-paper-trail.org/blog/columnar-storage/），Drill同时实现了他自己的内存数据结构，我们称之为ValueVector，ValueVector是一组byte集合，代表了一个column内的数据。在每一个Physical operator pull的消息中会返回一个RecordBatch，一个RecordBatch中包含一个或者多个ValueVector。（一个column会包含一个或者多个ValueVector，同时还有schema信息)。

在文章的例子中（图中），leaf fragment的顶端是这个Scan operator，该Scan operator被设置成为查询Parquet file，并且通过Parquet storage engine运行。这个Storage engine的作用就是从数据源中拉取数据，把数据转换为ValueVector，然后将这些ValueVector作为RecordBatch传递回他的child 。
　　最终，所有的Leaf fragment将会接管这些batch数据，通过Sender operator 发送给中间DrillBit。
　　中间fragment 一旦第一次接受到一个RecordBatch，会从HazleCast中通过RecordBatch中保留的fragment id查询相应的fragment，并且设置Receiver以及必要的physical operator来继续在DrillBit中进行处理计算。
　　中间Fragment包含一个Filtering operator，在这个Filtering operator内部，一旦他接收到一个RecordBatch，他就会查找新的schema，并且将schema传递给CodeGeneration,同时还会传递一个特殊定义的filter expression，type information，借此产生一段特殊的code来完成filter 操作。通过设计成避免casting，运行轻量级的loop，以及进行prefetching，来减少方法的调用，这种方式在Hive的新vectoried query engine（通过Stinger initiative）以及impala中很普遍。
　　中间fragment最终会议batch为单元，一次发送一个batch给Root DrillBit,在Root DrillBit中会由Screen operator 来接收相关数据，并且返回给client。
　　DrillClient接收RecordBatch，简单讲ValueVector转换成Rows并且显示给client。
　　
Query查询的入口是：Foreman线程的run方法中的queryRequest方法；客户端输入的查询,会通过RPC在Foreman上执行
protobuffer文件的定义在drill-protocol/src/main/protobuf下,比如User.proto对应了UserProtos。

关键看下run()上面的注释.

• 什么时候被调用: 在查询建立起来的时候；
• 以什么样的方式调用: 执行线程池；
• 功能是什么: 完成远程执行；
  注意这个方法的结束并不代表查询生命周期的Foreman角色的结束.

Called by execution pool to do query setup, and kick off remote execution.
Note that completion of this function is not the end of the Foreman's role in the query's lifecycle.

https://tnachen.wordpress.com/2013/11/05/lifetime-of-a-query-in-drill-...
http://yangyoupeng-cn-fujitsu-com.iteye.com/blog/1974556

Client

• 从Client端，查询语句被提交给sqlline，该sqlline只是一个简单用java编写的console，他能够和jdbc driver进行沟通，将SELET语句传递给DrillOptiq;
• Drill利用Optiq来解析query并且进行plan。Optiq提供可插入式的转换规则，利用这些规则来讲SQL语句的各个部分映射成为你想要的对象。Optiq内置一个查询优化器，Drill的开发者们利用这个优化器挑选出SQL语句执行的最佳顺序，而这个挑选过程不需要任何查询的统计数据。Drill 开发者们自定义很多Optiq 规则完成对SQL 操作符的对象映射，这里面包括（WHRER LIMIT等等），每一条规则都将SQL query中查询的某个操作符转换为drill能够识别的logical operator。
• Drill logical plan的唯一的目标就是Drill的数据流的工作流程，而没有做任何的优化，和分布式计算的分发等工作;
• 一旦client产生了logical plan，那么他会查询其中一个已经配置好的DrillBit的host/port的信息;
• 然后将logical plan传递给DrillBit(这个接收查询的DrillBit就是Foreman)；

参考

• http://zqhxuyuan.github.io/2015/07/13/2015-07-13-drill-phyplan/?spm=ata.13261165.0.0.1fa75075AfOeVu

• https://zqhxuyuan.github.io/2015/07/12/2015-07-12-drill-rpc/

• https://www.iteye.com/blog/yangyoupeng-cn-fujitsu-com-1974556

• Drill提供的Logical/Physical Plan生成文档：https://docs.google.com/document/d/1QTL8warUYS2KjldQrGUse7zp8eA72VKtLOHwfXy6c7I/edit#

• http://www.eenot.com/thread-107576-1-1.html

• https://segmentfault.com/a/1190000004341483