flink1.12.x Process Function-ProcessFunction函数，理解watermark计算方式

作者：马育民 • • 阅读：10066

说明

通过本案例，理解 ProcessFunction 函数的功能，可以获取时间戳、当前水位线，可以更好的理解 watermark 的生成方式

提示

  • 为了演示效果 ProcessFunction 的实现类，只是打印 数据、处理时间、水位线，没有对数据做额外处理，直接将数据返回了

  • 将并行度设置为1或很小的数字。因为每个子任务在第一次处理数据时，watermark初始值都是很小的负数。
    如：我的CPU是4核8线程，如果不设置并行度，就是8个子任务同时执行，那么处理前8个数据时，watermark就是很小的负数

  • 由 flink1.12.x 水位线生成时机、分配器（Periodic Watermarks周期性分配水位线、Punctuated Watermarks定点水位线） 可知，每间隔 200ms 发出一次watermark，所以休眠200ms以上，否则打印的水位线都是负数

  • 这里为了便于理解，延迟时间为 0

  • 这里为了便于理解，没有使用 window窗口

数据文件

在工程中创建 data 目录，然后创建 2.csv 文件，内容如下：

java从入门到精通,50,李雷,1
hadoop从入门到精通,80,李雷,2
flink从入门到精通,70,韩梅梅,4
mysql从入门到精通,100,李雷,15
hbase从入门到精通,50,李雷,6
hive从入门到精通,100,韩梅梅,3
html从入门到精通,120,韩梅梅,13
hbase从入门到精通,50,李雷,7
hive从入门到精通,100,韩梅梅,8
html从入门到精通,120,韩梅梅,9

解释：

  • 第一列：书名
  • 第二列：价格
  • 第三列：买家姓名
  • 第四列：订单时间，这里为了简单好理解，采用的是 时间戳，即：当前时间 距离 1970-01-01 00:00:00 的毫秒数

java

package processfunction;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class StdProcessFunction {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        /*
         必须将并行度设置为1或很小的数字
         因为每个子任务在第一次处理数据时，watermark初始值都是很小的负数
         我的CPU是4核8线程，如果不设置并行度，就是8个子任务同时执行，那么处理前8个数据时，watermark就是很小的负数
         */
        env.setParallelism(1);
        DataStream<String> ds = env.readTextFile("data/2.csv");
        SingleOutputStreamOperator<Tuple4<String,Double,String, Long>> soso = ds.flatMap(new FlatMapFunction<String, Tuple4<String,Double,String, Long>>() {
            @Override
            public void flatMap(String value, Collector<Tuple4<String,Double,String, Long>> out) throws Exception {
                String[] arr = value.split(",");
                Double price = Double.parseDouble(arr[1]);
                Long timestamp = Long.parseLong(arr[3]) ;
                Tuple4 data = Tuple4.of(arr[0],price,arr[2],timestamp);

                out.collect(data);
                // 需要休眠一会，水位线执行没那么快，否则都是负数
                Thread.sleep(210);
            }
        });
        SingleOutputStreamOperator<Tuple4<String, Double, String, Long>> watermarksSOSO = soso
                .assignTimestampsAndWatermarks(
                        // 这里为了简单，延迟时间为 0
                        WatermarkStrategy.<Tuple4<String, Double, String, Long>>forBoundedOutOfOrderness(Duration.ZERO)
                                .withTimestampAssigner(new SerializableTimestampAssigner<Tuple4<String, Double, String, Long>>() {

                                        @Override
                                        public long extractTimestamp(Tuple4<String, Double, String, Long> element, long recordTimestamp) {
                                            return element.f3;
                                        }
                                    })
        );
        SingleOutputStreamOperator<Tuple4<String, Double, String, Long>> processSOSO = watermarksSOSO.process(new ProcessFunction<Tuple4<String, Double, String, Long>, Tuple4<String, Double, String, Long>>() {
            @Override
            public void processElement(Tuple4<String, Double, String, Long> value, Context ctx, Collector<Tuple4<String, Double, String, Long>> out) throws Exception {
                Long timestamp = ctx.timestamp(); // 处理时间
                long watermark = ctx.timerService().currentWatermark();
                int indexOfThisSubtask = getRuntimeContext().getIndexOfThisSubtask();
                String s = String.format("%s,timestamp:%s,watermark:%s,indexOfThisSubtask:%s",value, timestamp, watermark, indexOfThisSubtask);
                System.out.println(s);

                //直接将源数据
                out.collect(value);
            }
        });
//        processSOSO.print();
        env.execute();
    }
}

执行结果

(java从入门到精通,50.0,李雷,1),timestamp:1,watermark:-9223372036854775808,indexOfThisSubtask:0
(hadoop从入门到精通,80.0,李雷,2),timestamp:2,watermark:0,indexOfThisSubtask:0
(flink从入门到精通,70.0,韩梅梅,4),timestamp:4,watermark:1,indexOfThisSubtask:0
(mysql从入门到精通,100.0,李雷,15),timestamp:15,watermark:3,indexOfThisSubtask:0
(hbase从入门到精通,50.0,李雷,6),timestamp:6,watermark:14,indexOfThisSubtask:0
(hive从入门到精通,100.0,韩梅梅,3),timestamp:3,watermark:14,indexOfThisSubtask:0
(html从入门到精通,120.0,韩梅梅,13),timestamp:13,watermark:14,indexOfThisSubtask:0
(hbase从入门到精通,50.0,李雷,7),timestamp:7,watermark:14,indexOfThisSubtask:0
(hive从入门到精通,100.0,韩梅梅,8),timestamp:8,watermark:14,indexOfThisSubtask:0
(html从入门到精通,120.0,韩梅梅,9),timestamp:9,watermark:14,indexOfThisSubtask:0

解释：

  • 收到第一个数据 java从入门到精通,50,李雷,1，watermark是负数，即：初始值
  • 收到第二个数据 hadoop从入门到精通,80,李雷,2，watermark才开始计算，是 0 ，计算过程：从 flink1.12.x watermark 源码分析 的 output.emitWatermark(new Watermark(maxTimestamp - outOfOrdernessMillis - 1)); 可知，maxTimestamp（上一个数据的时间戳，即：1） - outOfOrdernessMillis（延迟时间，这里设置为0） - 1 就是 1 - 0 - 1
  • 收到第三个数据 flink从入门到精通,70,韩梅梅,4，watermark是 1，计算过程：maxTimestamp（上一个数据的时间戳，即：2） - outOfOrdernessMillis（延迟时间，这里设置为0） - 1 就是 2 - 0 - 1 = 1
  • 收到第四个数据 mysql从入门到精通,100,李雷,15，watermark是 3，计算过程：maxTimestamp（上一个数据的时间戳，即：4） - outOfOrdernessMillis（延迟时间，这里设置为0） - 1 就是 4 - 0 - 1 = 3
  • 收到第五个数据 hbase从入门到精通,50,李雷,6，watermark是 14，计算过程：maxTimestamp（上一个数据的时间戳，即：15） - outOfOrdernessMillis（延迟时间，这里设置为0） - 1 就是 15 - 0 - 1 = 14
  • 收到后面的数据，从 flink1.12.x watermark 源码分析 的 maxTimestamp = Math.max(maxTimestamp, eventTimestamp); 可知，将 最大时间戳 与 当前数据的 事件时间 做比较，取出最大值。由于当前最大的时间就是 15，所以一直都是用 15 计算 watermark，所以一直是 14

原文出处：/show_1IX4FhqW81Mb.html

吉ICP备15003043号