一种基于Flink Window的实时指标统计方法

一种基于Flink Window的实时指标统计方法

作者：闻乃松

需求背景

假设有一种数据源（比如CDC binlog Events、Kafka 实时数据流等），需要实时展示时序数据的摄入数量趋势，并能查看任意时间范围内的数据分布质量（比如每个字段的数据密度、总取值数量、去重后的数据取值总量等），最小时间范围间隔为1分钟，最大范围不限制。展示样例见下图，如何在计算资源有限的情况下（不使用分布式MPP查询引擎等），在可接受的时间内给出响应。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-A6zp0oAC-1640934270369)(/Users/deepexi/Library/Application Support/typora-user-images/image-20211216190259551.png)]

解决方案

考虑到查询时间范围不限，数据摄入速率和规模不限，在可接受的时间内响应查询，必然需要提前物化查询结果。梳理指标，可以将这些指标分为两类：可物化的指标和不可物化的指标。

• 可物化的指标 指可以提前预计算的指标，包括密度分布（字段不为空的数量占总记录Events的比例）、总Events数量、最早处理时间、最晚处理时间等
• 不可物化的指标 指不能提前预计算的指标，这里主要是distinct去重值数量

先说可物化的指标，因为数据指标查询最小1分钟，因此可以使用Flink Window机制：先在内存窗口暂存1分钟数据，当时间窗口过期，触发数据指标的统计和输出。大范围时间的指标基于分钟级的指标汇总得出。总体方案如下图：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-SMAEsYhw-1640934270370)(/Users/deepexi/Library/Application Support/typora-user-images/image-20211216192041708.png)]

整个过程包括以下几个核心部分：

• 外部数据源的接入和数据解析 外部数据源支持多种，比如mysql、Kafka、S3等，数据格式也多种多样，比如JSON、CSV、AVRO等，数据解析将源格式解析成规范化的关系记录格式（含Schema）。

• 实时指标计算 基于Flink 的Window实时统计是整个流程的核心，完成分布式并行统计和结果汇总。

• 指标存储和查询 理论上每分钟1条统计结果，数据总量不算大，但是考虑到这样的实时统计作业会很多，因此一种支持可扩展的分布式存储系统显得至关重要，因为Iceberg数据湖存储框架轻量，不引入新的存储系统，是本方案设计的选项之一。另外，Iceberg支持SDK查询方式，在计算资源和查询资源有限，满足响应时间的情况下，可以在单JVM线程中运行查询。

详细设计与实现

外部数据源的接入和数据解析

这里以Kafka JSON格式数据接入为例说明：

KafkaSource kafkaSource =
        KafkaSource.builder()
                .setBootstrapServers(servers)
                .setGroupId(DataSinkIcebergJob.class.getName())
                .setTopics(topic)
                .setDeserializer(recordDeserializer)
                .setStartingOffsets(OffsetsInitializer.earliest())
                .setBounded(OffsetsInitializer.latest())
                .setProperties(properties)
                .build();

示例中按照String反序列方式将Kafka字节数据反序列化为String Json格式，设置从分区起始位置拉取，并在到达最新位置停止。

数据解析将JSON的每个Field按照所在path展平成一维关系型记录数据，比如下面的一条JSON数据，经过展平后存储在Map中的效果：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-LpfTHXqv-1640934270372)(/Users/deepexi/Library/Application Support/typora-user-images/image-20211216202022662.png)]

将原始JSON字段path用下划线连接起来，同时添加一些Kafka的元数据字段（topic,offset,partition）和额外的处理时间字段（processing），如下图所示：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-X8pUgExq-1640934270372)(/Users/deepexi/Library/Application Support/typora-user-images/image-20211216201921248.png)]

实时指标计算

实时指标计算拓扑结构如下所示，分为并行计算部分和汇总计算两部分：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Cjiba1qN-1640934270373)(/Users/deepexi/Library/Application Support/typora-user-images/image-20211216203452660.png)]

其中并行计算部分将ParseResult类型的数据流按照字段维度进行统计，每个字段分别统计最大值、最小值、不为空的记录数量、字段类型。另外分钟级的统计指标还包括整体性的指标，如本统计周期的开始时间、结束时间、成功解析的事件数量和解析失败的事件数量等。

//将原始JSON数据流转换为解析结果流
DataStream parseResultStream = sourceStream.transform("ParseResultStreamOperator", TypeInformation.of(ParseResult.class), new ParseResultStreamOperator()).rebalance();

//将解析结果流按照分钟级窗口划分
ProcessWindowFunction processWindowFunction = new StatProcessWindowsFunction();
SingleOutputStreamOperator> wndStream = parseResultStream.keyBy(pr -> pr.getProcessingTime() % 1000)
        .window(TumblingProcessingTimeWindows.of(Time.minutes(1))) //设置时间窗口
        .process(processWindowFunction);

