本文主要是介绍9、Flink 用户自定义 Functions 及 累加器详解，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！
1）用户自定义函数
1.实现接口
最基本的方法是实现提供的接口。
# 根据提供的接口创建自定义函数
class MyMapFunction implements MapFunction<String, Integer> {public Integer map(String value) { return Integer.parseInt(value); }
}# 调用创建的自定义函数
data.map(new MyMapFunction());
2.匿名类
可以将 function 当做匿名类传递。
data.map(new MapFunction<String, Integer> () {public Integer map(String value) { return Integer.parseInt(value); }
});
3.Java 8 Lambdas
Flink 在 Java API 中还支持 Java 8 Lambdas 表达式。
data.filter(s -> s.startsWith("http://"));data.reduce((i1,i2) -> i1 + i2);
4.Rich functions
所有需要用户自定义 function 的转化操作都可以将 rich function 作为参数。
class MyMapFunction implements MapFunction<String, Integer> {public Integer map(String value) { return Integer.parseInt(value); }
}
替换成
class MyMapFunction extends RichMapFunction<String, Integer> {public Integer map(String value) { return Integer.parseInt(value); }
}
并将 function 照常传递给 map transformation。
data.map(new MyMapFunction());
Rich functions 也可以定义成匿名类：
data.map (new RichMapFunction<String, Integer>() {public Integer map(String value) { return Integer.parseInt(value); }
});
2）累加器 和 计数器
1.概述
累加器是具有加法运算和最终累加结果的一种简单结构，可在作业结束后使用。
最简单的累加器是计数器: 可以使用 Accumulator.add(V value) 方法将其递增。
在作业结束时，Flink 会汇总（合并）所有部分的结果并将其发送给客户端，Flink 目前有如下内置累加器，每个都实现了累加器接口。
- IntCounter , LongCounter和 DoubleCounter
- Histogram（直方图）: 离散数量的柱状直方图实现；在内部，它只是整形到整形的映射，可以使用它来计算值的分布，例如，单词计数程序的每行单词的分布情况【详见 Metrics】。
2.使用累加器
首先，在需要使用累加器的用户自定义的转换 function 中创建一个累加器对象（此处是计数器）。
private IntCounter numLines = new IntCounter();
其次，必须在 rich function 的 open() 方法中注册累加器对象，也可以在此处定义累加器的名称。
getRuntimeContext().addAccumulator("num-lines", this.numLines);
在操作 function 中的任何位置（包括 open() 和 close() 方法中）使用累加器。
this.numLines.add(1);
最终整体结果会存储在由执行环境的 execute() 方法返回的 JobExecutionResult 对象中（当前只有等待作业完成后执行才起作用）。
myJobExecutionResult.getAccumulatorResult("num-lines");
单个作业的所有累加器共享一个命名空间，因此可以在不同的操作 function 里面使用同一个累加器；Flink 会在内部将所有具有相同名称的累加器合并起来。
关于累加器和迭代的注意事项：当前累加器的结果只有在整个作业结束后才可用；Flink 计划在下一次迭代中提供上一次的迭代结果；可以使用 聚合器 来计算每次迭代的统计信息，并基于此类统计信息来终止迭代。
3.定制累加器
自定义累加器只需要实现累加器接口，可以选择实现 Accumulator 或 SimpleAccumulator。
Accumulator 的实现十分灵活: 它定义了将要添加的值类型 V，并定义了最终的结果类型 R；例如，对于直方图，V 是一个数字且 R 是一个直方图。
SimpleAccumulator 适用于两种类型都相同的情况，例如计数器。
4.总结
1.通过调用 execute() 方法返回的 JobExecutionResult 对象获得累加器结果（只有等待作业完成后执行才起作用）。2.单个作业的所有累加器共享一个命名空间，可以在不同的操作 function 里面使用同一个累加器；Flink 会在内部将所有具有相同名称的累加器合并起来。
这篇关于9、Flink 用户自定义 Functions 及 累加器详解的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！
