开发者社区 > 博文 > Flink在诸葛智享库存预警系统的实践
分享

  • 打开微信扫码分享

  • 点击前往QQ分享

  • 点击前往微博分享

  • 点击复制链接

Flink在诸葛智享库存预警系统的实践

  • castechelon
  • 2021-08-30
  • IP归属：北京
  • 30720浏览

    作者：樊思国、王天阳、饶星、张志军

    一、前言

    随着数字化浪潮的到来，越来越多的电商企业利用数字化技术进行供应链管理。在促销、秒杀等场景下，库存往往会出现较为大的波动，如不及时处理，往往会带来无货可售的风险，从而带来一定的经济损失。因此，基于配置规则实现的实时库存告警平台能够让商家实时掌控库存动态，并作出应对策略。

    Apache Flink作为一种流式计算引擎，近年来在业界具有广泛的应用场景。京东诸葛智享系统利用Flink实现了基于用户自定义规则的实时告警系统，下面从业务和技术两方面进行介绍。

    二、业务介绍

    诸葛·智享智慧供应链产品是一款为京东第三方入仓商家提供库存优化管理决策支持的精细化、智能化、自动化库存决策产品。它由京东自主研发，结合京东过去12年的零售经验积累，为商家解决运营过程中诸多核心痛点，如仓储周转天数高导致仓租成本过高亏损，畅销品缺货错过销售机会，滞销品堆积占用公司现金，海量商品补货精力无法顾及，高度依赖人员经验使用传统工具，缺少系统支持等，从而帮助商家提升工作效率。是目前市场上功能模块比较全面的一款库存决策支持及库存管理工具。

    ​ 诸葛·智享中的库存预警平台主要为商家解决无货、低库存预警的功能，由于商家的SKU的数量非常多，商家不能实时关注到全部库存的情况，导致商品卖无货后，造成很多跑单的情况，给商家带来了大量的损失，每年由于无货导致C取消订单量高达数百万元，由于无货导致发货不及时的C投诉大量存在。库存预警平台旨在帮助商家全面管理库存，及时了解库存状态，降低人力成本，提升库存管理效率；同时避免无货导致销售机会的丧失、发货不及时，提升客户购物体验及商家的销售机会。

    三、技术选型

    流处理系统的技术架构发展历程大致可以分为批处理架构、Lambda架构和Kappa架构三个阶段：

    批处理架构。严格来讲，批处理可以认为是一种具有边界的流数据，这个边界可以是分钟、小时、天等时间维度，也可以是按照文件路径等逻辑存储维度等进行划分。将数据按照特定（一般是时间）维度积攒到一定程度进行触发计算，一次性输入所需数据，中间经过一个或多个步骤的迭代计算，将最终结果输出到外部存储引擎。典型的批处理实现如MapReduce、Spark等。批处理特点决定了往往有比较高的计算延时，不能满足实时输出结果的业务需要，另外，同外部系统的交互性较差。

    Lambda架构。为满足低延时场景的需要，在批处理的基础上衍生出了一种过渡期的Lambda架构，该架构的特点根据计算时延长短，分为实时计算和离线计算两部分。实时部分负责对实时到来的数据进行计算，离线部分负责历史数据的计算。一般情况下利用Storm实现实时部分计算、Hadoop实现离线部分计算。Lambda架构的缺点是需要维护实时和离线两套代码且需要保持计算结果相同，给系统运维带来很大挑战。另外Lambda架构中的实时部分往往只能得到近似的计算结果，不能满足Exactly-once语义要求。

    Kappa架构。随着流式计算技术的进步，出现了一种批流一体的架构模式，即Kappa架构。该架构特点是延时低，根据事件到达进行实时计算。同时一套架构支持批处理，只需要一套代码即可完成批流的计算需求，维护简单。支持Exactly-once语义。

    综上，我们可以从时延、维护成本、交互性及Exactly-once语义支持上对三种架构模式进行比较，如下表：

    当流处理架构经过以上三个阶段的发展后，流处理又存在两种模式的实现方式：微批模式和纯流式。前者使用批来对流进行模拟，认为流是一种更细粒度的批，在运行时通过制定批处理的时间，每次运行的job会处理一个批次中的数据，典型的实现如Spark Streaming。而纯流式系统是基于事件驱动的，事件可以理解为消息，事件驱动的应用是一种有状态的应用，它会从一个或多个流中摄入事件，通过触发计算更新内部状态，或外部动作对注入的事件作出反应。典型的实现如Flink。

    考虑到诸葛智享的库存告警模块需要根据用户的规则配置，结合库存变动消息（事件）以及无货PV消息（事件），以及仓网关系等状态进行实时告警的需要，我们选择Flink作为流计算引擎。

    四、Flink的应用

    4.1、Flink基本概念

    4.1.1、Flink运行时架构

    4.1.1.1、Flink运行时的组件

    Flink运行时架构主要包括四个不同的组件，它们会在运行流处理应用程序时协同工作：

    • 作业管理器（JobManager）

    控制一个应用程序执行的主进程，也就是说，每个应用程序都会被一个不同的JobManager所控制执行。

    JobManager会先接收到要执行的应用程序，这个应用程序会包括：

    • 作业图（JobGraph）
    • 逻辑数据流图（logical dataflow graph）
    • 打包了所有的类、库和其它资源的JAR包。

    JobManager会把JobGraph转换成一个物理层面的数据流图，这个图被叫做“执行图”（ExecutionGraph），包含了所有可以并发执行的任务。

    JobManager会向资源管理器（ResourceManager）请求执行任务必要的资源，也就是任务管理器（TaskManager）上的插槽（slot）。一旦它获取到了足够的资源，就会将执行图分发到真正运行它们的TaskManager上。

    在运行过程中，JobManager会负责所有需要中央协调的操作，比如说检查点（checkpoints）的协调。

    • 资源管理器（ResourceManager）

    主要负责管理任务管理器（TaskManager）的插槽（slot），TaskManger插槽是Flink中定义的处理资源单元。

    Flink为不同的环境和资源管理工具提供了不同资源管理器，比如YARN、Mesos、K8s，以及standalone部署。

    当JobManager申请插槽资源时，ResourceManager会将有空闲插槽的TaskManager分配给JobManager。如果ResourceManager没有足够的插槽来满足JobManager的请求，它还可以向资源提供平台发起会话，以提供启动TaskManager进程的容器。

    另外，ResourceManager还负责终止空闲的TaskManager，释放计算资源。

    • 任务管理器（TaskManager）

    Flink中的工作进程。通常在Flink中会有多个TaskManager运行，每一个TaskManager都包含了一定数量的插槽（slots）。插槽的数量限制了TaskManager能够执行的任务数量。

    启动之后，TaskManager会向资源管理器注册它的插槽；收到资源管理器的指令后，TaskManager就会将一个或者多个插槽提供给JobManager调用。JobManager就可以向插槽分配任务（tasks）来执行了。

    在执行过程中，一个TaskManager可以跟其它运行同一应用程序的TaskManager交换数据。

    • 分发器（Dispatcher）

    可以跨作业运行，它为应用提交提供了REST接口。

    当一个应用被提交执行时，分发器就会启动并将应用移交给一个JobManager。由于是REST接口，所以Dispatcher可以作为集群的一个HTTP接入点，这样就能够不受防火墙阻挡。Dispatcher也会启动一个Web UI，用来方便地展示和监控作业执行的信息。

    Dispatcher在架构中可能并不是必需的，这取决于应用提交运行的方式。

    这些组件都会在Java虚拟机上运行。

    4.1.1.2、任务提交流程

    上图是从一个较为高层级的视角来看应用中各组件的交互协作。

    如果部署的集群环境不同（例如YARN，Mesos，Kubernetes，standalone等），其中一些步骤可以被省略，或是有些组件会运行在同一个JVM进程中。

    具体地，如果我们将Flink集群部署到YARN上，那么就会有如下的提交流程

    ①Flink任务提交后，Client向HDFS上传Flink的Jar包和配置

    ②之后客户端向Yarn ResourceManager提交任务，ResourceManager分配Container资源并通知对应的NodeManager启动ApplicationMaster

    ③ApplicationMaster启动后加载Flink的Jar包和配置构建环境，去启动JobManager，之后JobManager向Flink自身的RM进行申请资源，自身的RM向Yarn 的ResourceManager申请资源(因为是yarn模式，所有资源归yarn RM管理)启动TaskManager

    ④Yarn ResourceManager分配Container资源后，由ApplicationMaster通知资源所在节点的NodeManager启动TaskManager

    ⑤NodeManager加载Flink的Jar包和配置构建环境并启动TaskManager，TaskManager启动后向JobManager发送心跳包，并等待JobManager向其分配任务。

    4.1.1.3、任务调度原理

    ①客户端不是运行时和程序执行的一部分，但它用于准备并发送dataflow(JobGraph)给Master(JobManager)，然后，客户端断开连接或者维持连接以等待接收计算结果。而Job Manager会产生一个执行图(Dataflow Graph)

    ②当 Flink 集群启动后，首先会启动一个 JobManger 和一个或多个的 TaskManager。由 Client 提交任务给 JobManager，JobManager 再调度任务到各个 TaskManager 去执行，然后 TaskManager 将心跳和统计信息汇报给 JobManager。TaskManager 之间以流的形式进行数据的传输。上述三者均为独立的 JVM 进程。

    ③Client 为提交 Job 的客户端，可以是运行在任何机器上（与 JobManager 环境连通即可）。提交 Job 后，Client 可以结束进程（Streaming的任务），也可以不结束并等待结果返回。

    ④JobManager 主要负责调度 Job 并协调 Task 做 checkpoint，职责上很像 Storm 的 Nimbus。从 Client 处接收到 Job 和 JAR 包等资源后，会生成优化后的执行计划，并以 Task 的单元调度到各个 TaskManager 去执行。

    ⑤TaskManager 在启动的时候就设置好了槽位数（Slot），每个 slot 能启动一个 Task，Task 为线程。从 JobManager 处接收需要部署的 Task，部署启动后，与自己的上游建立 Netty 连接，接收数据并处理。

    注：如果一个Slot中启动多个线程，那么这几个线程类似CPU调度一样共用同一个slot

    4.1.2、DataFlow编程模式

    顾名思义，Dataflow程序描述了数据如何在不同操作之间流动。Dataflow程序通常表现为有向无环图（DAG），图中顶点称为算子（Operator），表示计算。而边表示数据依赖关系。

    算子是Dataflow程序的基本功能单元，他们从输入获取数据，对其进行计算，然后产生数据并发往输出以供后续处理。而所有Flink程序都由三部分算子组成。

    · Source（数据源）：负责获取输入数据。

    · Transformation（数据处理）：对数据进行处理加工，通常对应着多个算子。

    · Sink（数据汇）：负责输出数据。

    类似上图的Dataflow图被称为逻辑图，因为它们表达了高层视角下的计算逻辑。为了执行Dataflow程序，需要将逻辑图转化为物理Dataflow图（执行图），后者会指定程序的执行细节。

    Flink 根据抽象程度分层，提供了三种不同的 API。每一种 API 在简洁性和表达力上有着不同的侧重，并且针对不同的应用场景。

    4.1.3、Window机制

    4.1.3.1、什么是window？

    window是一种切割无限数据为有限块进行处理的手段。

    在流处理应用中，数据是连续不断的，因此我们不可能等到所有数据都到了才开始处理。当然我们可以每来一个消息就处理一次，但是有时我们需要做一些聚合类的处理，例如：在过去的1分钟内有多少用户点击了我们的网页。在这种情况下，我们必须定义一个window窗口，用来收集最近一分钟内的数据，并对这个window窗口内的数据进行计算。

    4.1.3.2、window类型
    • 时间窗口（Time Window）
    • 滚动时间窗口
    • 滑动时间窗口
    • 会话窗口
    • 计数窗口（Count Window）
    • 滚动计数窗口
    • 滑动计数窗口

    • 滚动窗口

    滚动窗口特点：

    依据固定的窗口长度对数据进行切分，多个滚动窗口没有重叠

    • 滑动窗口

    滑动窗口的特点：

    • 可以按照固定的长度 slide 向后滑动固定的距离
    • 滑动窗口由固定的窗口长度 size和滑动间隔 slide组成
    • 可以有重叠(是否重叠和滑动距离有关系)
    • 滑动窗口是固定窗口的更广义的一种形式，滚动窗口可以看做是滑动窗口的一种特殊情况（即窗口大小和滑动间隔相等）
    • 会话窗口

    会话窗口的特点：

    会话窗口模式下，两个窗口之间有一个间隙，被称为Session Gap。当一个窗口在大于Session Gap的时间内没有接收到新数据时，窗口将关闭

    4.1.4、状态及状态后端

    4.1.4.1、State Backends 的作用

    有状态的流计算是Flink的一大特点，状态本质上是数据，数据是需要维护的，例如数据库就是维护数据的一种解决方案。State Backends 的作用就是用来维护State的。一个 State Backend 主要负责两件事：Local State Management(本地状态管理) 和 Remote State Checkpointing（远程状态备份）。

    Local State Management（本地状态管理）

    State Management 的主要任务是确保状态的更新和访问。类似于数据库系统对数据的管理，State Backends 的状态管理就是提供对 State 的访问或更新操作，从这一点上看，State Backends 与数据库很相似。Flink 提供的 State Backends 主要有两种形式的状态管理：

    · 直接将 State 以对象的形式存储到JVM的堆上面

    · 将 State 对象序列化后存储到 RocksDB 中（RocksDB会写到本地的磁盘上）

    以上两种方式，第一种存储到JVM堆中，因为是在内存中读写，延迟会很低，但State的大小受限于内存的大小；第二种方式存储到State Backends上（本地磁盘上），读写较内存会慢一些，但不受内存大小的限制，同时因为state存储在磁盘上，可以减少应用程序对内存的占用。根据使用经验，对延迟不是特别敏感的应用，选择第二种方式较好，尤其是State比较大的情况下。

    Remote State Checkpointing（远程状态备份）

    Flink程序是分布式运行的，而State都是存储到各个节点上的，一旦TaskManager节点出现问题，就会导致State的丢失。State Backend 提供了 State Checkpointing 的功能，将 TaskManager 本地的 State 的备份到远程的存储介质上，可以是分布式的存储系统或者数据库。不同的 State Backends 备份的方式不同，会有效率高低的区别。


    4.1.4.2、如何选择状态后端？

    根据上面的描述，状态后端的主要作用包括在每一个TaskManager节点上存储和管理状态，将状态进行远程备份两个部分。

    FLink 目前提供了三种状态后端，分别是：

    · MemoryStateBackend


    对于状态管理，MemoryStateBackend直接将State对象存储到TaskManager的JVM堆上，如MapState会被存储为一个HashMap对象。can suffer from garbage collection pauses because it puts many long-lived objects on the heap.

    对于远程备份，MemoryStateBackend会将State备份到JobManager的堆内存上，这种方式是非常不安全的，且受限于JobManager的内存大小。

    · FsStateBackend


    对于状态管理，FsStateBackend与MemoryStateBackend一样，将State存储到TaskManager的JVM堆上。

    对于远程备份，FsStateBackend会将State写入到远程的文件系统，如HDFS中。

    · RocksDBStateBackend


    对于状态管理，RocksDBStateBackend将state存储到TaskManager节点上的RocksDB数据库实例上。

    对于远程备份，RocksDBstateBackend会将State备份到远程的存储系统中。

    综上所述，MemoryStateBackend 和 FsStateBackend 都是在内存中进行状态管理，所以可以获取较低的读写延迟，但会受限于TaskManager的内存大小；而RocksDBStateBackend直接将State存储到RocksDB数据库中，所以不受JobManager的内存限制，但会有读写延迟，同时 RocksDBStateBackend 支持增量备份，这是其他两个都不支持的特性。一般来说，如果不是对延迟有极高的要求，RocksDBStateBackend是更好的选择。

    4.2 无货告警中的应用

    4.2.1、技术架构

    诸葛库存预警技术架构可以分为数据接入层、计算层、数据交换层和应用层。库存变动和无货PV消息通过消息队列以source的方式接入Flink计算引擎，经过白名单过滤、维度数据关联、SKU规则判断、品类品牌规则判断的RichFunction处理后，再经过异步IO的方式调用外部实时库存查询接口判断仓网关系下是否缺货，再关联无货PV流输出到下游。另外，Flink提供对外的状态查询接口，供状态管理模块调用，实现Flink内部状态的维护与监控。


    4.2.2、告警规则同步设计

    诸葛用户通过规则管理模块，实现规则的增删改查操作，系统为保证规则是实时生效，需要同步将规则推送到Flink，因此，我们将规则信息对应的MySQL库通过binlog的形式接入Kafka，通过source以实时流的形式接入Flink，对于Flink Job初始化场景，通过直接接入MySQL中规则的生成另外一个流，将两个流进行关联（connect）然后广播到下游算子，在下游算子中利用MapState状态存储存储，通过解析binlog消息后，逻辑上的映射，实现规则状态的增删改。

    告警规则广播流接入核心代码：

    /**
* 品类品牌规则广播流
*
* @param env
* @return
*/
private static final BroadcastStream<CateBrandRule> getCateBrandRuleBroadcastStream(StreamExecutionEnvironment env) {
    final SingleOutputStreamOperator<CateBrandRule> cateBrandRuleDataStream = Objects.requireNonNull(SourceFactory.getFactory(TypeConstant.JDQ))
       .buildJdqSource(ParamsConstant.CATE_BRAND_RULE_JDQ_PREFIX).buildJdwdata(env).name("cateBrandRuleStream(JDQ)").uid("cateBrandRuleStream-jdq")
       .map((MapFunction<JdwData, CateBrandRule>) Parser::parseCateBrandRuleFromJdwData).name("JdwData -> CateBrandRule").uid("formatChangeJdwData2CateBrandRule");
    return (BroadcastStream<CateBrandRule>) SourceFactory.getFactory(TypeConstant.MYSQL)
       .buildMysqlSource(ParamsConstant.CATE_BRAND_RULE_MYSQL).build(env)
       .connect(cateBrandRuleDataStream).process(new CateBrandCoProcessFunction()).name("cateBrandRuleBroadcastConnection").uid("cateBrandRuleBroadcastConnection")
       .broadcast(StateDescriptorBuilder.cateBrandStateDescriptor);
 }