计算结果样式：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-DNCzHJSj-1640934270374)(/Users/deepexi/Library/Application Support/typora-user-images/image-20211216204741296.png)]

图中的failEventCount表示本统计周期内的解析失败数量，statPeriodEnd表示本统计周期的开始时间（窗口开始时间），其他Map类型的字段存储本字段的统计信息，如本统计周期内的最大、最小值。

上一步骤中的并行窗口计算结果有两个维度：窗口时间和key，同一个统计窗口会输出多个key的子结果，汇总计算就是将这些子统计结果合并。这里的难点有三个：

1. 窗口计算基于Processing Time，每个Task运行在分布式环境下，无法保证系统时间的精确同步和系统处理能力一致。下游可能接收到不同窗口周期的子计算结果。
2. 同一窗口周期的子计算结果按key维度有多条记录，但数量不确定，下游不知道什么时候才可以触发合并动作。
3. 合并算子能够并行，合并不能成为影响性能的瓶颈。

我们的解决方法是将并行统计的窗口按窗口时间再次聚合后，在一个同样时间大小的窗口内合并结果，窗口时间属性也是基于处理时间的：

//将并行统计的窗口按窗口时间再次聚合后再合并结果
wndStream.keyBy(new KeySelector,String>(){
                    @Override
                    public String getKey(Map value) {
                        return ((LocalDateTime) value.get("statPeriodBegin")).toString();
                    }
                })
.window(TumblingProcessingTimeWindows.of(Time.milliseconds(winTimeMills)))
.process(mergeStatProcessWindowFunction)

下面用一张图来看，是否能解决上面提到的问题，首先基于时间窗口，保证上游的结果数据在最长1分钟内全部到达下游窗口，在达到下游窗口之前是经过时间聚合后的，这保证同一个窗口周期的计算结果不会落到下游不同窗口内，而上游不同窗口的数据即使在下游同一个窗口处理，但是因为key隔离到不同的窗口处理函数调用，所以不会结果混在一起。另外，假设TM1和TM2主机时间严格一致，因为上游到达窗口计算右边沿，触发结果计算，由于计算本身有一定时延，理论上上游第一个时间窗口的数据一定会落到下游的第二个时间窗口，依次类推，但是因为下游合并结果时，只要使用上游的时间窗口属性，就可以保证结果数据的正确性不受影响。

指标存储和查询

基于Iceberg存储指标，能够以不引入外部存储系统，免维护的方式支持可扩展，同时Iceberg表能够对计算引擎和查询引擎开放。这部分包括存储表Schema的动态创建、数据实时存储和和指标查询。

首先是TableSchema的动态创建，基于JSON解析成的Schema创建Iceberg TableSchema：

private static TableSchema getStatTableSchema(Schema schema) {
  TableSchema.Builder schemaBuilder = TableSchema.builder();
  List fieldList = schema.columns();
  for (Map.Entry entry : dataTypeByName.entrySet()) {
    schemaBuilder.field(entry.getKey(), entry.getValue());
  }
  for (Types.NestedField field : fieldList) {
    DataType dataType = DataTypes.MAP(DataTypes.STRING(), DataTypes.STRING());
    schemaBuilder.field(field.name(), dataType);
  }
  return schemaBuilder.build();
}

其次，数据实时存储基于Flink实时流：

TableLoader statTableLoader = TableLoader.fromCatalog(catalogLoader, statIdentifier);
FlinkSink.forRow(statStream, statTableSchema)
        .tableLoader(statTableLoader)
        .tableSchema(statTableSchema)
        .build();

最后是指标查询，基于SDK即可实现数据的便利，此外，SDK还支持时间范围和字段级别的过滤，借助Iceberg本身存储的统计元数据信息，查询过程还是很快的。

TableLoader statTableLoader = TableLoader.fromHadoopTable(statWarehouseDir.getAbsolutePath());
statTableLoader.open();
Table statTable = statTableLoader.loadTable();
CloseableIterable statIterable = IcebergGenerics.read(statTable)
        //.where(expressions.equal()
        .build();
statIterable.forEach(record -> {
    System.out.println(record.toString());
});

总结

本文介绍了一种基于Flink Window实现实时数据指标统计的方法，内容包括数据源的解析、实时指标统计和存储查询，解决了在大数据集和资源受限情况下，快速查询数据质量的问题。另外，内容还涉及到了存储表和源数据Schema的联动，但是因为Iceberg SDK的限制，TableSchema一开始就固定下来了，无法实现在数据解析过程中动态修改Schema。最后需要一提的是，不可预计算指标，因为本文篇幅限制，此处不再展开，但是不可预计算指标确实是一大难点，为了解决快速查询，可能需要占用更多的存储和计算资源。