Java 类名：com.alibaba.alink.operator.stream.onlinelearning.FtrlPredictStreamOp
Python 类名：FtrlPredictStreamOp

功能介绍

实时更新ftrl 训练得到的模型流，并使用实时的模型对实时的数据进行预测。

参数说明

| 名称 | 中文名称 | 描述 | 类型 | 是否必须？ | 取值范围 | 默认值 | | —- | —- | —- | —- | —- | —- | —- |

| predictionCol | 预测结果列名 | 预测结果列名 | String | ✓ | | |

| predictionDetailCol | 预测详细信息列名 | 预测详细信息列名 | String | | | |

| reservedCols | 算法保留列名 | 算法保留列 | String[] | | | null |

| vectorCol | 向量列名 | 向量列对应的列名，默认值是null | String | | 所选列类型为 [DENSE_VECTOR, SPARSE_VECTOR, STRING, VECTOR] | null |

| numThreads | 组件多线程线程个数 | 组件多线程线程个数 | Integer | | | 1 |

代码示例

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

Python 代码

  1. trainData0 = RandomTableSourceBatchOp() \
  2.             .setNumCols(5) \
  3.             .setNumRows(100) \
  4.             .setOutputCols(["f0", "f1", "f2", "f3", "label"]) \
  5.             .setOutputColConfs("label:weight_set(1.0,1.0,2.0,5.0)")
  7. model = LogisticRegressionTrainBatchOp() \
  8.             .setFeatureCols(["f0", "f1", "f2", "f3"]) \
  9.             .setLabelCol("label") \
  10.             .setMaxIter(10).linkFrom(trainData0)
  12. trainData1 = RandomTableSourceStreamOp() \
  13.             .setNumCols(5) \
  14.             .setMaxRows(10000) \
  15.             .setOutputCols(["f0", "f1", "f2", "f3", "label"]) \
  16.             .setOutputColConfs("label:weight_set(1.0,1.0,2.0,5.0)") \
  17.             .setTimePerSample(0.1)
  19. models = FtrlTrainStreamOp(model, None) \
  20.             .setFeatureCols(["f0", "f1", "f2", "f3"]) \
  21.             .setLabelCol("label") \
  22.             .setTimeInterval(10) \
  23.             .setAlpha(0.1) \
  24.             .setBeta(0.1) \
  25.             .setL1(0.1) \
  26.             .setL2(0.1)\
  27.             .setVectorSize(4)\
  28.             .setWithIntercept(True) \
  29.             .linkFrom(trainData1)
  31. FtrlPredictStreamOp(model) \
  32.         .setPredictionCol("pred") \
  33.         .setReservedCols(["label"]) \
  34.         .setPredictionDetailCol("details") \
  35.         .linkFrom(models, trainData1).print()
  36. StreamOperator.execute()

Java 代码

  1. package com.alibaba.alink.operator.stream.ml.onlinelearning;
  3. import com.alibaba.alink.operator.batch.BatchOperator;
  4. import com.alibaba.alink.operator.batch.classification.LogisticRegressionTrainBatchOp;
  5. import com.alibaba.alink.operator.batch.source.RandomTableSourceBatchOp;
  6. import com.alibaba.alink.operator.stream.StreamOperator;
  7. import com.alibaba.alink.operator.stream.onlinelearning.FtrlPredictStreamOp;
  8. import com.alibaba.alink.operator.stream.onlinelearning.FtrlTrainStreamOp;
  9. import com.alibaba.alink.operator.stream.source.RandomTableSourceStreamOp;
  10. import org.junit.Test;
  12. public class FtrlTrainTestTest {
  13.     @Test
  14.     public void FtrlClassification() throws Exception {
  15.         StreamOperator.setParallelism(2);
  16.         BatchOperator trainData0 = new RandomTableSourceBatchOp()
  17.             .setNumCols(5)
  18.             .setNumRows(100L)
  19.             .setOutputCols(new String[]{"f0", "f1", "f2", "f3", "label"})
  20.             .setOutputColConfs("label:weight_set(1.0,1.0,2.0,5.0)");
  22.         BatchOperator model = new LogisticRegressionTrainBatchOp()
  23.             .setFeatureCols(new String[]{"f0", "f1", "f2", "f3"})
  24.             .setLabelCol("label")
  25.             .setMaxIter(10).linkFrom(trainData0);
  27.         StreamOperator trainData1 = new RandomTableSourceStreamOp()
  28.             .setNumCols(5)
  29.             .setMaxRows(1000L)
  30.             .setOutputCols(new String[]{"f0", "f1", "f2", "f3", "label"})
  31.             .setOutputColConfs("label:weight_set(1.0,1.0,2.0,5.0)")
  32.             .setTimePerSample(0.1);
  34.         StreamOperator smodel = new FtrlTrainStreamOp(model)
  35.             .setFeatureCols(new String[]{"f0", "f1", "f2", "f3"})
  36.             .setLabelCol("label")
  37.             .setTimeInterval(10)
  38.             .setAlpha(0.1)
  39.             .setBeta(0.1)
  40.             .setL1(0.1)
  41.             .setL2(0.1)
  42.             .setVectorSize(4)
  43.             .setWithIntercept(true)
  44.             .linkFrom(trainData1);
  46.         new FtrlPredictStreamOp(model)
  47.         .setPredictionCol("pred")
  48.         .setReservedCols(new String[]{"label"})
  49.         .setPredictionDetailCol("details")
  50.         .linkFrom(smodel, trainData1).print();
  51.         StreamOperator.execute();
  52.     }
  53. }

运行结果

| label | pred | details | | —- | —- | —- |

| 2.0000 | 2.0000 | {“2.0”:”0.8407811313273308”,”1.0”:”0.1592188686726692”} |

| 2.0000 | 2.0000 | {“2.0”:”0.8094960632541983”,”1.0”:”0.19050393674580168”} |

| 2.0000 | 2.0000 | {“2.0”:”0.8685396820088952”,”1.0”:”0.1314603179911048”} |

| 2.0000 | 2.0000 | {“2.0”:”0.781050184076571”,”1.0”:”0.218949815923429”} |

| 1.0000 | 2.0000 | {“2.0”:”0.8347637657816113”,”1.0”:”0.16523623421838873”} |

| 2.0000 | 2.0000 | {“2.0”:”0.9211808843291631”,”1.0”:”0.07881911567083688”} |