flink源码笔记

JSONObject

JSONArray

JSON.parse

StreamTransformation

operation

shuffle

broadcast

reblance

rescale

global

注册表

select

分支主题

parse

StreamGraphGenerator.generate(this, transformations);

StreamGraph

StreamingJobGraphGenerator.createJobGraph(this, jobID);

start-cluster.sh

handleMessage

sum

max

分支主题

分支主题

分支主题

DefaultOperatorStateBackend

HeapKeyedStateBackend

分支主题

分支主题

分支主题

Keyed State:ValueState<T>:保持一个可以更新和获取的值（每个Key一个value），可以用来update(T)更新，用来T value()获取。ListState<T>: 保持一个值的列表，用add(T) 或者 addAll(List<T>)来添加，用Iterable<T> get()来获取。ReducingState<T>: 保持一个值，这个值是状态的很多值的聚合结果，接口和ListState类似，但是可以用相应的ReduceFunction来聚合。AggregatingState<IN, OUT>:保持很多值的聚合结果的单一值，与ReducingState相比，不同点在于聚合类型可以和元素类型不同，提供AggregateFunction来实现聚合。FoldingState<T, ACC>: 与AggregatingState类似，除了使用FoldFunction进行聚合。MapState<UK, UV>: 保持一组映射，可以将kv放进这个状态，使用put(UK, UV) or putAll(Map<UK, UV>)添加，或者使用get(UK)获取。

分支主题

调用了sourceContext.collectWithTimestamp(record, timestamp);这个方法

分支主题

相当于增加了一个operator transformation

StreamInputProcessor

构造方法

也是OneInputStreamOperator

例子connect(stream2).process(new CoProcessFunction())

JoinedStreasm.where(keySelector).equalTo(keySelector).window.apply()

全量 apply

增量 reduce

分支主题

当Buffer满了之后自己就会被阻塞在向localBufferPool申请内存块的过程中

ui

api

载入

传递

特点

DataStream<T> input = ...;=// 基于event-time的滚动窗口input .keyBy(<key 选择器>) .window(TumblingEventTimeWindows.of(Time.seconds(5))) .aggregate(new .aggregate(new AggregateFunction<String,LongCounter,Long>(){ @Override public LongCounter createAccumulator() { return new LongCounter(); //定义一个累加器的类型， } @Override public void add(String value, LongCounter accumulator) { accumulator.add(Long.valueof(value)) } @Override public Long getResult(LongCounter accumulator) { return accumulator.getLocalValue(); } @Override public LongCounter merge(LongCounter a, LongCounter b) { a.merge(b); return a; }},new WindowFunction<Long, String, String, TimeWindow>() { @Override public void apply(String key, TimeWindow window, Iterable<Long> input, Collector<Long> out) throws Exception { out.collect(key+":" + input.iterator.next()) }});

上面的代码的意思是，首先AggregateFunction负责进行每来一条数据就进行计算的逻辑。内部使用了flink的state编程模型。其中LongCounter是flink自带的累加器。方法：createAccumulator，定义了存在state中的数据类型，数据每来一条都将调用add方法进行计算，然后调用merge来合并数据存回state，最后窗口触发的时候调用getResult方法输出结果。最后的结果进入WindowFunction，这时的WindowFunction里面Iterable<Long> input，其实只包含一个元素。就是AggregateFunction的计算结果。考虑这样的场景，我们定义一个大小为5分钟，甚至是1h的窗口，我们期望的结果是，5m的聚合结果或者是1h的聚合结果。但通常情况下，我不想等待这么久才得到结果怎么办？也就是说我想要一个增量的结果。flink提供一个名为ContinuousEventTimeTrigger的窗口触发器,该触发器是基于事件时间的按照指定时间间隔持续触发的触发器，它的首次触发取决于Watermark。首次触发的判断位于onElement中，它注册下一次（也是首次）触发eventTime 定时器的时机，然后将其first状态标识为false。概念我翻译一下，就是你可以定义一个间隔的触发时间，该触发器周期的触发窗口计算，输出增量结果。使用方法如下：

默认情况下，当watermark通过end-of-window之后，再有之前的数据到达时，这些数据会被删除。为了避免有些迟到的数据被删除，因此产生了allowedLateness的概念。默认情况下，如果不指定allowedLateness，其值是0，简单来讲，allowedLateness就是针对event time而言，对于watermark超过end-of-window之后，还允许有一段时间（也是以event time来衡量）来等待之前的数据到达，以便再次处理这些数据。这里提一下window的触发条件：watermark < end-of-window + allowedLateness 之内到达的数据，都会被再次触发窗口的计算。注意只要晚到的数据，且满足条件是会持续不断触发窗口计算的！！

DataStream<T> input = ...;=// 基于event-time的滚动窗口input.assignTimestampsAndWatermarks(new AssignWatermark(Time.seconds(60))) .keyBy(<key 选择器>) .window(TumblingEventTimeWindows.of(Time.seconds(5))) .allowedLateness(<time>); .<windowed transformation>(<窗口函数>);

注意：对于trigger是默认的EventTimeTrigger的情况下，allowedLateness会再次触发窗口的计算，而之前触发的数据，会buffer起来，直到watermark超过end-of-window + allowedLateness（）的时间，窗口的数据及元数据信息才会被删除。当数据还是晚于 allowedLateness怎么办？如果你不想丢弃这部分数据你可以使用Flink的side output功能时，你可以获取到因为晚到被丢弃的元素流。之后你可以针对此流做进一步处理。首先你需要在windowed流上通过sideOutputLateData(OutputTag)指明你想要获取晚到的元素，然后你就能在windowed operation的结果中获取到side-output流：

