Flink | 六十八、FlinkCEP之处理迟到数据

作者: 雨中散步撒哈拉
  • [一、数据迟到]
  • [二、水位线和暂存]
  • [三、侧输出解决]

一、数据迟到

CEP 主要处理的是先后发生的一组复杂事件,所以事件的顺序非常关键。前面已经说过, 事件先后顺序的具体定义与时间语义有关。如果是处理时间语义,那比较简单,只要按照数据处理的系统时间算就可以了;而如果是事件时间语义,需要按照事件自身的时间戳来排序。这就有可能出现时间戳大的事件先到、时间戳小的事件后到的现象,也就是所谓的"乱序数据" 或"迟到数据"。

二、水位线和暂存

在 Flink CEP 中沿用了通过设置水位线(watermark)延迟来处理乱序数据的做法。当一个事件到来时,并不会立即做检测匹配处理,而是先放入一个缓冲区(buffer)。缓冲区内的数据,会按照时间戳由小到大排序;当一个水位线到来时,就会将缓冲区中所有时间戳小于水位线的事件依次取出,进行检测匹配。这样就保证了匹配事件的顺序和事件时间的进展一致,处理的顺序就一定是正确的。这里水位线的延迟时间,也就是事件在缓冲区等待的最大时间。

三、侧输出解决

这样又会带来另一个问题:水位线延迟时间不可能保证将所有乱序数据完美包括进来,总会有一些事件延迟比较大,以至于等它到来的时候水位线早已超过了它的时间戳。这时之前的数据都已处理完毕,这样的"迟到数据"就只能被直接丢弃了------这与窗口对迟到数据的默认处理一致。

我们自然想到,如果不希望迟到数据丢掉,应该也可以借鉴窗口的做法。Flink CEP 同样提供了将迟到事件输出到侧输出流的方式: 我们可以基于 PatternStream 直接调用.sideOutputLateData()方法,传入一个 OutputTag,将迟到数据放入侧输出流另行处理。代码中调用方式如下:

PatternStream<Event> patternStream = CEP.pattern(input, pattern);
// 定义一个侧输出流的标签
OutputTag<String> lateDataOutputTag = new OutputTag<String>("late-data") {
};
SingleOutputStreamOperator<ComplexEvent> result = patternStream
        .sideOutputLateData(lateDataOutputTag)    // 将迟到数据输出到侧输出流
        .select(
        // 处理正常匹配数据
                new PatternSelectFunction<Event, ComplexEvent>() {...
                }
        );
// 从结果中提取侧输出流
DataStream<String> lateData = result.getSideOutput(lateDataOutputTag);

可以看到,整个处理流程与窗口非常相似。经处理匹配数据得到结果数据流之后,可以调用.getSideOutput()方法来提取侧输出流,捕获迟到数据进行额外处理。

文章作者: 雨中散步撒哈拉

文章列表

更多推荐

更多
  • Apache StreamPark-用户、团队、角色以及成员管理 用户管理 ADMIN 创建或修改用户时可以指定用户类型,用户类型有 ADMIN 和 USER 两种。ADMIN 表示系统管理员,即:StreamPark 的超级管理员,有 StreamPark 管理页面以及各个团队的所有权限。USER ...
  • Apache StreamPark-Docker 快速使用教程 使用 Docker 完成StreamPark的部署。 前置条件 Docker 1.13.1+ Docker Compose 1.28.0+ 安装docker 使用 docker 启动服务,使用 docker-compose ...
  • Apache StreamPark-快速开始 本章节看看如果用 streampark-console 快速部署运行一个作业, 用 streampark 开发的项目都做了很好的支持,下面我们使用 streampark-quickstart 来快速开启 streampark-console...
  • Apache StreamPark-变量管理 背景介绍 ...
  • Apache StreamPark-LDAP 快速使用教程 LDAP简介 LDAP(Light Directory Access Portocol),它是基于X.500标准的轻量级目录访问协议。 ...
  • Apache StreamPark-安装部署 StreamPark 总体组件栈架构如下, 由 streampark-core 和 streampark-console 两个大的部分组成, 定位是一个综合实时数据平台,流式数仓平台, 低代码 ( Low Code )...
  • Apache StreamPark FlinkSQL-数据类型 Flink SQL有一组丰富的本地数据类型可供用户使用。 数据类型描述表生态系统中值的逻辑类型,它可用于声明操作的输入和/或输出类型。 ...
  • Apache StreamPark FlinkSQL-查询配置 任务执行配置 以下选项可用于调优查询执行的性能。table.exec.async-lookup,table.exec.deduplicate,以下配置可用于调整查询优化器,以获得更好的执行计划。table.optimizer.agg-...
  • Apache StreamPark FlinkSQL-性能调整 SQL是数据分析中使用最广泛的语言。Flink的Table API和SQL使用户可以用更少的时间和精力去开发高效的流分析应用程序。 此外,Flink Table API和SQL都被进行了有效的优化,集成了大量查询优化和算子优化实现。
  • Apache StreamPark FlinkSQL-读写hive Apache Hive 已经成为了数据仓库生态系统中的核心。它不仅仅是一个用于大数据分析和ETL场景的SQL引擎。 Flink 与 Hive 的集成包含两个层。一是利用了 Hive 的 MetaStore 作为持久化的 ...
  • 近期文章

    更多
    文章目录

      推荐作者

      更多