在线FM训练 (OnlineFmTrainStreamOp)

Java 类名：com.alibaba.alink.operator.stream.onlinelearning.OnlineFmTrainStreamOp

Python 类名：OnlineFmTrainStreamOp

功能介绍

OnlineFm 算法是Ftrl算法的升级，在 Ftrl 算法的基础上考虑了二阶项对模型的影响。算法支持稀疏和稠密两种类型的训练数据。

算法原理

FM模型是线性模型的升级，是在线性表达式后面加入了新的交叉项特征及对应的权值，FM模型的表达式如下所示：

这里我们使用 Ftrl 优化算法求解该模型。算法原理细节可以参考文献[1]，优化算法请参考文献[2]。

算法使用

FM算法是推荐领域被验证的效果较好的推荐方案之一，在电商、广告、视频、信息流、游戏的推荐领域有广泛应用。

• 备注 ：该组件训练的时候 FeatureCols 和 VectorCol 是两个互斥参数，只能有一个参数来描述算法的输入特征。

文献

[1] S. Rendle, “Factorization Machines,” 2010 IEEE International Conference on Data Mining, 2010, pp. 995-1000, doi: 10.1109/ICDM.2010.127.

[2] McMahan, H. Brendan, et al. “Ad click prediction: a view from the trenches.” Proceedings of the 19th ACM SIGKDD international conference on Knowledge discovery and data mining. 2013.

参数说明

代码示例

以下代码仅用于示意，可能需要修改部分代码或者配置环境后才能正常运行！

Python 代码

trainData0 = RandomTableSourceBatchOp() \
            .setNumCols(5) \
            .setNumRows(100) \
            .setOutputCols(["f0", "f1", "f2", "f3", "label"]) \
            .setOutputColConfs("label:weight_set(1.0,1.0,2.0,5.0)")
model = FmClassifierTrainBatchOp() \
            .setFeatureCols(["f0", "f1", "f2", "f3"]) \
            .setNumEpochs(1) \
            .setWithIntercept(true) \
            .setWithLinearItem(true) \
            .setNumFactor(10) \
            .setLabelCol("label").linkFrom(trainData0)
trainData1 = RandomTableSourceStreamOp() \
            .setNumCols(5) \
            .setMaxRows(10000) \
            .setOutputCols(["f0", "f1", "f2", "f3", "label"]) \
            .setOutputColConfs("label:weight_set(1.0,1.0,2.0,5.0)") \
            .setTimePerSample(0.1)
smodel = OnlineFmTrainStreamOp(model)\
            .setFeatureCols(["f0", "f1", "f2", "f3"])\
            .setLabelCol("label")\
            .setTimeInterval(5)\
            .setAlpha(0.1)\
            .setBeta(0.1)\
            .setL1(1.0e-4)\
            .setL2(1.0e-4)\
            .setWithIntercept(true)\
            .setWithLinearItem(true)\
            .setNumFactor(10)\
            .setMiniBatchSize(4)\
            .linkFrom(trainData1)
smodel.print()
StreamOperator.execute()

Java 代码

package com.alibaba.alink.operator.stream.ml.onlinelearning;
import com.alibaba.alink.operator.batch.BatchOperator;
import com.alibaba.alink.operator.batch.classification.LogisticRegressionTrainBatchOp;
import com.alibaba.alink.operator.batch.source.RandomTableSourceBatchOp;
import com.alibaba.alink.operator.stream.StreamOperator;
import com.alibaba.alink.operator.stream.onlinelearning.FtrlTrainStreamOp;
import com.alibaba.alink.operator.stream.source.RandomTableSourceStreamOp;
import org.junit.Test;
public class FtrlTrainTestTest {
    @Test
    public void FtrlClassification() throws Exception {
        StreamOperator.setParallelism(2);
        BatchOperator trainData0 = new RandomTableSourceBatchOp()
            .setNumCols(5)
            .setNumRows(100L)
            .setOutputCols(new String[]{"f0", "f1", "f2", "f3", "label"})
            .setOutputColConfs("label:weight_set(1.0,1.0,2.0,5.0)");
        BatchOperator model = new FmClassifierTrainBatchOp()
            .setFeatureCols(new String[]{"f0", "f1", "f2", "f3"})
            .setNumEpochs(1)
            .setWithIntercept(true)
            .setWithLinearItem(true)
            .setNumFactor(10)
            .setLabelCol("label").linkFrom(trainData0);
        StreamOperator trainData1 = new RandomTableSourceStreamOp()
            .setNumCols(5)
            .setMaxRows(100L)
            .setOutputCols(new String[]{"f0", "f1", "f2", "f3", "label"})
            .setOutputColConfs("label:weight_set(1.0,1.0,2.0,5.0)")
            .setTimePerSample(0.1);
        StreamOperator smodel = new OnlineFmTrainStreamOp(model)
            .setFeatureCols(new String[]{"f0", "f1", "f2", "f3"})
            .setLabelCol("label")
            .setTimeInterval(5)
            .setAlpha(0.1)
            .setBeta(0.1)
            .setL1(1.0e-4)
            .setL2(1.0e-4)
            .setWithIntercept(true)
            .setWithLinearItem(true)
            .setNumFactor(10)
            .setMiniBatchSize(4)
            .linkFrom(trainData1);
        smodel.print();
        StreamOperator.execute();
    }
}

运行结果