如何添加一个新的Operator

翻译自How to add a new Operator

---------------------------------

Java API中可以通过多种途径添加Operator

1.    在DataSet上，以已存在的Operator为基础，组合或具现化（speciallzation）而形成新的Operator

2.    设计新的自定义扩展Operator（custom extension operator）

3.    设计新的运行时Operator（runtime operator）

前两种方法实现起来较为容易且轻量级。而对于运行时Operator，有时新的设计的确需要新的运行时Operator，或者使用运行时Operator会更加高效

一、在DataSet上实现一个新的Operator

许多Operator可以通过具现化另一个Operator或是UDF两种方式实现。

最简单的例子有DataSet上诸如sum(), min(), max()等方法。这些方法仅简单地用一些预先定义的参数调用其他Operator：

 public AggregateOperator<T> sum (int field) {

 　　return this.aggregate (Aggregations.SUM, field);

 }

一些Operator可以通过多个其他Operator的组合实现，如通过map和aggregate的组合来实现count()方法。实现此功能更简单的方法是在DataSet上定义一个方法，按需调用map和reduce

 public DataSet<Long> count() {

 　　return this.map(new MapFunction<T, Long>() {

 　　　　public Long map(T value) {

 　　　　　　return 1L;

 　　　　}

 　　})

 　　.reduce(new ReduceFunction<Long>() {

 　　　　public Long reduce(Long val1, Long val1) {

 　　　　　　return val1 + val2;

 　　　　}

 　　});

 }

如果我们定义一个新的Operator的同时，不想修改DataSet类，可以以静态方法的形式定义在另一个类中，此时，count()
Operator则如下所示：

 public static <T>DataSet<Long> count(DataSet<T> data) {

 　　return data.map(...).reduce(...);

 }

1.1 更加复杂的Operator

通过具现化实现的更加复杂的例子是Java API中的Aggregation Operation，它通过GroupReduce
UDF的方法实现。

Aggregate Operation从其自己的Operator演化而来，将自己转换为Common API的GroupReduceOperatorBase。Java API的aggregation
Operator仅是一个接收聚合类型（aggregation
type）和成员位置（field
position）的构建者，它将这些信息作为参数给GroupReduce
UDF，由GroupReduce
UDF来进行聚合操作

由于操作被转换成一个 GroupReduce操作，它将在优化器和运行时环境中作为一个GroupReduceOperator出现。

二、实现一个自定义扩展Operator

DataSet提供了一个自定义Operator的方法：

DataSet<X>
runOperation(CustomUnaryOperation<T, X> operation)。接口CustomUnaryOperation通过两个方法来定义Operator。

 void setInput(DataSet<IN> inputData);

 DataSet<OUT> createResult();

VertexCentricIteration Operator即通过这种方式实现，下面是一个以此种方式实现count() Operator的例子：

 public class Counter<T> implements CustomUnaryOperation<T, Long> {

 　　private DataSet<T> input;

 　　public void setInput(DataSet<IN> inputData) { this.input = inputData; }

 　　public DataSet<Long> createResult() {

 　　　　return input.map(...).reduce(...);

 　　}

 }

该Operator调用方式如下：

 DataSet<String> lines = ...;

 DataSet<Long> count = lines.runOperation(new Counter<String>());

三、实现一个新的运行时Operator

添加一个新的Runtime Operator需要对整个技术栈做出修改，从API到运行时：

1.    Java API

2.    Common API

3.    Optimizer

4.    Runtime

我们将自底向上描述，以方法mapPartition()为例（类似map方法，只不过每个并行分区仅调用一次）

1. 运行时（RunTime）

Runtime Operator使用接口Drive实现，该接口定义了描述运行时中Operator的方法。MapDriver便是那些Operator如何工作的简单例子。

与运行时一同运行的还有MutableObjectIterator，它描述了可以重用对象的数据流，以达到减少垃圾回收的压力的目的。

mapPartition Operator的核心方法run()可能具有以下形式：

 public void run() throws Exception {

 　　final MutableObjectIterator<IN> input = this.taskContext.getInput(0);

 　　final MapPartitionFunction<IN, OUT> function = this.taskContext.getStub();

 　　final Collector<OUT> output = this.taskContext.getOutputCollector();

 　　final TypeSerializer<IN> serializer = this.taskContext.getInputSerializer(0);

 　　// we assume that the UDF takes a java.util.Iterator, so we wrap the MutableObjectIterator

 　　Iterator<IN> iterator = new MutableToRegularIteratorWrapper(input, serializer);

 　　function.mapPartition(iterator, output);

 }

为了提高运行效率，以链式（chained
version）实现一个Operator总是有好处的。链接在一起的Operator作为它们的前驱Operator，在同一个线程下运行，并且可以使用嵌套循环地调用。这会省去许多序列化/反序列化的开销，从而大大增加效率。

