当前位置：首页 > news >正文

Flink预加载分区维表，实时更新配置信息

news 2025/8/9 23:47:57

当前我们的业务场景，是基于dataStream代码，维表数据量很大，实时性要求很高，所以采用预加载分区维表模式， kafka广播流实时更新配置。

实现方案
1：job初始化时每个分区open 只加载自己那部分的配置，不用每个分区都全量加载。
2：配置实时更新，采用kafka topic传到flink job广播流中，使用ConfigBroadcastProcessFunction更新分区内的配置信息。

衡量指标

总体来讲，关联维表有三个基础的方式：实时数据库查找关联（Per-Record Reference Data Lookup）、预加载维表关联（Pre-Loading of Reference Data）和维表变更日志关联（Reference Data Change Stream），而根据实现上的优化可以衍生出多种关联方式，且这些优化还可以灵活组合产生不同效果（不过为了简单性这里不讨论同时应用多种优化的实现方式）。对于不同的关联方式，我们可以从以下 7 个关键指标来衡量（每个指标的得分将以 1-5 五档来表示）:

实现简单性: 设计是否足够简单，易于迭代和维护。

吞吐量: 性能是否足够好。

维表数据的实时性: 维度表的更新是否可以立刻对作业可见。

数据库的负载: 是否对外部数据库造成较大的负载（负载越低分越高）。

内存资源占用: 是否需要大量内存来缓存维表数据（内存占用越少分越高）。

可拓展性: 在更大规模的数据下会不会出现瓶颈。

结果确定性: 在数据延迟或者数据重放情况下，是否可以得到一致的结果。

启动预加载分区维表
对于维表比较大的情况，可以启动预加载维表基础之上增加分区功能。简单来说就是将数据流按字段进行分区，然后每个 Subtask 只需要加在对应分区范围的维表数据。值得注意的是，这里的分区方式并不是用 keyby 这种通用的 hash 分区，而是需要根据业务数据定制化分区策略，然后调用 DataStream#partitionCustom。比如按照 userId 等区间划分，0-999 划分到 subtask 1，1000-1999 划分到 subtask 2，以此类推。而在 open() 方法中，我们再根据 subtask 的 id 和总并行度来计算应该加载的维表数据范围。

在这里插入图片描述
启动预加载分区维表介绍：
通过这种分区方式，维表的大小上限理论上可以线性拓展，解决了维表大小受限于单个 TaskManager 内存的问题（现在是取决于所有 TaskManager 的内存总量），但同时给带来设计和维护分区策略的复杂性。

缓存方式：
在这里插入图片描述
之前业务场景是采用的第一种，但是配置数据量越来越大，已经不能支撑业务，所以模拟调研第三种方式，设计和维护分区策略

代码实验：
Flink设置4个并行度， 2个taskmanager

-m yarn-cluster -p 4 -yjm 1024m -ytm 2048m -ynm $application_name -ys 2

在这里插入图片描述

采用自定义Partition设计和维护分区策略，数据流和维表connect

.filter(_.nonEmpty)
.map(_.get)
.partitionCustom(new CustomPartitioner(),data => {s"${data.datas.controlPlanId}"
})
.connect(indicatorConfigBroadcastStream)
.process(new FdcIndicatorConfigBroadcastProcessFunction)
.name("FdcGenerateIndicator")
.uid("FdcGenerateIndicator")

自定义Partition分区类

import org.apache.flink.api.common.functions.Partitioner
import org.slf4j.{Logger, LoggerFactory}class CustomPartitioner extends Partitioner[String]{lazy private val logger: Logger = LoggerFactory.getLogger(classOf[CustomPartitioner])override def partition(key: String, numPartitions: Int): Int = {logger.warn("分区总数"+numPartitions)return (key.hashCode % numPartitions).abs}
}

BroadcastProcessFunction

class ConfigBroadcastProcessFunctionextends BroadcastProcessFunction[fdcWindowData, JsonNode,(ListBuffer[(ALGO, IndicatorConfig)], ListBuffer[RawData])] {lazy private val logger: Logger = LoggerFactory.getLogger(classOf[FdcIndicatorConfigBroadcastProcessFunction])// 初始化override def open(parameters: Configuration): Unit = {logger.warn(s"getIndexOfThisSubtask: ${getRuntimeContext.getIndexOfThisSubtask}")logger.warn(s"getNumberOfParallelSubtasks: ${getRuntimeContext.getNumberOfParallelSubtasks}")super.open(parameters)// 获取全局变量val p = getRuntimeContext.getExecutionConfig.getGlobalJobParameters.asInstanceOf[ParameterTool]ProjectConfig.getConfig(p)}// 数据流override def processElement(windowData: fdcWindowData, ctx: BroadcastProcessFunction[fdcWindowData,JsonNode, (ListBuffer[(ALGO, IndicatorConfig)], ListBuffer[RawData])]#ReadOnlyContext,out: Collector[(ListBuffer[(ALGO, IndicatorConfig)], ListBuffer[RawData])]): Unit = {logger.warn(s"${getRuntimeContext.getIndexOfThisSubtask}")}// 广播流override def processBroadcastElement(value: JsonNode, ctx: BroadcastProcessFunction[fdcWindowData, JsonNode, (ListBuffer[(ALGO, IndicatorConfig)], ListBuffer[RawData])]#Context,out: Collector[(ListBuffer[(ALGO, IndicatorConfig)], ListBuffer[RawData])]): Unit = {}
}

打印结果：
taskmanager1； open的时候打印信息
在这里插入图片描述
taskmanager2； open的时候打印信息

当数据流来时， processElement中的打印信息

参考：
https://blog.csdn.net/weixin_44904816/article/details/104305824
https://codeantenna.com/a/IcVVHYGUVi