openLooKeng 算子接口和执行流程

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 小助手