我们可以通过MapDriver（正常的）和ChainedMapDriver（链式变种）来学习如何实现链式Operator

2. 优化器/编译器

该部分简单讨论了添加Operator的重要步骤，有关优化器的工作原理见Optimizer。为了使优化器将新的Operator纳入其优化方案，我们需要向它提供一些信息，如下所示：

1.    DriverStrategy：要使得优化器可以访问到新的Operation，新加的Operation需要加入枚举类。枚举类入口（entry）参数定义了什么类实现了runtime operator，它的链接的版本是什么，Operator是否需要累积数据（即需要内存），以及它是否需要Comparator（用于key）。在我们的例子中，我们可以添加~~~java MAP_PARTITION(MAPPartitionDriver.class, null/*或链接的版本*/, PIPELINED, false); ~~~

2.    Cost
function:：类CostEstimator需要Operation对系统的开销的信息。这里的“开销”是指Operator的non-UDF的部分。由于我们的Operator本质上并没有这部分工作（直接将数据流传递给UDF），则此开销为0。我们通过向costOperator(…)中的switch语句中添加常量MAP_PARTITION，类似MAP常量，以标识该操作没有开销。

3.    OperatorDescription：Operator的描述类定义了优化器如何处理一个Operation。它描述了Operation需要什么样的输入数据（如有序的、分区的），和允许的优化器以全局方法来优化数据操作（data
movement）、排序、分组的方法。为了描述上述信息，我们需要描述Operator拥有什么RequestedGlobalProperties（分区操作、拷贝操作）和RequestLocalProperties（排序、分组、单一提取（uniqueness）），以及Operator如何影响已存在的GlobalProperties和LocalProperties。此外，该OperatorDescription还定义了一些支持方法，例如实例化一个候选Operator（Operator
candidate）的方法等。由于mapPartition()的功能非常简单（无需分区/分组等），它的描述类也十分简单，其他Operator则具有更加复杂的需求，如Hash Join 1，Hash Join 2，SortMerge Join。下面的示例代码解释了如何为MapPartitionOperator创建描述类：

 public DriverStrategy getStrategy() {

 　　return MAP_PARTITION;

 }

 // Instantiate the operator with the strategy over the input given in the form of the Channel

 public SingleInputPlanNode instantiate(Channel in, SingleInputNode node) {

 　　return new SingleInputPlanNode(node, "MapPartition", in, MAP_PARTITION);

 }

 // The operation accepts data with default global properties (arbitrary distribution)

 protected List<RequestedGlobalProperties> createPossibleGlobalProperties() {

 　　return Collections.singletonList(new RequestedGlobalProperties());

 }

 // The operation can accept data with any local properties. No grouping/sorting is necessary

 protected List<RequestedLocalProperties> createPossibleLocalProperties() {

 　　return Collections.singletonList(new RequestedLocalProperties());

 }

 // the operation itself does not affect the existing global properties.

 // The effect of the UDF's semantics// are evaluated separately (by interpreting the

 // semantic assertions)

 public GlobalProperties computeGlobalProperties(GlobalProperties gProps) {

 　　return gProps;

 }

 // since the operation can mess up all order, grouping, uniqueness, we cannot make any statements

 // about how local properties are preserved

 public LocalProperties computeLocalProperties(LocalProperties lProps) {

 　　return LocalProperties.EMPTY;

 }

4.    OptimizerNode：优化器节点控制着所有该方面工作，它创建了OperatorDescriptor的列表，实现了结果数据集规模的估计，并且给Operator赋予其名字。此外，它相对来说是一个较小的类，故而可以从MapNode重新拷贝

3. Common API

为了使得Operation可以用于更高级的API，需要将它添加到Common API中去。最简单的方法就是添加一个base
operator。我们以类MapOperatorBase为模板，创建类MapPartitionOperatorBase。

此外，优化器需要清楚OptimizerNode如何从OperatorBase创建一个OptimizerNode，该功能在Optimizer中的调用GraphCreatingVisitor类实现。

注意：我们仍在考虑通过统一OptimizerNode和Common API
Operator来跳过这一步骤，因为它们本质上实现的是同一个功能。Common API
Operator存在的原因仅仅是使得flink-java和flink-scala包不依赖与Optimizer。

4. Java API

创建一个Java API的方式与MapOperator的方式是一样的，其中核心方法就是translateToDataFlow(…)方法，它为Java API
operator创建了Common API
operator。

最后一步就是向类DataSet添加相关方法

 public <R> DataSet<R> mapPartition(MapPartitionFunction<T, R> function) {

 　　return new MapPartitionOperator<T, R>(this, function);

 }