​
/**
* sku规则广播流
*
* @param env
* @return
*/
private static final BroadcastStream<SkuRule> getSkuRuleBroadcastStream(StreamExecutionEnvironment env) {
    final DataStream<SkuRule> skuRuleDataStream = SourceFactory.getFactory(TypeConstant.JDQ)
     .buildJdqSource(ParamsConstant.SKU_RULE_JDQ_PREFIX).buildJdwdata(env).name("skuRuleStream").uid("skuRuleStream")
       .map((MapFunction<JdwData, SkuRule>) Parser::parseSkuRuleFromJdwData).name("JdwData -> SkuRule");
    /**mysql库中规则静态流同实时sku规则流关联后广播*/
    return (BroadcastStream<SkuRule>) SourceFactory.getFactory(TypeConstant.MYSQL)
       .buildMysqlSource(ParamsConstant.SKU_RULE_MYSQL).build(env)
       .connect(skuRuleDataStream).process(new SkuRuleCoProcessFunction()).name("skuRuleBroadcastConnection").uid("skuRuleBroadcastConnection")
       .broadcast(StateDescriptorBuilder.skuRuleStateDescriptor);
 }

    4.2.3、状态设计

    规则通过广播流到下游算子，通过继承KeyedBroadcastProcessFunction，重写父类的processElement和processBroadcastElement方法，实现广播流的处理，其中，processElement方法负责处理数据流，通过入参readOnlyContext可以获取广播状态变量，从而进行状态的判断；processBroadcastElement方法提供对广播流的处理，可以根据业务对广播状态实现增删改等操作。

    /**
 * 广播状态函数类，用于对sku维度规则广播状态更新
 *
 * @author fansiguo
 */
