用户定义函数
— 焉知非鱼User Defined Functions
用户自定义函数
大多数操作符都需要用户定义的函数。本节列出了如何指定这些函数的不同方法。我们还涵盖了累加器,它可以用来深入了解您的 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