0x1 摘要
最近业务要实时统计半小时维度的UV、PV数据,经过调研准备用Flink时间窗来实现,主要是Flink对eventTime
的支持,可以做到更精准的统计,由于第一次尝试使用Flink,所以过程中遇到不少问题,记录下来方便后续查阅。
0x2 执行计划输出JSON问题
Flink对执行计划分析提供了支持,可以通过代码将执行计划打出来,并利用官网提供的图生成工具可以方便分析,通过env.getExecutionPlan()
方法可以获取JSON格式的执行计划,将JSON字符串拷贝到http://flink.apache.org/visualizer/
网站文本框就可以查看。
但我们在项目中调用env.getExecutionPlan()
方法后报以下异常信息:
Caused by: java.lang.IllegalArgumentException: Comparison method violates its general contract!
at java.util.TimSort.mergeLo(TimSort.java:777)
at java.util.TimSort.mergeAt(TimSort.java:514)
at java.util.TimSort.mergeForceCollapse(TimSort.java:457)
at java.util.TimSort.sort(TimSort.java:254)
at java.util.Arrays.sort(Arrays.java:1512)
at java.util.ArrayList.sort(ArrayList.java:1454)
at java.util.Collections.sort(Collections.java:175)
at org.apache.flink.streaming.api.graph.JSONGenerator.getJSON(JSONGenerator.java:61)
at org.apache.flink.streaming.api.graph.StreamGraph.getStreamingPlanAsJSON(StreamGraph.java:663)
... 2 more
通过异常信息可以知道问题发生在TimSort
排序算法,意思就是比较方法违反约束,具体约束规范大家可以自行网上查阅:自反性、传递性、对称性。
我们来看一下Flink的源码,只要看JSONGenerator
类61
行就可以:
public String getJSON() throws JSONException {
JSONObject json = new JSONObject();
JSONArray nodes = new JSONArray();
json.put("nodes", nodes);
List<Integer> operatorIDs = new ArrayList<Integer>(streamGraph.getVertexIDs());
Collections.sort(operatorIDs, new Comparator<Integer>() {
@Override
public int compare(Integer o1, Integer o2) {
// put sinks at the back
if (streamGraph.getSinkIDs().contains(o1)) {
return 1;
} else if (streamGraph.getSinkIDs().contains(o2)) {
return -1;
} else {
return o1 - o2;
}
}
});
visit(nodes, operatorIDs, new HashMap<Integer, Integer>());
return json.toString();
}
从源码可以看在对操作ID做排序时Flink自己实现compare
方法,具体这个方法的实际意义不是很明白,有明白的赐教一下。后来通过网上查阅已经有人提过此issues,地址:https://issues.apache.org/jira/browse/FLINK-8498
,但状态是关闭的,也没有回复什么时候解决,但我们通过查看Flink GitHub源码发现,此处实现已经发生变更,源码地址https://github.com/apache/flink/blob/master/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/JSONGenerator.java
新实现源码: