Alink漫谈(十三) :在线学习算法FTRL 之 具体实现

前端之家收集整理的这篇文章主要介绍了Alink漫谈(十三) :在线学习算法FTRL 之 具体实现前端之家小编觉得挺不错的,现在分享给大家,也给大家做个参考。

Alink漫谈(十三) :在线学习算法FTRL 之 具体实现

0x00 摘要

Alink 是阿里巴巴基于实时计算引擎 Flink 研发的新一代机器学习算法平台,是业界首个同时支持批式算法、流式算法的机器学习平台。本文和上文一起介绍了在线学习算法 FTRL 在Alink中是如何实现的,希望对大家有所帮助。

0x01 回顾

书接上回 Alink漫谈(十二) :在线学习算法FTRL 之 整体设计 。到目前为止,已经处理完毕输入,接下来就是在线训练。训练优化的主要目标是找到一个方向,参数朝这个方向移动之后使得损失函数的值能够减小,这个方向往往由一阶偏导或者二阶偏导各种组合求得。

为了让大家更好理解,我们再次贴出整体流程图:

在这里插入图片描述

0x02 在线训练

在线训练主要逻辑是:

  • 1)加载初始化模型到 dataBridge;dataBridge = DirectReader.collect(model);
  • 2)获取相关参数。比如vectorSize默认是30000,是否 hasInterceptItem;
  • 3)获取切分信息。splitInfo = getSplitInfo(featureSize,hasInterceptItem,parallelism); 下面马上会用到。
  • 4)切分高维向量。初始化数据做了特征哈希,会产生高维向量,这里需要进行切割。 initData.flatMap(new SplitVector(splitInfo,vectorSize,vectorTrainIdx,featureIdx,labelIdx));
  • 5)构建一个 IterativeStream.ConnectedIterativeStreams iteration,这样会构建(或者说连接)两个数据流:反馈流和训练流;
  • 6)用iteration来构建迭代体 iterativeBody,其包括两部分:CalcTask,ReduceTask;
    • 6.1)CalcTask分成两个部分。flatMap1 是分布计算FTRL迭代需要的predict,flatMap2 是FTRL的更新参数部分;
    • 6.2)ReduceTask分为两个功能:“归并这些predict计算结果“ / ”如果满足条件则归并模型 & 向下游算子输出模型“;
  • 7)result = iterativeBody.filter;基本是以时间间隔为标准来判断(也可以认为是时间驱动),"时间未过期&向量有意义" 的数据将被发送回反馈数据流,继续迭代,回到步骤 6),进入flatMap2
  • 8)output = iterativeBody.filter;符合标准(时间过期了)的数据将跳出迭代,然后算法会调用WriteModel将LineModelData转换为多条Row,转发给下游operator(也就是在线预测阶段);即定时把模型更新给在线预测阶段

2.1 预置模型

前面说到,FTRL先要训练出一个逻辑回归模型作为FTRL算法的初始模型,这是为了系统冷启动的需要。

2.1.1 训练模型

具体逻辑回归模型设定/训练是 :

  1. // train initial batch model
  2. LogisticRegressionTrainBatchOp lr = new LogisticRegressionTrainBatchOp()
  3. .setVectorCol(vecColName)
  4. .setLabelCol(labelColName)
  5. .setWithIntercept(true)
  6. .setMaxIter(10);
  7. BatchOperator<?> initModel = featurePipelineModel.transform(trainBatchData).link(lr);

训练好之后,模型信息是DataSet类型,位于变量 BatchOperator<?> initModel之中,这是一个批处理算子。

2.1.2 加载模型

FtrlTrainStreamOp将initModel作为初始化参数。

  1. FtrlTrainStreamOp model = new FtrlTrainStreamOp(initModel)

在FtrlTrainStreamOp构造函数中会加载这个模型;

  1. dataBridge = DirectReader.collect(initModel);

具体加载时通过MemoryDataBridge直接获取初始化模型DataSet中的数据。

  1. public MemoryDataBridge generate(BatchOperator batchOperator,Params globalParams) {
  2. return new MemoryDataBridge(batchOperator.collect());
  3. }

2.2 分割高维向量

从前文可知,Alink的FTRL算法设置的特征向量维度是30000。所以算法第一步就是切分高维度向量,以便分布式计算。

  1. String vecColName = "vec";
  2. int numHashFeatures = 30000;

首先要获取切分信息,代码如下,就是将特征数目featureSize 除以 并行度parallelism,然后得到了每个task对应系数的初始位置。

  1. private static int[] getSplitInfo(int featureSize,boolean hasInterceptItem,int parallelism) {
  2. int coefSize = (hasInterceptItem) ? featureSize + 1 : featureSize;
  3. int subSize = coefSize / parallelism;
  4. int[] poses = new int[parallelism + 1];
  5. int offset = coefSize % parallelism;
  6. for (int i = 0; i < offset; ++i) {
  7. poses[i + 1] = poses[i] + subSize + 1;
  8. }
  9. for (int i = offset; i < parallelism; ++i) {
  10. poses[i + 1] = poses[i] + subSize;
  11. }
  12. return poses;
  13. }
  14. //程序运行时变量如下
  15. featureSize = 30000
  16. hasInterceptItem = true
  17. parallelism = 4
  18. coefSize = 30001
  19. subSize = 7500
  20. poses = {int[5]@11660}
  21. 0 = 0
  22. 1 = 7501
  23. 2 = 15001
  24. 3 = 22501
  25. 4 = 30001
  26. offset = 1

然后根据切分信息对高维向量进行切割。

  1. // Tuple5<SampleId,taskId,numSubVec,SubVec,label>
  2. DataStream<Tuple5<Long,Integer,Vector,Object>> input
  3. = initData.flatMap(new SplitVector(splitInfo,labelIdx))
  4. .partitionCustom(new CustomBlockPartitioner(),1);

具体切分在SplitVector.flatMap函数完成,结果就是把一个高维度向量分割给各个CalcTask

代码摘要如下:

  1. public void flatMap(Row row,Collector<Tuple5<Long,Object>> collector) throws Exception {
  2. long sampleId = counter;
  3. counter += parallelism;
  4. Vector vec;
  5. if (vectorTrainIdx == -1) {
  6. .....
  7. } else {
  8. // 输入row的第vectorTrainIdx个field就是那个30000大小的系数向量
  9. vec = VectorUtil.getVector(row.getField(vectorTrainIdx));
  10. }
  11. if (vec instanceof SparseVector) {
  12. Map<Integer,Vector> tmpVec = new HashMap<>();
  13. for (int i = 0; i < indices.length; ++i) {
  14. .....
  15. // 此处迭代完成后,tmpVec中就是task number个元素,每一个元素是分割好的系数向量。
  16. }
  17. for (Integer key : tmpVec.keySet()) {
  18. //此处遍历,给后面所有CalcTask发送五元组数据。
  19. collector.collect(Tuple5.of(sampleId,key,subNum,tmpVec.get(key),row.getField(labelIdx)));
  20. }
  21. } else {
  22. ......
  23. }
  24. }
  25. }

这个Tuple5.of(sampleId,row.getField(labelIdx) )就是后面CalcTask的输入。

2.3 迭代训练

此处理论上有以下几个重点:

  • 预测方法:在每一轮t中,针对特征样本xt,以及迭代后(第一次则是给定初值)的模型参数wt,我们可以预测该样本的标记值:pt=σ(wt,xt),其中σ(a)=1/(1+exp(−a))是一个sigmoid函数

  • 损失函数:对一个特征样本xt,其对应的标记为yt ∈ 0,1,则通过 logistic loss 来作为损失函数

  • 迭代公式:我们的目的是使得损失函数尽可能的小,即可以采用极大似然估计来求解参数。首先求梯度,然后使用FTRL进行迭代。

代码思路大致如下

  1. double p = learner.predict(x); //预测
  2. learner.updateModel(x,p,y); //更新模型
  3. double loss = LogLossEvalutor.calLogLoss(p,y); //计算损失
  4. evalutor.addLogLoss(loss); //更新损失
  5. totalLoss += loss;
  6. trainedNum += 1;

具体实施上Alink有自己的特点和调整。

机器学习都需要迭代训练,Alink这里利用了Flink Stream的迭代功能

IterativeStream的实例是通过DataStream的iterate方法创建的˙。iterate方法存在两个重载形式:

  • 一种是无参的,表示不限定最大等待时间;
  • 一种提供一个长整型maxWaitTimeMillis参数,允许用户指定等待反馈边的下一个输入元素的最大时间间隔。

Alink选择了第二种。

在创建ConnectedIterativeStreams时候,用迭代流的初始输入作为第一个输入流,用反馈流作为第二个输入

每一种数据流(DataStream)都会有与之对应的流转换(StreamTransformation)。IterativeStream对应的转换是FeedbackTransformation。

迭代流(IterativeStream)对应的转换是反馈转换(FeedbackTransformation),它表示拓扑中的一个反馈点(也即迭代头)。一个反馈点包含一个输入边以及若干个反馈边,且Flink要求每个反馈边的并行度必须跟输入边的并行度一致,这一点在往该转换中加入反馈边时会进行校验。

当IterativeStream对象被构造时,FeedbackTransformation的实例会被创建并传递给DataStream的构造方法

迭代的关闭是通过调用IterativeStream的实例方法closeWith来实现的。这个函数指定了某个流将成为迭代程序的结束,并且这个流将作为输入的第二部分(second input)被反馈回迭代。

2.3.2 迭代构建

对于Alink来说,迭代构建代码是:

  1. // train data format = <sampleId,subSampleTaskId,SparseVector(subSample),label>
  2. // Feedback format = Tuple7<sampleId,label,wx,timeStamps>
  3. IterativeStream.ConnectedIterativeStreams<
  4. Tuple5<Long,Object>,Tuple7<Long,Object,Double,Long>>
  5. iteration = input.iterate(Long.MAX_VALUE)
  6. .withFeedbackType(TypeInformation
  7. .of(new TypeHint<Tuple7<Long,Long>>() {}));
  8. // 即iteration是一个 IterativeStream.ConnectedIterativeStreams<...>
2.3.2.1 迭代的输入

代码和注释可以看出,迭代的两种输入是:

  • train data format = <sampleId,label>;这种其实是训练数据
  • Tuple7<sampleId,timeStamps>;这种其实是反馈数据,就是“迭代的反馈流”作为这个第二输入 (second input);
2.3.2.2 迭代的反馈

反馈流的设置是通过调用IterativeStream的实例方法closeWith来实现的。Alink这里是

  1. DataStream<Tuple7<Long,Long>>
  2. result = iterativeBody.filter(
  3. return (t3.f0 > 0 && t3.f2 > 0); // 这里是省略版本代码
  4. );
  5. iteration.closeWith(result);

前面已经提到过,result filter 的判断是 return (t3.f0 > 0 && t3.f2 > 0)如果满足条件,则说明时间未过期&向量有意义,所以此时应该反馈回去,继续训练

反馈流的格式是:

  • Tuple7<sampleId,timeStamps>;

2.3.3 迭代体 CalcTask / ReduceTask

迭代体由两部分构成:CalcTask / ReduceTask。

CalcTask每一个实例都拥有初始化模型dataBridge

  1. DataStream iterativeBody = iteration.flatMap(
  2. new CalcTask(dataBridge,splitInfo,getParams()))
2.3.3.1 迭代初始化

迭代是由 CalcTask.open 函数开始,主要做如下几件事

  • 设定各种参数,比如
    • 工作task个数,numWorkers = getRuntimeContext().getNumberOfParallelSubtasks();
    • 本task的id,workerId = getRuntimeContext().getIndexOfThisSubtask();
  • 读取初始化模型
    • List modelRows = DirectReader.directRead(dataBridge);
    • 把Row类型数据转换为线性模型 LinearModelData model = new LinearModelDataConverter().load(modelRows);
  • 读取本task对应的系数 coef[i - startIdx],这里就是把整个模型切分到numWorkers这么多的Task中,并行更新
  • 指定本task的开始时间 startTime = System.currentTimeMillis();
2.3.3.2 处理输入数据

CalcTask.flatMap1主要实现的是FTRL算法中的predict部分(注意,不是FTRL预测)。

解释:pt=σ(Xt⋅w)是LR的预测函数,求出pt的唯一目的是为了求出目标函数(在LR中采用交叉熵损失函数作为目标函数)对参数w的一阶导数g,gi=(pt−yt)xi。此步骤同样适用于FTRL优化其他目标函数,唯一的不同就是求次梯度g(次梯度是左导和右导之间的集合,函数可导--左导等于右导时,次梯度就等于一阶梯度)的方法不同。

函数的输入是 "训练输入数据",即SplitVector.flatMap的输出 ----> CalcCalcTask的输入。输入数据是一个五元组,其格式为 train data format = <sampleId,label>;

有三点需要注意:

  • 是如果是第一次进入,则需要savedFristModel;
  • 这里是有输入就处理,然后立即输出和flatMap2不同,flatMap2有输入就处理,但不是立即输出,而是当时间到期了再输出);
  • predict的实现:((SparseVector)vec).getValues()[i] * coef[indices[i] - startIdx];

大家会说,不对!predict函数应该是 sigmoid = 1.0 / (1.0 + np.exp(-w.dot(x)))。是的,这里还没有做 sigmoid 操作。当ReduceTask做了聚合之后,会把聚合好的 p 反馈回迭代体,然后在 CalcTask.flatMap2 中才会做 sigmoid 操作

  1. public void flatMap1(Tuple5<Long,Object> value,Collector<Tuple7<Long,Long>> out) throws Exception {
  2. if (!savedFristModel) { //第一次进入需要存模型
  3. out.collect(Tuple7.of(-1L,getRuntimeContext().getIndexOfThisSubtask(),new DenseVector(coef),labelValues,-1.0,modelId++));
  4. savedFristModel = true;
  5. }
  6. Long timeStamps = System.currentTimeMillis();
  7. double wx = 0.0;
  8. Long sampleId = value.f0;
  9. Vector vec = value.f3;
  10. if (vec instanceof SparseVector) {
  11. int[] indices = ((SparseVector)vec).getIndices();
  12. // 这里就是具体的Predict
  13. for (int i = 0; i < indices.length; ++i) {
  14. wx += ((SparseVector)vec).getValues()[i] * coef[indices[i] - startIdx];
  15. }
  16. } else {
  17. ......
  18. }
  19. //处理了就输出
  20. out.collect(Tuple7.of(sampleId,value.f1,value.f2,value.f3,value.f4,timeStamps));
  21. }
2.3.3.3 归并数据

ReduceTask.flatMap 负责归并数据。

  1. public static class ReduceTask extends
  2. RichFlatMapFunction<Tuple7<Long,Long>,Long>> {
  3. private int parallelism;
  4. private int[] poses;
  5. private Map<Long,List<Object>> buffer;
  6. private Map<Long,List<Tuple2<Integer,DenseVector>>> models = new HashMap<>();
  7. }

flatMap函数大致完成如下功能,即两种归并:

  • 为了输出模型使用。判断是否时间过期 if (value.f0 < 0),如果过期,则归并模型
    • 生成一个List<Tuple2<Integer,DenseVector>> model = models.get(value.f6); 以value.f6,即时间戳为key,插入到HashMap中。
    • 如果全部收集完成,则向下游算子输出模型,并且从HashMap中删除暂存的模型。
  • 为了归并predict使用。归并每个CalcTask计算的predict,形成一个 lable y;
    • 用 label y 更新 Tuple7的f5,即Tuple7<sampleId,timeStamps> 中的 label,也就是预测的 y。
    • 给每个下游算子(就是每个CalcTask了,不过是作为flatMap2的输入)发送这个新Tuple7;

当具体用作输出模型使用时,其变量如下:

  1. models = {HashMap@13258} size = 1
  2. {Long@13456} 1 -> {ArrayList@13678} size = 1
  3. key = {Long@13456} 1
  4. value = {ArrayList@13678} size = 1
  5. 0 = {Tuple2@13698} "(1,0.0 -8.244533295515879E-5 0.0 -1.103997743166529E-4 0.0 -3.336931546279811E-5....."
2.3.3.4 判断是否反馈

这个 filter result 是用来判断是否反馈的。这里t3.f0 是sampleId,t3.f2是subNum。

  1. DataStream<Tuple7<Long,Long>>
  2. result = iterativeBody.filter(
  3. new FilterFunction<Tuple7<Long,Long>>() {
  4. @Override
  5. public boolean filter(Tuple7<Long,Long> t3)
  6. throws Exception {
  7. // if t3.f0 > 0 && t3.f2 > 0 then Feedback
  8. return (t3.f0 > 0 && t3.f2 > 0);
  9. }
  10. });

对于 t3.f0,有两处代码会设置为负值。

  • 会在savedFirstModel 这里设置一次"-1";即

    1. if (!savedFristModel) {
    2. out.collect(Tuple7.of(-1L,modelId++));
    3. savedFristModel = true;
    4. }
  • 也会在时间过期时候设置为 "-1"。

    1. if (System.currentTimeMillis() - startTime > modelSaveTimeInterval) {
    2. startTime = System.currentTimeMillis();
    3. out.collect(Tuple7.of(-1L,modelId++));
    4. }

对于 t3.f2,如果 subNum 大于零,说明在高维向量切分时候,是得到了有意义的数值。

因此 return (t3.f0 > 0 && t3.f2 > 0) 说明时间未过期&向量有意义,所以此时应该反馈回去,继续训练。

2.3.3.5 判断是否输出模型

这里是filter output。

value.f0 < 0 说明时间到期了,应该输出模型。

  1. DataStream<Row> output = iterativeBody.filter(
  2. new FilterFunction<Tuple7<Long,Long> value)
  3. {
  4. /* if value.f0 small than 0,then output */
  5. return value.f0 < 0;
  6. }
  7. }).flatMap(new WriteModel(labelType,getVectorCol(),featureCols,hasInterceptItem));
2.3.3.6 处理反馈数据/更新参数

CalcTask.flatMap2实际完成的是FTRL算法的其余部分,即更新参数部分。主要逻辑如下:

  • 计算时间间隔 timeInterval = System.currentTimeMillis() - value.f6;
  • 正式计算predict, p = 1 / (1 + Math.exp(-p)); 即sigmoid 操作;
  • 计算梯度 g = (p - label) * values[i] / Math.sqrt(timeInterval); 这里除以了时间间隔;
  • 更新参数;
  • 输入。注意,这里是有输入就处理,但 不是立即输出,而是累积参数,当时间到期了再输出,也就是做到了定期输出模型;

Logistic Regression 中,sigmoid函数是σ(a) = 1 / (1 + exp(-a)) ,预估 pt = σ(xt . wt),则 LogLoss 函数

\[l_t(w_t) = -y_t log(p_t) - (1-y_t)log(1-p_t) \]

直接计算可以得到

\[∇l(w) = (σ(w.x_t) - y_t)x_t = (p_t - y_t)x_t \]

具体 LR + FTRL 算法实现如下:

  1. @Override
  2. public void flatMap2(Tuple7<Long,Long> value,Long>> out)
  3. throws Exception {
  4. double p = value.f5;
  5. // 计算时间间隔
  6. long timeInterval = System.currentTimeMillis() - value.f6;
  7. Vector vec = value.f3;
  8. /* eta */
  9. // 正式计算predict,之前只是计算了一半,这里计算后半部,即
  10. p = 1 / (1 + Math.exp(-p));
  11. .....
  12. if (vec instanceof SparseVector) {
  13. // 这里是更新参数
  14. int[] indices = ((SparseVector)vec).getIndices();
  15. double[] values = ((SparseVector)vec).getValues();
  16. for (int i = 0; i < indices.length; ++i) {
  17. // update zParam nParam
  18. int id = indices[i] - startIdx;
  19. // values[i]是xi
  20. // 下面的计算基本和Google伪代码一致
  21. double g = (p - label) * values[i] / Math.sqrt(timeInterval);
  22. double sigma = (Math.sqrt(nParam[id] + g * g) - Math.sqrt(nParam[id])) / alpha;
  23. zParam[id] += g - sigma * coef[id];
  24. nParam[id] += g * g;
  25. // update model coefficient
  26. if (Math.abs(zParam[id]) <= l1) {
  27. coef[id] = 0.0;
  28. } else {
  29. coef[id] = ((zParam[id] < 0 ? -1 : 1) * l1 - zParam[id])
  30. / ((beta + Math.sqrt(nParam[id]) / alpha + l2));
  31. }
  32. }
  33. } else {
  34. ......
  35. }
  36. // 当时间到期了再输出,即做到了定期输出模型
  37. if (System.currentTimeMillis() - startTime > modelSaveTimeInterval) {
  38. startTime = System.currentTimeMillis();
  39. out.collect(Tuple7.of(-1L,modelId++));
  40. }
  41. }

2.4 输出模型

WriteModel 类实现了输出模型功能,大致逻辑如下:

  • 生成一个LinearModelData,用训练好的Tuple7来填充这个 LinearModelData。其中两个重要点:
    • modelData.coefVector = (DenseVector)value.f3;
    • modelData.labelValues = (Object[])value.f4;
  • 把模型数据转换成List rows。LinearModelDataConverter().save(modelData,listCollector);
  • 序列化,发送给下游算子。因为模型可能会很大,所以这里打散之后分布发送给下游算子
  1. public void flatMap(Tuple7<Long,Collector<Row> out){
  2. //输入value变量打印如下:
  3. value = {Tuple7@13296}
  4. f0 = {Long@13306} -1
  5. f1 = {Integer@13307} 0
  6. f2 = {Integer@13308} 2
  7. f3 = {DenseVector@13309} "-0.7383426732137565 0.0 0.0 0.0 1.5885293675862715E-4 -4.834608575902742E-5 0.0 0.0 -6.754208708318647E-5 ......"
  8. data = {double[30001]@13314}
  9. f4 = {Object[2]@13310}
  10. f5 = {Double@13311} -1.0
  11. f6 = {Long@13312} 0
  12. //生成模型
  13. LinearModelData modelData = new LinearModelData();
  14. ......
  15. modelData.coefVector = (DenseVector)value.f3;
  16. modelData.labelValues = (Object[])value.f4;
  17. //把模型数据转换成List<Row> rows
  18. RowCollector listCollector = new RowCollector();
  19. new LinearModelDataConverter().save(modelData,listCollector);
  20. List<Row> rows = listCollector.getRows();
  21. for (Row r : rows) {
  22. int rowSize = r.getArity();
  23. for (int j = 0; j < rowSize; ++j) {
  24. .....
  25. //序列化
  26. }
  27. out.collect(row);
  28. }
  29. iter++;
  30. }
  31. }

0x03 在线预测

预测功能是在 FtrlPredictStreamOp 完成的。

  1. // ftrl predict
  2. FtrlPredictStreamOp predictResult = new FtrlPredictStreamOp(initModel)
  3. .setVectorCol(vecColName)
  4. .setPredictionCol("pred")
  5. .setReservedCols(new String[]{labelColName})
  6. .setPredictionDetailCol("details")
  7. .linkFrom(model,featurePipelineModel.transform(splitter.getSideOutput(0)));

从上面代码我们可以看到

  • FtrlPredict 功能同样需要初始模型 initModel,我们也是把逻辑回归模型赋予它。这样也是为了冷启动,即当FTRL训练模块还没有产生模型之前,FTRL预测模块也是可以对其输入数据做预测的。
  • model 是 FtrlTrainStreamOp 的输出,即 FTRL 的训练输出。所以 WriteModel 就直接把输出传给了 FtrlPredict功能
  • splitter.getSideOutput(0) 这里是前面提到的测试输入,就是测试数据集。

linkFrom函数完成了业务逻辑,大致功能如下:

  • 使用 inputs[0].getDataStream().flatMap ------> partition ----> map ----> flatMap(new CollectModel()) 得到了模型 LinearModelData modelstr;
  • 使用 DataStream.connect 把输入的测试数据集 和 模型 LinearModelData modelstr关联起来,这样每个task都拥有了在线模型 modelstr,就可以通过 flatMap(new PredictProcess(...) 进行分布式预测;
  • 使用 setOutputTable 和 LinearModelMapper 把预测结果输出

FTRL的预测功能有三个输入

  • 初始模型 initModel ----->  最后被 PredictProcess.open 加载,作为冷启动的预测模型;
  • 测试数据流 -----> 被 PredictProcess.flatMap1处理,进行预测;
  • FTRL训练阶段产生的模型数据流 ----> 被 PredictProcess.flatMap2 处理,进行在线模型更新;

3.1 初始化

构造函数中完成了初始化,即获取事先训练好的逻辑回归模型

  1. public FtrlPredictStreamOp(BatchOperator model) {
  2. super(new Params());
  3. if (model != null) {
  4. dataBridge = DirectReader.collect(model);
  5. } else {
  6. throw new IllegalArgumentException("Ftrl algo: initial model is null. Please set a valid initial model.");
  7. }
  8. }

3.2 获取在线训练模型

CollectModel完成了 获取在线训练模型 功能

其逻辑主要是:模型被分成若干块,其中 (long)inRow.getField(1) 这里记录了具体有多少块。所以 flatMap 函数会把这些块累积起来,最后组装成模型,统一发送给下游算子

具体是通过一个 HashMap<> buffers 来完成临时拼装/最后组装的。

  1. public static class CollectModel implements FlatMapFunction<Row,LinearModelData> {
  2. private Map<Long,List<Row>> buffers = new HashMap<>(0);
  3. @Override
  4. public void flatMap(Row inRow,Collector<LinearModelData> out) throws Exception {
  5. // 输入参数如下
  6. inRow = {Row@13389} "0,19,{"hasInterceptItem":"true","vectorCol":"\"vec\"","modelName":"\"Logistic Regression\"","labelCol":null,"linearModelType":"\"LR\"","vectorSize":"30000"},null"
  7. fields = {Object[5]@13405}
  8. 0 = {Long@13406} 0
  9. 1 = {Long@13403} 19
  10. 2 = {Long@13406} 0
  11. 3 = "{"hasInterceptItem":"true","vectorSize":"30000"}"
  12. "
  13. long id = (long)inRow.getField(0);
  14. Long nTab = (long)inRow.getField(1);
  15. Row row = new Row(inRow.getArity() - 2);
  16. for (int i = 0; i < row.getArity(); ++i) {
  17. row.setField(i,inRow.getField(i + 2));
  18. }
  19. if (buffers.containsKey(id) && buffers.get(id).size() == nTab.intValue() - 1) {
  20. buffers.get(id).add(row);
  21. // 如果累积完成,则组装成模型
  22. LinearModelData ret = new LinearModelDataConverter().load(buffers.get(id));
  23. buffers.get(id).clear();
  24. // 发送给下游算子。
  25. out.collect(ret);
  26. } else {
  27. if (buffers.containsKey(id)) {
  28. //如果有key。则往list添加
  29. buffers.get(id).add(row);
  30. } else {
  31. // 如果没有key,则添加list
  32. List<Row> buffer = new ArrayList<>(0);
  33. buffer.add(row);
  34. buffers.put(id,buffer);
  35. }
  36. }
  37. }
  38. }
  39. //变量类似这种
  40. this = {FtrlPredictStreamOp$CollectModel@13388}
  41. buffers = {HashMap@13393} size = 1
  42. {Long@13406} 0 -> {ArrayList@13431} size = 2
  43. key = {Long@13406} 0
  44. value = 0
  45. value = {ArrayList@13431} size = 2
  46. 0 = {Row@13409} "0,null"
  47. 1 = {Row@13471} "1048576,{"featureColNames":null,"featureColTypes":null,"coefVector":{"data":[-0.7383426732137549,0.0,1.5885293675862704E-4,-4.834608575902738E-5,-6.754208708318643E-5,-1.5904172331763155E-4,-1.315219790338925E-4,-4.994749246390495E-4,2.755456604395511E-4,-9.616429481614131E-4,-9.601054004112163E-5,-1.6679174640370486E-4,......"

3.3 在线预测

PredictProcess 完成了在线预测功能,LinearModelMapper 是具体预测实现。

  1. public static class PredictProcess extends RichCoFlatMapFunction<Row,LinearModelData,Row> {
  2. private LinearModelMapper predictor = null;
  3. private String modelSchemaJson;
  4. private String dataSchemaJson;
  5. private Params params;
  6. private int iter = 0;
  7. private DataBridge dataBridge;
  8. }

3.3.1 加载预设置模型

其构造函数获得了 FtrlPredictStreamOp 类的 dataBridge,即事先训练好的逻辑回归模型。每一个Task都拥有完整的模型。

open函数会加载逻辑回归模型

  1. public void open(Configuration parameters) throws Exception {
  2. this.predictor = new LinearModelMapper(TableUtil.fromSchemaJson(modelSchemaJson),TableUtil.fromSchemaJson(dataSchemaJson),this.params);
  3. if (dataBridge != null) {
  4. // read init model
  5. List<Row> modelRows = DirectReader.directRead(dataBridge);
  6. LinearModelData model = new LinearModelDataConverter().load(modelRows);
  7. this.predictor.loadModel(model);
  8. }
  9. }

3.3.2 在线预测

FtrlPredictStreamOp.flatMap1 函数完成了在线预测。

  1. public void flatMap1(Row row,Collector<Row> collector) throws Exception {
  2. collector.collect(this.predictor.map(row));
  3. }

