1、演讲者：林东-阿里巴巴高级技术专家 Apache Kafka PMC高赟-阿里巴巴技术专家 Apache Flink Committer共同作者：张智鹏 阿里巴巴高级开发工程师为实时机器学习设计的算法接口与迭代引擎为实时机器学习设计的算法接口与迭代引擎流批一体的迭代引擎流批一体的迭代引擎FlinkFlink MLML 生态建设生态建设面向实时机器面向实时机器学习的学习的APIAPI#1#2#3#1#1面向实时机器学习的面向实时机器学习的APIAPI为什么需要Flink ML API减少重复开发。提升社区合作效率。复用复用算法模块算法模块隐藏实现细节。统一算法使用体验。提升业务开发效率。提升算法

2、开发效率。支持以json格式表达组合。组合并封装组合并封装算法模块算法模块优化优化用户体验用户体验Flink ML API需要改进的要素传统的Estimator/ModelAPI无法表达需要多输入多输出的算法表达能力表达能力传统的Estimator/Model API无法透出会被实时更改的模型数据实时训练实时训练之前的Flink ML API存取大型模型需要用户的繁琐操作易用性易用性支持模型数据实时传输的API145632提供无模型语义的抽象类更易用的模块存取API流批一体的训练和推理API更易用的参数存取API支持多输入图结构的机器学习作业Flink ML API设计要素Flink ML A

3、PI代码概览更易用的参数存取API更易用的模块存取API支持多输入的训练API支持多输入多输出的推理API支持模型数据实时传输的API提供无模型语义的抽象类WithParams.Stagevoid save(String path)static Stage load(String path)EstimatorModel fit(Table inputs)TransformerAlgoOperatorTable transform(Table inputs)ModelModel setModelData(Table data)Table getModelData()在线训练+实时模型部署静态数据

4、(Flink HDFS Source)动态数据(Flink Kafka Source)特征处理(AlgoOperator)在线训练(Estimator)前端服务器模型数据（Flink Kafka sink）训练数据原始数据outputs=algoOperator.transform(inputs)model=estimator.fit(inputs)model.getModelData.sinkTo()前端服务器前端服务器model=new ModelA()model.setModelData()outputs=model.transform(inputs)支持DAG图结构的模块组合及封装特征

5、处理(AlgoOperator)在线训练(Estimator)特征处理(AlgoOperator)在线推理(Model)EstimatorAModelA训练数据A训练数据B推理数据A推理数据B推理结果生成ModelAGraphBuilder生成EstimatorA算法开发者不需要为生成的算子实现fit/transform/save/load构建DAG#2 2流批一体的迭代引擎流批一体的迭代引擎流批一体迭代：场景Unified Iteration for Streaming/Batch:Scenarios离线训练算法Offline Training线性分类/拟合决策树/Boost Tree聚类K

6、NN频繁项集LDA实时训练算法Online TrainingFTRL在线聚类算法前向反馈Backward Notification预测服务实时参数调整需要同时支持离线与在线场景Required supports for both offline and online cases流批一体迭代：场景Unified Iteration for Streaming/Batch:Scenarios初始模型Initial Model模型缓存Model Cache训练数据源（有限）Training Data Source(Bounded)模型更新Model Update训练节点*nTraining Node

7、*n 离线训练离线训练1.模型缓存节点读入初始模型2.训练节点读入所有训练数据3.while not converged:3.1 模型缓存节点发布最新模型到训练节点3.2 训练节点选择下一个mini-batch的数据并更新模型同步：模型缓存节点等到收齐n个更新后发布新版模型异步：模型缓存节点每收到一个更新后立刻向其发布新版模型初始模型Initial Model模型缓存Model Cache训练数据源（无限）Training Data Source(Unbounded)模型更新Model Update在线训练在线训练1.模型缓存节点读入初始模型2.While True:3.1 模型缓存节点发布最

8、新模型到训练节点3.2 训练节点读入下一个下一个MiniMini-batchbatch的数据的数据并更新模型并更新模型3.3训练节点阻塞训练数据直到读取完模型更新同步：模型缓存节点等到收齐n个更新后发布新版模型异步：模型缓存节点每收到一个更新后立刻向其发布新版模型参数更新对每一条数据的处理结果进行评估，按需更新参数输入数据InputData模型服务Model Servering结果评估Evaluation参数更新Parameter Update流批一体迭代：需求Unified Iteration for Streaming/Batch:Requirements统一的迭代图结构统一的迭代终止判断

