1 openLooKeng 算子接口

1.1 openLooKeng算子相关类

▲ 图1-1 算子相关类

openLooKeng生成物理执行计划后，真正执行计划的是一个一个的算子（即Operator）。openLooKeng中将算子抽象为Operator接口，将算子工厂抽象为OperatorFactory接口，如图1-1所示。

而具体的算子则实现相应的OperatorFactory接口和Operator接口即可。例如Limit算子，在openLooKeng中会相应的有LimitOperatorFactory和LimitOperator。

1.2 openLooKeng算子接口

OperatorFactory提供的接口如表1-1所示：

1. createOperator，创建算子，返回相应的算子实例对象；

2. noMoreOperators，不再创建算子，可以释放OperatorFactory相关的资源；

3. duplicate，在right outer join或者full outer join时用到，用于复制OperatorFactory，返回OperatorFactory实例对象。

▲表1-1 OperatorFactory接口

Operator提供的接口如表1-2所示：

1. isBlocked，当前算子是否被Block，返回ListenableFuture；
2. isFinished，当前算子处理是否结束，结束返回true，不再输出page；
3. needsInput，当前算子是否可以接收输入page，可以则返回true；
4. addInput，当前算子接收输入page，前提是当前算子的needsInput返回true；
5. getOutput，当前算子输出page，如果没有输出page则返回null；
6. finish，通知当前算子不再接收输入page，当前算子可以开始计算或者结束计算；
7. close，当前算子释放相关资源；
8. getOperatorContext，返回OperatorContext；
9. startMemoryRevoke，内存不足时，将中间数据spill to disk，实现可以参考HashAggregationOperator，返回ListenableFuture；
10. finishMemoryRevoke,startMemoryRevoke完成后调用，用于清理资源。

▲表1-2 Operator接口

2 openLooKeng 算子执行流程

openLooKeng算子的执行流程代码在Driver#processInternal()方法中，其中核心代码片段如下图所示：

翻译如下：

如果!current.isFinished() && next.isBlocked()空值 && next.needsInput() ：

     // 上游算子没有结束，下游算子没有阻塞且需要数据输入current.getOutput()

    // 上游算子输出

next.addInput()

    // 下游算子输入

如果current.isFinished() ：

   // 上游算子已结束（不再有输出），通知下游算子finish（不再有输入）next.finish()

 // 通知下游算子结束

▲图2-1 pipeline算子执行

通常，一个stage由多个pipeline组成，一个pipeline由多个operator组成，一个driver运行一个pipeline。图2-1展示的是一个pipeline内算子的执行流程。如果某一个时刻operator1被阻塞，线程并不会等operator1，而是继续往后走，即current指向operator2开始新一轮的处理。如果当前Driver没有结束且运行时间在1s内会继续从operator1开始处理。

对于同一个算子而言，其可能的接口调用顺序为：

1. isBlocked()
2. needsInput()
3. addInput()
4. getOperatorContext()
5. finish()
6. isFinished()
7. getOutput()
8. close()

---

本文作者：刘玉，转载请联系openLooKeng小助手