人工智能机器学习分布式训练图计算后端【免费下载链接】angelA Flexible and Powerful Parameter Server for large-scale machine learning项目地址https://gitcode.com/gh_mirrors/an/angel点击查看免费下载Angel 是一套面向大规模机器学习的灵活参数服务器Parameter Server框架其自研的计算图Computational Graph以Layer层为最小节点专为推荐系统相关的深度学习算法设计。本文将深入讲解 Angel 计算图的设计动机、Layer 与 AngelGraph 的源码级结构、PlaceHolder 数据入口机制以及基于状态机的训练执行原理并给出可直接运行的 JSON 网络配置示例。读完本文你将掌握如何在 Angel 上以“写配置、不写代码”的方式构建 DNN、DeepFM、FM 等网络并理解其前向/反向传播与 PS 参数更新的完整调用链。1. 什么是计算图Angel 与 TensorFlow 的差异化设计计算图是 TensorFlow、Caffe、MXNet 等主流深度学习框架的核心抽象Spark 等大数据框架也借助计算图完成任务调度。Angel 同样提供计算图框架但与 TensorFlow 相比更加轻量主要体现在三个方面粗粒度Coarse GranularityAngel 计算图中的一个节点代表一个Layer层而不是一个算子operator。TensorFlow 以 operator 为节点的设计虽然灵活、利于二次封装但也带来更高的学习曲线和更大的开发工作量旧版 TensorFlow 曾因“API 过于底层、开发效率低”而备受诟病。因此 Angel 只提供基于 Layer 的粗粒度计算图。特征交叉Feature Intersection推荐类算法中Embedding 之后的特征往往需要先做交叉再送入 DNN。注意这里指对 Embedding 输出的特定运算区别于通用特征工程中的人工交叉。在 TensorFlow、Caffe、Torch 上实现特征交叉较为繁琐而 Angel 直接内置了特征交叉层。自动生成网络Auto-Generated Network借鉴 Caffe 的思路Angel 支持从 JSON 文件直接生成深度神经网络用户无需编写任何代码即可构建自己的网络。需要特别说明的是Angel 当前不支持 CNN、RNN 等网络只聚焦推荐场景中常用的深度学习算法。这一点在 docs/basic/layers_on_angel.md 中有同样清晰的界定。2. 计算图的构建2.1 Layer 的基本结构理解计算图之前先理解其组成单元 Layer 的结构更详细的 Layer 说明见 Layers on Angel。在源码 Layer.scala 中abstract class Layer(val name: String, val outputDim: Int)(implicit val graph: AngelGraph) extends Serializable { var status: STATUS.Value STATUS.Null val input new ListBuffer[Layer]() val consumer new ListBuffer[Layer]() def addInput(layer: Layer): Unit { input.append(layer) } def addConsumer(layer: Layer): Unit { consumer.append(layer) } def calOutput(): Matrix def gatherGrad(): Matrix { ... } }上述抽象类清晰展示了 Layer 的核心职责statusAngel 计算图中的节点是有状态的节点状态由状态机管理详见第 3.1 节。状态枚举定义在同文件顶部的STATUS对象中Null, Forward, Backward, Gradient, Update。inputListBuffer对象记录该节点的输入。一个 Layer 可以拥有多个输入层通过addInput(layer: Layer)依次添加。outputDimAngel 中每个节点至多有一个输出outputDim用于指定输出维度。consumer虽然每个 Layer 只有一个输出但该输出可以被多个消费者消费同样用ListBuffer表示。构建图时用户调用输入层的addConsumer(layer: Layer)方法指定输出层被哪些层消费。值得注意的还有gatherGrad()当某层有多个 consumer 时它会遍历所有消费者调用LinearLayer.calGradOutput()或JoinLayer.calGradOutput(idx)获取梯度并累加iadd这是后续反向传播中“多消费者梯度汇聚”的关键实现。实际上用户无需关心建图的具体操作因为建图已在InputLayer、LinearLayer、JoinLayer三个基类中完成abstract class InputLayer(name: String, outputDim: Int)(implicit graph: AngelGraph) extends Layer(name, outputDim)(graph) { graph.addInput(this) def calBackward(): Matrix } abstract class JoinLayer(name: String, outputDim: Int, val inputLayers: Array[Layer])(implicit graph: AngelGraph) extends Layer(name, outputDim)(graph) { inputLayers.foreach { layer layer.addConsumer(this) this.addInput(layer) } def calGradOutput(idx: Int): Matrix } abstract class LinearLayer(name: String, outputDim: Int, val inputLayer: Layer)(implicit graph: AngelGraph) extends Layer(name, outputDim)(graph) { inputLayer.addConsumer(this) this.addInput(inputLayer) def calGradOutput(): Matrix }注意LossLayer是一种特殊的LinearLayer因此上文中未单独列出。从这三类基类的构造逻辑可以看到建图就是通过互相登记addConsumer/addInput完成拓扑关系编织的。从拓扑结构看Angel 的 Layer 可分为三类详见 layers_on_angel_en.mdverge边界节点只出现在输入或输出端如InputLayerSimpleInputLayer、Embedding和LossLayerSimpleLossLayer、SoftmaxLossLayer。linear单输入单输出全连接层FCLayer以及一系列特征交叉层如 PNN(inner) 使用的BiInnerCross、FM 使用的二阶隐式交叉BiInnerSumCross、NFM 使用的BiInteractionCross等。join多输入单输出ConcatLayer拼接多个输入输出稠密矩阵、SumPooling求和、MulPooling相乘、DotPooling对应元素相乘后按行求和。这些类的具体实现位于 network/layers 目录的join/、linear/、verge/子目录中。2.2 AngelGraph 的基本结构通过 input/consumer 已经构建起一张复杂的图。虽然图可以从任意节点遍历但 AngelGraph 仍然存储边界verge节点以便对图做整体操作。源码见 AngelGraph.scalaclass AngelGraph(val placeHolder: PlaceHolder, val conf: SharedConf) extends Serializable { def this(placeHolder: PlaceHolder) this(placeHolder, SharedConf.get()) private val inputLayers new ListBuffer[InputLayer]() private var lossLayer: LossLayer _ private val trainableLayer new ListBuffer[Trainable]() def addInput(layer: InputLayer): Unit { inputLayers.append(layer) } def setOutput(layer: LossLayer): Unit { lossLayer layer } def getOutputLayer: LossLayer { lossLayer } def addTrainable(layer: Trainable): Unit { trainableLayer.append(layer) } def getTrainable: ListBuffer[Trainable] { trainableLayer } }AngelGraph 维护两类边界节点inputLayer输入层数据是这类节点的输入。AngelGraph 保存 inputLayers 是为了方便反向传播——只需依次调用每个 inputLayer 的calBackward方法即可。为了将所有输入层收集齐全Angel 要求每个输入层通过调用 AngelGraph 的addInput方法显式注册自己。实际上这些操作已封装在InputLayer基类中graph.addInput(this)用户无需关心。lossLayer损失层Angel 当前不支持多任务学习因此每个计算图只能有一个 lossLayer。损失层节点是为前向传播设计的只需调用 lossLayer 节点的predict或calOutput方法即可完成前向。由于LossLayer是LinearLayer的子类用户可以手动调用setOutput(layer: LossLayer)添加自定义损失层更常见的做法是直接配置一个 lossFunc见下文 JSON 示例中的lossfunc字段。有了 inputLayers 与 lossLayer前向传播通过 lossLayer 的predict和反向传播通过 inputLayer 的calBackward就变得非常容易。最后一个问题是梯度计算中的参数更新。AngelGraph 通过trainableLayer变量统一收集所有带可更新参数的层从而便捷地完成参数更新。Trainabletrait 定义了pullParams、pushGradient、update、init、loadParams、saveParams等方法见 Layer.scalaAngelGraph 中的pullParams、pushGradient、update都是对trainableLayer的遍历调度。从源码还可以看到AngelGraph 额外维护了一个matrixCtxsMatrixContext列表负责为各个可训练层在 PS 上创建参数矩阵createMatrices并支持模型加载loadModel与保存saveModel。这意味着“计算图”不仅是网络结构也直接承载了参数矩阵的生命周期管理。2.3 数据入口PlaceHolder前面已经通过 input/consumer 明确了层间关系图的边并存储了特殊节点inputLayer/lossLayer/trainableLayer用于前向/反向传播和参数更新。那么数据如何进入计算图答案是PlaceHolder。构建 AngelGraph 时PlaceHolder 被传入 Graph再作为隐式参数传给所有 Layer。这样每个 Layer 都能访问 PlaceHolder即输入数据。实现见 PlaceHolder.scalaclass PlaceHolder(val conf: SharedConf) extends Serializable { def feedData(data: Array[LabeledData]): Unit def getFeats: Matrix def getLabel: Matrix def getBatchSize: Int def getFeatDim: Long def getIndices: Vector }Angel 目前每个图只允许一个 PlaceHolder未来版本会移除该限制以支持多数据输入且 PlaceHolder 只保存一个 mini-batch 的数据。将Array[LabeledData]格式的数据通过feedData灌入 PlaceHolder 后可以从中取得特征Feature特征维度Feature dimension标签Label批量大小BatchSize特征索引Feature index从源码看getFeats会根据每条样本向量IntDoubleVector/IntFloatVector/LongDoubleVector/LongFloatVector的稠密/稀疏属性分别生成稠密矩阵或RBComp*行式压缩矩阵getIndices则会依据SharedConf.keyType()int或long对当前 batch 内出现的所有特征下标做去重、排序后返回——这正是后续“按需拉取参数”pullParams的依据PS 上只拉取当前 mini-batch 真正用到的参数列。3. Angel 中计算图的工作原理3.1 状态机Angel 的状态机包含以下状态Null初始状态。每次feedData之后Graph 都会被置为该状态。Forward表示前向传播已完成。Backward表示反向传播已完成但参数梯度尚未计算。Gradient表示梯度已计算并推送到 PS。Update表示模型更新已完成。这些状态依次流转如下图所示状态机设计的主要目的是保证操作顺序、减少重复计算。例如当多个层共享消费同一层的输出时程序在执行运算前会先检查该层的当前状态从而保证该计算只执行一次。状态机在代码中的体现如下同时可见于 AngelGraph.scala 的feedData实现def feedData(data: Array[LabeledData]): Unit { deepFirstDown(lossLayer.asInstanceOf[Layer])( (lay: Layer) lay.status ! STATUS.Null, (lay: Layer) lay.status STATUS.Null ) placeHolder.feedData(data) } override def calOutput(): Matrix { status match { case STATUS.Null // do come forward calculation status STATUS.Forward case _ } output } override def calBackward(): Matrix { status match { case STATUS.Forward val gradTemp gatherGrad() // do backward calculation status STATUS.Backward case _ } backward } override def pushGradient(): Unit { status match { case STATUS.Backward // calculate gradient and push to PS status STATUS.Gradient case _ } } override def update(epoch: Int 0): Unit { status match { case STATUS.Gradient optimizer.update(weightId, 1, epoch) status STATUS.Update case _ throw new AngelException(STATUS Error, please calculate Gradient frist!) } }从源码实现可以看到feedData会从 lossLayer 出发做一次深度优先遍历deepFirstDown把图中所有非 Null 状态的节点重置为 Null从而开启新一轮迭代而calOutput/calBackward/pushGradient/update各自都只会“认领”前置状态下的节点若状态不匹配则直接跳过或抛异常这就保证了训练流程的严格有序性。3.2 Graph 的训练过程这里给出训练过程的整体框架详细代码在GraphLearner中见 GraphLearner.scaladef trainOneEpoch(epoch: Int, iter: Iterator[Array[LabeledData]], numBatch: Int): Double { var batchCount: Int 0 var loss: Double 0.0 while (iter.hasNext) { graph.feedData(iter.next()) graph.pullParams(epoch) loss graph.calLoss() // forward graph.calBackward() // backward graph.pushGradient() // pushgrad PSAgentContext.get().barrier(ctx.getTaskId.getIndex) if (decayOnBatch) { graph.setLR(ssScheduler.next()) } if (ctx.getTaskId.getIndex 0) { graph.update(epoch * numBatch batchCount, 1) // update parameters on PS } PSAgentContext.get().barrier(ctx.getTaskId.getIndex) batchCount 1 LOG.info(sepoch $epoch batch $batchCount is finished!) } loss }整个训练流程按如下步骤工作feedData将数据灌入 PlaceHolder并将 Graph 的状态置为null。Pull parameters拉取参数只拉取当前 mini-batch 计算所需的参数依据 PlaceHolder 的getIndices这正是 Angel 能够训练超高维模型的关键。以Embedding层为例见 Embedding.scala其pullParams使用GetColsFunc按placeHolder.getIndices仅拉取命中列首次拉取epoch 0时还会带上RandomNormalInitFunc(0.0, 0.00001)完成参数初始化。Forward calculation前向计算从 LossLayer 开始级联调用各层 inputLayer 的calOutput方法依次计算输出随后将层状态置为forward。对于已处于forward状态的层Angel 直接返回上一次计算结果避免重复计算。Backward calculation反向计算依次调用 Graph 的 inputLayer从而级联调用第一层的calGradOutput完成反向传播。计算完成后层状态置为backward。同样已处于backward状态的层会直接返回之前的计算结果。梯度的计算与推送上一步只计算了节点的梯度参数梯度在本步通过Trainable的pushGradient方法计算。该方法先计算梯度再推送到 PS最后把状态置为gradient。以Embedding为例其pushGradient会按 batch 内每条样本的特征下标聚合更新mergeUpdate除以归一化因子divider由OptUtils.getNormal依据运行模式计算再通过UpdateColsFunc推送回 PS。Gradient update梯度更新由于梯度在 PS 上更新只需要一个 WorkerSpark on Angel 模式下由 Driver发送一个梯度更新的 PSF 即可。不同优化器的更新方法不同而 Angel 中的优化器本质上就是一个 PSFParameter Server Function。更新前需要一次同步barrier确保所有梯度都已推送更新后再做一次同步确保 Worker 拉到的参数是最新的。更新完成后状态置为update。值得一提的是源码实现中还支持按 batch 或按 epoch 做学习率衰减当配置项ml.opt.decay.on.batch为 true 时每个 batch 后调用graph.setLR(ssScheduler.next())更新学习率否则在每个 epoch 开始时更新一次见 GraphLearner.scala。4. 实战用 JSON 构建计算图网络Angel 的计算图支持通过 JSON 配置自动生成网络。JSON 文件由data、model、train、layers四大部分组成可选default_optimizer、default_transfunc、default_lossfunc作为层的默认值。JsonUtils会解析该 JSON先用data/train/model段更新SharedConf配置再通过fillGraph按layers数组顺序实例化 Layer 并建立拓扑连接见 JsonUtils.scala最后从simplelosslayer出发构建完整网络。以下给出仓库中真实的测试配置示例来源mllib/src/test/jsons。4.1 一个完整的 DNN 配置dnn.json{ data: { format: libsvm, indexrange: 300, validateratio: 0.2, posnegratio: 0.1 }, model: { modeltype: T_DOUBLE_SPARSE, modelsize: 300 }, train: { epoch: 5, lr: 0.1, decayclass: WarmRestarts, decayalpha: 0.05 }, default_optimizer: momentum, layers: [ { name: denseinputlayer, type: simpleinputlayer, outputdim: 500, transfunc: relu }, { name: fclayer, type: FCLayer, outputdims: [100, 1], transfuncs: [relu, identity], inputlayer: denseinputlayer }, { name: simplelosslayer, type: simplelosslayer, lossfunc: logloss, inputlayer: fclayer } ] }各字段含义与取值要点data.format输入数据格式支持libsvm、dummy、dense。data.indexrange特征索引范围特征维度需与训练数据匹配。data.validateratio验证集比例data.posnegratio正负样本采样比例。model.modeltype模型矩阵类型如T_DOUBLE_SPARSE、T_FLOAT_DENSE等默认T_FLOAT_DENSE对应配置项ml.model.type见 MLConf.scala。model.modelsize模型规模有效特征数。train.epoch训练轮数对应ml.epoch.num默认 30train.lr初始学习率对应ml.learn.rate默认 0.5。train.decayclass / decayalpha / decaybeta学习率衰减策略如StandardDecay、WarmRestarts及衰减参数。default_optimizer未显式指定 optimizer 的层所采用的默认优化器如momentum、adam、ftrl等。layers按顺序声明网络层层间通过inputlayer/inputlayers字段引用前面的层名完成连线。注意FCLayer支持用outputdims数组一次声明多层全连接这里即 100→1 两层与之对应的transfuncs数组声明每层的激活函数Angel 会自动展开并合并连续的全连接层。4.2 含 Embedding 与特征交叉的 DeepFM 配置deepfm.json{ data: { format: dummy, indexrange: 148, numfield: 13, validateratio: 0.1, sampleratio: 0.2 }, model: { modeltype: T_FLOAT_DENSE, modelsize: 148 }, train: { epoch: 5, numupdateperepoch: 10, lr: 0.5, decayclass: WarmRestarts, decayalpha: 0.05 }, default_optimizer: Momentum, layers: [ { name: wide, type: simpleinputlayer, outputdim: 1, transfunc: identity }, { name: embedding, type: embedding, numfactors: 8, outputdim: 104, optimizer: { type: momentum, momentum: 0.9, reg2: 0.01 } }, { name: fclayer, type: FCLayer, outputdims: [100, 100, 1], transfuncs: [relu, relu, identity], inputlayer: embedding }, { name: biinnersumcross, type: BiInnerSumCross, inputlayer: embedding, outputdim: 1 }, { name: sumPooling, type: SumPooling, outputdim: 1, inputlayers: [wide, biinnersumcross, fclayer] }, { name: simplelosslayer, type: simplelosslayer, lossfunc: logloss, inputlayer: sumPooling } ] }这个配置完整展示了 Angel 计算图“边”的多种形态Embedding 层type: embeddingnumfactors指定 Embedding 向量的维度outputdim为查表输出的维度。embedding 矩阵在 PS 上的行数为(numSlot 1) * numFactorsnumSlot由优化器决定见 Embedding.scala。data.numfield对应ml.fm.field.num声明样本的字段数。特征交叉层BiInnerSumCrossFM 使用的二阶特征隐式交叉输入为 embedding 输出属于linear类层。Join 层SumPooling通过inputlayers数组聚合wide线性部分、biinnersumcrossFM 二阶部分与fclayer深度部分三个输入输出一个标量——这正是 DeepFM 将“宽”“深”“交叉”三路合一的实现方式。损失层simplelosslayer通过inputlayer指向sumPooling并指定lossfunc如logloss作为整张图的输出边界lossLayer。FM 配置fm.json结构类似只是sumPooling只聚合wide与biinnersumcross两路。仓库中还有pnn.json使用BiInnerCross实现 PNN(inner) 特征交叉、nfm.json使用BiInteractionCross、softmax.json使用SoftmaxLoss与多分类逻辑等示例可作为不同特征交叉/损失函数的对照参考。4.3 如何运行以计算图为基础的模型DNN、DeepFM、FM、PNN、NFM 等通过GraphRunner提交训练对应的训练任务是GraphTrainTask模型封装是GraphModel均位于 graphsubmit 目录。核心链路为GraphModel构造AngelGraph与PlaceHolder并通过JsonUtils解析 JSON 构建网络buildNetworkGraphLearner按第 3.2 节的流程驱动trainOneEpoch执行前向、反向、梯度推送与参数更新训练与预测任务分别由GraphTrainTask/GraphPredictTask承载。实际提交时需要将 JSON 文件路径通过-Dangel.ml.confjson路径传给任务JsonUtils.init()中读取该配置项见 JsonUtils.scala并搭配ml.model.class.name指定对应的模型类。关于本地运行与 YARN 集群运行的完整命令可参考 local_run_en.md 与 run_on_yarn_en.md各算法的模型类与运行方式见 docs/algo 目录下的算法文档。5. 小结Angel 的计算图以 Layer 为节点、以 JSON 为网络描述语言将前向传播、反向传播、参数梯度推送与 PS 更新统一封装为基于状态机的有序流程。与 TensorFlow 等框架相比它牺牲了算子级灵活性换来了面向推荐场景的开箱即用内置 Embedding 查表、特征交叉层、多路聚合层并通过PlaceHolder按 batch 按需拉取参数从而支撑超高维稀疏模型的高效训练。如果你正在搭建 DNN、DeepFM、FM 等推荐模型且希望避开繁琐的底层算子开发Angel 的这套轻量计算图是值得直接上手的方案。更多 Layer 级细节可继续阅读 Layers on AngelJSON 配置的完整字段说明见 json_conf_en.md。赞分享人工智能机器学习分布式训练图计算后端【免费下载链接】angelA Flexible and Powerful Parameter Server for large-scale machine learning项目地址https://gitcode.com/gh_mirrors/an/angel点击查看免费下载相关推荐米家API终极实战5个智能家居自动化场景深度解析米家API终极实战5个智能家居自动化场景深度解析 米家API是Python开发者控制小米智能家居设备的完整解决方案提供了从账号认证到设备控制的全套工具链。本智能家居物联网MCP 服务AI 技能Angel 图计算基于 Spark On Angel 的 K-Core 大规模网络核值计算实践Angel 图计算基于 Spark On Angel 的 K Core 大规模网络核值计算实践 K Corek degenerate graphk 核/核人工智能机器学习分布式训练图计算后端基于 Angel 的 LINE 大规模网络嵌入训练指南从算法原理到 Spark On Angel 实战基于 Angel 的 LINE 大规模网络嵌入训练指南从算法原理到 Spark On Angel 实战 LINELarge scale Informatio人工智能机器学习分布式训练图计算后端上一篇如何防御Rogue-MySql-Server文件泄露攻击LOCAL INFILE安全加固完整指南下一篇LunaTV 直播接入 IPTV一条 M3U 订阅变成高清频道列表的完整教程创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考