9、提供整个数据集处理完一轮的通知由上层提供mini-batch处理的功能-更好的兼容逐个mini-batch/并行多个mini-batch的处理-更好的兼容per-record类型的处理-简化迭代本身的实现，避免功能耦合产生的复杂设计流批一体迭代：设计Unified Iteration for Streaming/Batch:DesignUnionUnion回边回边迭代体有回边的输入无回边的输入输出输出迭代体In:有回边的输入In:无回边的输入Out:回边Out:输出进度追踪所有数据每处理一轮后给出通知无回边的输入每轮重放每轮不重放算子生命周期每轮重建每轮不重建终止所有输入结束&迭代中没有数据特

10、定算子某一轮没有输出数据有回边的输入流批一体迭代:APIUnified Iteration for Streaming/Batch:APIDataStream initParameters=DataStreamTuple2 dataset=DataStreamList resultStreams=Iterations.iterate(DataStreamList.of(initParameters),ReplayableDataStreamList.notReplay(dataset),IterationConfig.newBuilder().setOperatorRoundMode(ALL_

11、ROUND).build();(variableStreams,dataStreams)-DataStream modelUpdate=variableStreams.get(0);DataStreamTuple2 dataset=dataStreams.get(0);DataStream newModelUpdate=DataStream modelOutput=return new IterationBodyResult(DataStreamList.of(newModelUpdate),DataStreamList.of(modelOutput);DataStream finalMode

12、l=resultStreams.get(final_model);有回边的输入数据无回边的输入数据算子每轮不重建迭代体回边更新最终输出流批一体迭代:APIUnified Iteration for Streaming/Batch:APIpublic static class ModelCacheFunction extends ProcessFunctionimplements IterationListener private final double parameters=new doubleN_DIM;public void processElement(double update,Co

13、ntext ctx,Collector output)/Suppose we have a util to add the second array to the first.ArrayUtils.addWith(parameters,update);void onEpochWatermarkIncremented(int epochWatermark,Context context,Collector collector)if(epochWatermark N_EPOCH*N_BATCH_PER_EPOCH)collector.collect(parameters);public void

14、onIterationEnd(int round,Context context)context.output(FINAL_MODEL_OUTPUT_TAG,parameters);初始模型Initial Model模型缓存Model Cache训练数据源（有限）Training Data Source(Bounded)模型更新Model Update训练节点*nTraining Node*n 最终输出收齐所有更新后再发布新模型，从而实现同步计算流批一体迭代:实现Unified Iteration for Streaming/Batch:ImplementationInput节点Input节点

15、Head节点Tail节点基于内存的队列Head OperatorCoordinator算子Wrapper增加新的节点对用户节点进行Wrap，从而可以管理算子的生命周期基于colocation和本地内存队列的回边流批一体迭代:实现Unified Iteration for Streaming/Batch:ImplementationInput节点Input节点Head节点Tail节点基于内存的队列Head OperatorCoordinator算子Wrapper对输入数据进行Wrapperepoch=0element=5每经过一轮epoch加1epoch=1element=5所有数据之后增加标记

16、元素epoch 0 finished每经过一轮epoch加1Input节点Input节点Head节点Tail节点基于内存的队列Head OperatorCoordinator算子Wrapper正常的CheckpointBarierOperatorCoordinator的虚拟Barrier带环的Chandy-Lamport算法：同时存储来自回边的数据增加来自OperatorCoordinator的虚拟Barrier-保证全局对齐消息与Barrier的顺序对齐消息后的Checkpoint：最小化需要缓存的数据（开发中）流批一体迭代:实现Unified Iteration for Streaming

17、/Batch:Implementation流批一体迭代:总结Unified Iteration for Streaming/Batch:Summary 流批统一的迭代执行引擎 支持Exactly-Once容错 未来将进一步支持Batch模式未来将提供更多的上层算法开发工具#3 3FlinkFlink MLML 生态建设生态建设FlinkFlink MLML 现状现状 PMC授权建立的中立的Flink生态项目组织 Learning on Flink。适配Flink ML API并支持基于TensorFlow的在线训练+在线推理FlinkFlink MLML 未来工作未来工作支持Python程序编

18、写机器学习作业和调用Java算子支持支持PythonPython调用调用ClinkClink提供C+特征处理算子。适配Flink ML API并对接flink-ml中的Estimator。开箱即用的开箱即用的JavaJava算子算子将Alink适配Flink ML API并迁移进入flink-ml代码库参考资料 FLIP-173:Support DAG of algorithms FLIP-174:Improve the WithParam interface FLIP-175:Compose Estimator/Model/AlgoOperator from DAG of Estimator/Model/AlgoOperator FLIP-176:Unified Iteration to Support Algorithms Flink ML repo: Deep Learning on F