调用栈如下:

  1. predictWithProb:157,LinearModelMapper (com.alibaba.alink.operator.common.linear)
  2. predictResultDetail:114,LinearModelMapper (com.alibaba.alink.operator.common.linear)
  3. map:90,RichModelMapper (com.alibaba.alink.common.mapper)
  4. flatMap1:174,FtrlPredictStreamOp$PredictProcess (com.alibaba.alink.operator.stream.onlinelearning)
  5. flatMap1:143,FtrlPredictStreamOp$PredictProcess (com.alibaba.alink.operator.stream.onlinelearning)
  6. processElement1:53,CoStreamFlatMap (org.apache.flink.streaming.api.operators.co)
  7. processRecord1:135,StreamTwoInputProcessor (org.apache.flink.streaming.runtime.io)

具体是通过 LinearModelMapper 完成。

  1. public abstract class RichModelMapper extends ModelMapper {
  2. public Row map(Row row) throws Exception {
  3. if (isPredDetail) {
  4. // 我们的示例代码在这里
  5. Tuple2<Object,String> t2 = predictResultDetail(row);
  6. return this.outputColsHelper.getResultRow(row,Row.of(t2.f0,t2.f1));
  7. } else {
  8. return this.outputColsHelper.getResultRow(row,Row.of(predictResult(row)));
  9. }
  10. }
  11. }

预测代码如下,可以看出来使用了sigmoid。

  1. /**
  2. * Predict the label information with the probability of each label.
  3. */
  4. public Tuple2 <Object,Double[]> predictWithProb(Vector vector) {
  5. double dotValue = MatVecOp.dot(vector,model.coefVector);
  6. switch (model.linearModelType) {
  7. case LR:
  8. case SVM:
  9. double prob = sigmoid(dotValue);
  10. return new Tuple2 <>(dotValue >= 0 ? model.labelValues[0] : model.labelValues[1],new Double[] {prob,1 - prob});
  11. }
  12. }

3.3.3 在线更新模型

FtrlPredictStreamOp.flatMap2 函数完成了处理在线训练输出的模型数据流,在线更新模型。

LinearModelData参数是由CollectModel完成加载并且传输出来的。

在模型加载过程中,是不能预测的,没有看到相关保护机制。如果我疏漏请大家指出。

  1. public void flatMap2(LinearModelData linearModel,Collector<Row> collector) throws Exception {
  2. this.predictor.loadModel(linearModel);
  3. }

0x04 问题解答

针对之前我们提出的问题,现在总结归纳如下:

  • 训练阶段和预测阶段都有预制模型以应对"冷启动"嘛?都有预制模型
  • 训练阶段和预测阶段是如何关联起来的?用 linkFrom 直接把训练阶段和预测阶段的算子连在一起
  • 如何把训练出来的模型传给预测阶段?训练阶段用 Flink collector.collect 把模型发给下游算子
  • 输出模型时候,模型过大怎么处理?在线训练会 模型打散 之后分布发送给下游算子
  • 在线训练的模型通过什么机制实现更新?是定时驱动更新嘛?定时更新
  • 预测阶段加载模型过程中,还可以预测嘛?有没有机制保证这段时间内也能预测?目前没有发现类似保护机制
  • 训练阶段中,有哪些阶段用到了并行处理?训练过程中主要是FTRL算法的"预测predict" 和 "更新参数"两个部分,以及发送模型
  • 预测阶段中,有哪些阶段用到了并行处理?预测过程中主要是分布式接受模型和分布式预测
  • 遇到高维向量如何处理?切分开嘛?切分处理

0xFF 参考

【机器学习】逻辑回归(非常详细)

逻辑回归(logistics regression)

【机器学习】LR的分布式(并行化)实现

并行逻辑回归

机器学习算法及其并行化讨论

Online LR—— FTRL 算法理解

在线优化算法 FTRL 的原理与实现

LR+FTRL算法原理以及工程化实现

Flink流处理之迭代API分析

FTRL公式推导

FTRL论文笔记

在线机器学习FTRL(Follow-the-regularized-Leader)算法介绍

FTRL代码实现

FTRL实战之LR+FTRL(代码采用的稠密数据)

在线学习算法FTRL-Proximal原理

基于FTRL的在线CTR预测算法

CTR预测算法之FTRL-Proximal

各大公司广泛使用的在线学习算法FTRL详解

在线最优化求解(Online Optimization)之五:FTRL

FOLLOW THE REGULARIZED LEADER (FTRL) 算法总结

猜你在找的大数据相关文章