public class SkuRuleProcessFunction extends KeyedBroadcastProcessFunction<String, SkuStockMessage, SkuRule, SkuStockMessage> {
  private final static Logger logger = LoggerFactory.getLogger(SkuRuleProcessFunction.class);
  private static final long serialVersionUID = 5227296503052845858L;
​
  @Override
  public void processElement(SkuStockMessage skuStockMessage, ReadOnlyContext readOnlyContext, Collector<SkuStockMessage> collector) throws Exception {
    if (skuStockMessage == null || skuStockMessage.getSkuId() == null) {
      return
   }
    if (skuStockMessage.getHitFlag()) {
      logger.info("已存在命中规则，不进行sku规则的判断,msg={}", JSON.toJSONString(skuStockMessage));
      collector.collect(skuStockMessage);
      return;
   }
    try {
      String skuId = String.valueOf(skuStockMessage.getSkuId());
      ReadOnlyBroadcastState<String, SkuRule> state = readOnlyContext.getBroadcastState(StateDescriptorBuilder.skuRuleStateDescriptor);
      if (state != null && state.contains(skuId) && state.get(skuId).getStatus() == 1) {
        logger.info("命中sku维度规则,msg={},skuRule={}", JSON.toJSONString(skuStockMessage), JSON.toJSONString(state.get(skuId)));
        skuStockMessage.setHitFlag(true);
        skuStockMessage.setWarningRuleDTO(JSON.parseObject(state.get(skuId).getParamInfo(), WarningRuleDTO.class));
     } else {
        logger.info("未命中sku维度规则,msg={}", JSON.toJSONString(skuStockMessage));
        skuStockMessage.setHitFlag(false);
     }
      collector.collect(skuStockMessage);
   } catch (Exception e) {
      logger.error("sku规则判断错误,msg={}",JSON.toJSONString(skuStockMessage),e);
   }
 }
​
  @Override
  public void processBroadcastElement(SkuRule skuRule, Context context, Collector<SkuStockMessage> collector) throws Exception {
    if (skuRule == null || StringUtils.isBlank(skuRule.getSkuId())) {
      logger.error("无效的skuRule新状态无效");
      return;
   }
    logger.info("收到广播sku规则详情:{}", JSON.toJSONString(skuRule));
    String skuId = skuRule.getSkuId();
    BroadcastState<String, SkuRule> state = context.getBroadcastState(StateDescriptorBuilder.skuRuleStateDescriptor);
    final SkuRule oldSkuRule = context.getBroadcastState(StateDescriptorBuilder.skuRuleStateDescriptor).get(skuId);
    logger.info("skuId={},状态后端存储的sku规则:{}", skuId, JSON.toJSONString(oldSkuRule));
    if (!state.contains(skuId)) {
      logger.info("skuId={}新增sku规则状态:{}", skuId, JSON.toJSONString(skuRule));
      state.put(skuId, skuRule);
      return;
   }
    if (state.contains(skuId) && skuRule.getStatus() == 1) {
      logger.info("skuId={}更新sku规则状态，旧状态:{},新状态:{}", skuId,
          JSON.toJSONString(oldSkuRule), JSON.toJSONString(skuRule));
      state.put(skuId, skuRule);
      return;
   }
    logger.info("skuId={},移除sku规则:{}", skuId, JSON.toJSONString(oldSkuRule));
    state.remove(skuId);
 }
}

    4.2.4、白名单缓存设计

    白名单的缓存设计主要要解决如下3个问题

    1. 如何加载白名单缓存到Flink的算子任务当中，由于白名单数据很大，无法每个节点都存储全量数据，如果进行白名单数据的拆分存储，要解决输入数据与缓存数据的匹配问题。
    2. 如何进行白名单数据的加载，保证数据所在的算子任务，如果有白名单数据一定也在此算子任务上。
    3. 由于白名单数据量很大，在拆分加载时如何直接加载此算子任务需要的数据，而不是读取所有数据进行过滤，提升加载效率。

    为了解决这些问题，我们整体采用如下方案：

    如何进行输入流数据与缓存数据的匹配

    要从flink的输入流的流转原理着手解决此问题，此方案中需要使用keyby对数据进行分组处理，而对数据进行keyby操作时，默认情况下使用系统的分区器对输入的key进行分区计算具体落在哪个下游的算子任务上，使用的方法如下：

    KeyGroupRangeAssignment.assignKeyToParallelOperator

    采用的分区算法如下：

    MathUtils.murmurHash(keyHash) % maxParallelism * parallelism / maxParallelism

    在进行白名单加载时，采用的加载算法与分区算法一致，则可以保证输入流数据与缓存数据的匹配

    如何加载白名单缓存

    白名单的KEY是商品编码，在进行缓存加载时使用商品编码，最大maxParallelism，任务的并行度parallelism计算当前商品编码所在数据会落到的算子任务编号，再与当前的算子任务编号getRuntimeContext().getIndexOfThisSubtask()进行比较，如果数据一致才进行加载，但此种加载方式每个算子任务都需要遍历所有缓存数据，本地计算再进行筛选，有比较大的数据读取及计算开销，可以采用一定的方法直接让算子任务加载到需要的数据。

    如何为算子任务直接加载需要的数据

    此方法是基于一个数学公式如果B是C的整数倍，那么(A % B) % C = A % C ，结合分区算法：

    MathUtils.murmurHash(keyHash) % maxParallelism * parallelism / maxParallelism

    我们把MathUtils.murmurHash(keyHash)当做A，把maxParallelism当做C，选用一个较大的并且是C的整数倍且有一定扩展性的数当做B。

    1. 这里根据数量大小比如我们把白名单的缓存数据分为10240部分，每一部分的标号即是A%B分配的算法为：
    2. MathUtils.murmurHash(keyHash) % 10240
    3. 要求Flink在设置最大并发数maxParallelism时必须可以被10240整除，可以根据对maxParallelism的设置接扩展需求灵活选取白名单划分的数量
    4. 进行白名单加载时，把1-10239这些编号带入公式，以当前算子任务所在编号为结果就能算出哪些部分的数据是当前节点需要的数据。
    5. 对筛选出来的数据进行加载即完成指定加载需要的缓存。

    五、总结

    限于篇幅和个人水平，本文重点介绍了京东诸葛智享库存预警平台的架构和设计，从运行原理、DataFlow编程模式和状态角度，展示了一种Flink在实时预警的典型应用场景。目前诸葛智享正在向越来越多的商家开放，业务的复杂度也与日俱增，后续系统的技术也会随之迭代升级，也请期待我们的方案会进一步得到优化，读者有任何的意见或建议也欢迎同我们取得联系。


    注：诸葛智享首页 https://ysc.jd.com/

    castechelon
    文章数
    3
    阅读量
    1988

    作者其他文章

    01 研发效能提升之第一性原理
     
    01 研发提效：想快速定制一个OLAP应用？你可以这么做