用户自定义函数

大多数操作符都需要用户定义的函数。本节列出了如何指定这些函数的不同方法。我们还涵盖了累加器，它可以用来深入了解您的 Flink 应用程序。

Lambda 函数 #

在前面的例子中已经看到，所有的操作符都接受 lambda 函数来描述操作。

val data: DataSet[String] = // [...]
data.filter { _.startsWith("http://") }

val data: DataSet[Int] = // [...]
data.reduce { (i1,i2) => i1 + i2 }
// 或
data.reduce { _ + _ }

富函数(Rich functions) #

所有以 lambda 函数作为参数的变换都可以以富函数作为参数。例如，我们可以不使用:

data.map { x => x.toInt }

你可以编写:

class MyMapFunction extends RichMapFunction[String, Int] {
  def map(in: String):Int = { in.toInt }
};

并将该函数传递给 map 转换:

data.map(new MyMapFunction())

丰富的函数也可以定义为匿名类:

data.map (new RichMapFunction[String, Int] {
  def map(in: String):Int = { in.toInt }
})

丰富的函数除了提供用户定义的函数（map、reduce等）外，还提供了四个方法：open、close、getRuntimeContext 和 setRuntimeContext。这些方法可以用于为函数设置参数（参见 Passing Parameters to Functions）、创建和最终确定局部状态、访问广播变量（参见 Broadcast Variables）、访问运行时信息，如累加器和计数器（参见 Accumulators and Counters）以及迭代信息（参见 Iterations）。

累积器和计数器 #

累积器是一个简单的构造，有一个加法运算和一个最终的累积结果，在作业结束后就可以使用。

最直接的累加器是一个计数器，你可以使用 Accumulator.add(V value) 方法对它进行增量。在作业结束时，Flink 将对所有部分结果进行加总（合并）并将结果发送给客户端。累积器在调试期间或如果你快速想了解更多的数据时是很有用的。

Flink 目前有以下内置的累加器。它们每个都实现了 Accumulator 接口。

• IntCounter、LongCounter 和 DoubleCounter。请看下面一个使用计数器的例子。
• 直方图。一个离散数量的直方块的直方图实现。在内部，它只是一个从 Integer 到 Integer 的映射。你可以用它来计算值的分布，例如字数程序的每行字数分布。

如何使用累加器:

首先你必须在用户定义的转换函数中创建一个累加器对象(这里是一个计数器)，在你想使用它的地方。

private IntCounter numLines = new IntCounter();

其次，你必须注册累加器对象，通常是在富函数的 open() 方法中。在这里你还需要定义名称。

getRuntimeContext().addAccumulator("num-lines", this.numLines);

现在你可以在运算函数的任何地方使用累加器，包括在 open() 和 close() 方法中。

this.numLines.add(1);

整体结果将存储在 JobExecutionResult 对象中，该对象由执行环境的 execute() 方法返回（目前只有在执行等待作业完成的情况下才有效）。

myJobExecutionResult.getAccumulatorResult("num-lines")

所有的累加器在每个作业中共享一个命名空间。因此你可以在你的工作的不同操作函数中使用同一个累加器。Flink 会在内部合并所有同名的累加器。

关于累加器和迭代的说明。目前，累加器的结果只有在整个作业结束后才会出现。我们计划在下一次迭代中也能获得上一次迭代的结果。你可以使用 Aggregators 来计算每次迭代的统计数据，并根据这些统计数据来终止迭代。

自定义累加器:

要实现你自己的累加器，你只需要编写你的 Accumulator 接口的实现。如果你认为你的自定义累加器应该和Flink一起发布，请随时创建一个pull request。

你可以选择实现 Accumulator 或 SimpleAccumulator。

Accumulator<V,R> 是最灵活的。它为要添加的值定义了一个类型 V，为最终结果定义了一个结果类型 R。例如，对于一个直方图，V 是一个数字，R 是一个直方图。 SimpleAccumulator 适用于两种类型都相同的情况，例如计数器。

原文链接: https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/user_defined_functions.html