final OutputTag<T> lateOutputTag = new OutputTag<T>("late-data"){};DataStream<T> input = ...;DataStream<T> result = input .keyBy(<key 选择器>) .window(<窗口 分配器>) .allowedLateness(<time>) .sideOutputLateData(lateOutputTag) .<windowed transformation>(<窗口函数>);DataStream<T> lateStream = result.getSideOutput(lateOutputTag);

其中，allowedLateness只针对Event Time有效；allowedLateness可用于TumblingEventTimeWindow、SlidingEventTimeWindow以及EventTimeSessionWindows，要注意这可能使得窗口再次被触发，相当于对前一次窗口的窗口的修正（累加计算或者累加撤回计算);要注意再次触发窗口时，UDF中的状态值的处理，要考虑state在计算时的去重问题。最后要注意的问题，就是sink的问题，由于同一个key的同一个window可能被sink多次，因此sink的数据库要能够接收此类数据。

默认情况下，元素不会逐个传输到每个算子的，如果来一个传输一个势必导致不必要的网络流量，flink缓存待发送到网络的元素。虽然这种方法对于优化吞吐量有好处，但是当输入流不够快时或者输入流过小的时候，它可能会导致延迟问题(因为要等待buffer慢的时候才会发送)。要控制吞吐量和延迟，你可以在execution environment（或单个operator）上使用env.setBufferTimeout(timeoutMillis)来设置缓冲区填满的最大等待时间。如果超过该最大等待时间，即使缓冲区未满，也会被自动发送出去。该最大等待时间默认值为100 ms。除非你的应用对数据的实时性精确到毫秒，你可以调整这个参数。一般默认就好。

LocalStreamEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();env.setBufferTimeout(timeoutMillis); // 设置这个参数env.generateSequence(1,10).map(new MyMapper()).setBufferTimeout(timeoutMillis); // 设置这个参数

为了最大化吞吐量，可以设置setBufferTimeout(-1)，这样就没有了超时机制，缓冲区只有在满时才会发送出去。为了最小化延迟，可以把超时设置为接近0的值（例如5或10 ms）。应避免将该超时设置为0，因为这样可能导致性能严重下降。

private static StreamExecutionEnvironment getExecutionEnvironment(ParameterTool params) throws IOException { //int parallelism = params.getInt("parallelism", 1); //获取流处理的执行环境 final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); //使用rockdb,并使用增量检查点的方式 env.setStateBackend(new RocksDBStateBackend("hdfs://hadoop-meituan/user/hadoop-launcher/flink/checkpoints", true)); //设置每6秒触发一次检查点，检查点是异步的，当腰读取kafka的时候，这个时间间隔6-10s为宜。 env.enableCheckpointing(6000, CheckpointingMode.EXACTLY_ONCE); //设置并行度 //env.setParallelism(parallelism); env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); //这是检查点的最大个数 env.getCheckpointConfig().setMaxConcurrentCheckpoints(1); //设置operate的缓冲时间，The default value for this timeout is 100 ms. env.setBufferTimeout(100); //设置job的全局参数 env.getConfig().setGlobalJobParameters(params); return env;}

public class PreAggregateFunction implements AggregateFunction<MultiValueMetric, Map<String, Accumulator>, Map<String, List<Double>>> { @Override public Map<String, Accumulator> createAccumulator() { return new HashMap<>(); } @Override public void add(MultiValueMetric strData, Map<String, Accumulator> accumulatorMap) { String metric = strData.getMetric(); String type = strData.getType(); String key = metric + "," + type + "," + strData.getTime_granularity() +"," +strData.getTags() +"," + strData.getAppName(); //原始数据中的值 List<Double> currentValues = strData.getValues(); if (accumulatorMap.containsKey(key)) { Accumulator valueMap = accumulatorMap.get(key); for (Double currentValue : currentValues) { valueMap.add(currentValue); } } else { Accumulator accumulator = null; if (type.equals("counter")) { accumulator = new DoubleCounter(); } else if (type.equals("max")) { accumulator = new DoubleMaximum(); } else if (type.equals("mix")) { accumulator = new DoubleMinimum(); } else if (type.equals("avg")) { accumulator = new AverageAccumulator(); } else if (type.equals("timer")) { accumulator = new ListAccumulator<>(); } if (accumulator != null) { for (Double currentValue : currentValues) { accumulator.add(currentValue); } accumulatorMap.put(key, accumulator); } } } @Override public Map<String, List<Double>> getResult(Map<String, Accumulator> accumulator) { Map<String, List<Double>> result = new HashMap<>(); for (Map.Entry<String, Accumulator> entry : accumulator.entrySet()) { List<Double> values = new ArrayList<>(); String key = entry.getKey(); String type = key.split(",")[1]; if ((type.equals("timer")) && entry.getValue() instanceof ListAccumulator) { values = (ArrayList) entry.getValue().getLocalValue(); result.put(key, values); } else { values.add((double) entry.getValue().getLocalValue()); result.put(key, values); } } return result; } /** * Merge two collections of accumulators. The second will be merged into the first. * * @param target The collection of accumulators that will be updated * @param toMerge The collection of accumulators that will be merged into the other */ @Override public Map<String, Accumulator> merge(Map<String, Accumulator> target, Map<String, Accumulator> toMerge) { for (Map.Entry<String, Accumulator> otherEntry : toMerge.entrySet()) { Accumulator ownAccumulator = target.get(otherEntry.getKey()); if (ownAccumulator == null) { // Create initial counter (copy!) target.put(otherEntry.getKey(), otherEntry.getValue().clone()); } else { // Both should have the same type AccumulatorHelper.compareAccumulatorTypes(otherEntry.getKey(), ownAccumulator.getClass(), otherEntry.getValue().getClass()); // Merge target counter with other counter ownAccumulator.merge(otherEntry.getValue()); } } return target; }}