Demo code

  1. package benchmark.online;
  3. import org.apache.flink.api.common.typeinfo.TypeInformation;
  4. import org.apache.flink.api.java.tuple.Tuple2;
  5. import org.apache.flink.api.java.typeutils.RowTypeInfo;
  6. import org.apache.flink.table.functions.TableFunction;
  7. import org.apache.flink.types.Row;
  9. import com.alibaba.alink.common.AlinkGlobalConfiguration;
  10. import com.alibaba.alink.common.AlinkTypes;
  11. import com.alibaba.alink.common.MLEnvironmentFactory;
  12. import com.alibaba.alink.common.utils.TableUtil;
  13. import com.alibaba.alink.operator.batch.BatchOperator;
  14. import com.alibaba.alink.operator.batch.classification.LogisticRegressionTrainBatchOp;
  15. import com.alibaba.alink.operator.batch.sink.AkSinkBatchOp;
  16. import com.alibaba.alink.operator.batch.sink.AppendModelStreamFileSinkBatchOp;
  17. import com.alibaba.alink.operator.batch.source.CsvSourceBatchOp;
  18. import com.alibaba.alink.operator.stream.StreamOperator;
  19. import com.alibaba.alink.operator.stream.classification.LogisticRegressionPredictStreamOp;
  20. import com.alibaba.alink.operator.stream.dataproc.JsonValueStreamOp;
  21. import com.alibaba.alink.operator.stream.dataproc.SplitStreamOp;
  22. import com.alibaba.alink.operator.stream.evaluation.EvalBinaryClassStreamOp;
  23. import com.alibaba.alink.operator.stream.onlinelearning.FtrlTrainStreamOp;
  24. import com.alibaba.alink.operator.stream.sink.ModelStreamFileSinkStreamOp;
  25. import com.alibaba.alink.operator.stream.source.CsvSourceStreamOp;
  26. import com.alibaba.alink.pipeline.LocalPredictor;
  27. import com.alibaba.alink.pipeline.Pipeline;
  28. import com.alibaba.alink.pipeline.PipelineModel;
  29. import com.alibaba.alink.pipeline.classification.LogisticRegression;
  30. import com.alibaba.alink.pipeline.dataproc.StandardScaler;
  31. import com.alibaba.alink.pipeline.feature.FeatureHasher;
  32. import org.apache.commons.lang3.ArrayUtils;
  33. import org.junit.Test;
  35. /**
  36.  * https://www.kaggle.com/c/avazu-ctr-prediction/data
  37.  */
  38. public class FtrlTest {
  40.     private static final String[] ORIGIN_COL_NAMES = new String[] {
  41.         "id", "click", "dt", "C1", "banner_pos",
  42.         "site_id", "site_domain", "site_category", "app_id", "app_domain",
  43.         "app_category", "device_id", "device_ip", "device_model", "device_type",
  44.         "device_conn_type", "C14", "C15", "C16", "C17",
  45.         "C18", "C19", "C20", "C21"
  46.     };
  48.     private static final String[] ORIGIN_COL_TYPES = new String[] {
  49.         "string", "string", "string", "string", "int",
  50.         "string", "string", "string", "string", "string",
  51.         "string", "string", "string", "string", "string",
  52.         "string", "int", "int", "int", "int",
  53.         "int", "int", "int", "int"
  54.     };
  56.     private static final String[] COL_NAMES = new String[] {
  57.         "id", "click",
  58.         "dt_year", "dt_month", "dt_day", "dt_hour",
  59.         "C1", "banner_pos",
  60.         "site_id", "site_domain", "site_category", "app_id", "app_domain",
  61.         "app_category", "device_id", "device_ip", "device_model", "device_type",
  62.         "device_conn_type", "C14", "C15", "C16", "C17",
  63.         "C18", "C19", "C20", "C21"
  64.     };
  66.     private static final String DATA_DIR = "https://alink-release.oss-cn-beijing.aliyuncs.com/data-files/";
  67.     private static final String SMALL_FILE = "avazu-small.csv";
  68.     private static final String LARGE_FILE = "avazu-ctr-train-8M.csv";
  70.     private static final String FEATURE_PIPELINE_MODEL_FILE = "/tmp/feature_model.csv";
  72.     private static final String labelColName = "click";
  73.     private static final String vecColName = "vec";
  75.     static final String[] FEATURE_COL_NAMES =
  76.         ArrayUtils.removeElements(COL_NAMES, labelColName, "id", "dt_year", "dt_month",
  77.             "site_id", "site_domain", "app_id", "device_id", "device_ip", "device_model");
  79.     static final String[] HIGH_FREQ_FEATURE_COL_NAMES = new String[] {"site_id", "site_domain", "device_id",
  80.         "device_model"};
  82.     static final String[] CATEGORY_FEATURE_COL_NAMES = new String[] {
  83.         "C1", "banner_pos",
  84.         "site_category", "app_domain",
  85.         "app_category", "device_type",
  86.         "device_conn_type"
  87.     };
  89.     static final String[] NUMERICAL_FEATURE_COL_NAMES =
  90.         ArrayUtils.removeElements(FEATURE_COL_NAMES, CATEGORY_FEATURE_COL_NAMES);
  92.     @Test
  93.     public void trainFeatureModel() throws Exception {
  94.         MLEnvironmentFactory.getDefault().getStreamExecutionEnvironment().setParallelism(4);
  95.         MLEnvironmentFactory.getDefault().getExecutionEnvironment().setParallelism(1);
  97.         int numHashFeatures = 30000;
  98.         Pipeline feature_pipeline = new Pipeline()
  99.             .add(
  100.                 new StandardScaler()
  101.                     .setSelectedCols(NUMERICAL_FEATURE_COL_NAMES)
  102.             )
  103.             .add(
  104.                 new FeatureHasher()
  105.                     .setSelectedCols(ArrayUtils.addAll(FEATURE_COL_NAMES, HIGH_FREQ_FEATURE_COL_NAMES))
  106.                     .setCategoricalCols(ArrayUtils.addAll(CATEGORY_FEATURE_COL_NAMES, HIGH_FREQ_FEATURE_COL_NAMES))
  107.                     .setOutputCol(vecColName)
  108.                     .setNumFeatures(numHashFeatures).setReservedCols("click")
  109.             );
  111.         feature_pipeline.fit(getSmallBatchSet()).save(FEATURE_PIPELINE_MODEL_FILE,
  112.             true);
  113.         BatchOperator.execute();
  114.     }
  116.     @Test
  117.     public void onlineTrainAndEval() throws Exception {
  118.         PipelineModel featurePipelineModel = PipelineModel.load(FEATURE_PIPELINE_MODEL_FILE);
  119.         AlinkGlobalConfiguration.setPrintProcessInfo(true);
  120.         Tuple2 <StreamOperator, StreamOperator> sources = getStreamTrainTestData();
  121.         StreamOperator <?> trainStream = sources.f0;
  122.         StreamOperator <?> testStream = sources.f1;
  124.         trainStream = featurePipelineModel.transform(trainStream);
  125.         testStream = featurePipelineModel.transform(testStream);
  127.         BatchOperator <?> trainBatch = featurePipelineModel.transform(getSmallBatchSet());
  129.         StreamOperator.setParallelism(2);
  131.         BatchOperator <?> model = new LogisticRegressionTrainBatchOp()
  132.             .setVectorCol(vecColName)
  133.             .setLabelCol(labelColName)
  134.             .setWithIntercept(true)
  135.             .linkFrom(trainBatch);
  137.         StreamOperator <?> models = new FtrlTrainStreamOp(model)
  138.             .setVectorCol(vecColName)
  139.             .setLabelCol(labelColName)
  140.             .setMiniBatchSize(1024)
  141.             .setTimeInterval(10)
  142.             .setWithIntercept(true)
  143.             .setModelStreamFilePath("/tmp/avazu_fm_models")
  144.             .linkFrom(trainStream);
  146.         StreamOperator <?> predictResults = new LogisticRegressionPredictStreamOp(model)
  147.             .setPredictionCol("predict")
  148.             .setReservedCols(labelColName)
  149.             .setPredictionDetailCol("details")
  150.             .linkFrom(testStream, models);
  152.         new EvalBinaryClassStreamOp()
  153.             .setPredictionDetailCol("details").setLabelCol(labelColName).setTimeInterval(10).linkFrom(predictResults)
  154.             .link(new JsonValueStreamOp().setSelectedCol("Data")
  155.                 .setReservedCols(new String[] {"Statistics"})
  156.                 .setOutputCols(new String[] {"Accuracy", "AUC", "ConfusionMatrix"})
  157.                 .setJsonPath("$.Accuracy", "$.AUC", "ConfusionMatrix")).print();
  159.         StreamOperator.execute();
  160.     }
  162.     @Test
  163.     public void onlineTrainAndSave() throws Exception {
  164.         PipelineModel featurePipelineModel = PipelineModel.load(FEATURE_PIPELINE_MODEL_FILE);
  165.         AlinkGlobalConfiguration.setPrintProcessInfo(true);
  166.         Tuple2 <StreamOperator, StreamOperator> sources = getStreamTrainTestData();
  167.         StreamOperator <?> trainStream = sources.f0;
  169.         trainStream = featurePipelineModel.transform(trainStream);
  171.         BatchOperator <?> trainBatch = featurePipelineModel.transform(getSmallBatchSet());
  173.         StreamOperator.setParallelism(2);
  175.         BatchOperator <?> model = new LogisticRegressionTrainBatchOp()
  176.             .setVectorCol(vecColName)
  177.             .setLabelCol(labelColName)
  178.             .setWithIntercept(true)
  179.             .linkFrom(trainBatch);
  181.         StreamOperator <?> models = new FtrlTrainStreamOp(model)
  182.             .setVectorCol(vecColName)
  183.             .setLabelCol(labelColName)
  184.             .setMiniBatchSize(1024)
  185.             .setTimeInterval(10)
  186.             .setWithIntercept(true)
  187.             .setModelStreamFilePath("/tmp/rebase_ftrl_models")
  188.             .linkFrom(trainStream);
  190.         models.link(new ModelStreamFileSinkStreamOp().setFilePath("/tmp/ftrl_models"));
  191.         StreamOperator.execute();
  192.     }
  194.     @Test
  195.     public void BatchTrainAndSaveRebaseModel() throws Exception {
  196.         PipelineModel featurePipelineModel = PipelineModel.load(FEATURE_PIPELINE_MODEL_FILE);
  197.         BatchOperator <?> trainBatch = featurePipelineModel.transform(getSmallBatchSet());
  198.         StreamOperator.setParallelism(2);
  199.         BatchOperator <?> model1 = new LogisticRegressionTrainBatchOp()
  200.             .setVectorCol(vecColName)
  201.             .setLabelCol(labelColName)
  202.             .setWithIntercept(true)
  203.             .linkFrom(trainBatch);
  204.         model1.link(new AppendModelStreamFileSinkBatchOp().setFilePath("/tmp/rebase_ftrl_models"));
  205.         BatchOperator.execute();
  206.     }
  208.     @Test
  209.     public void savePipelineModel() throws Exception {
  210.         BatchOperator <?> trainBatch = getSmallBatchSet();
  211.         int numHashFeatures = 30000;
  212.         PipelineModel pipelineModel = new Pipeline()
  213.             .add(
  214.                 new StandardScaler()
  215.                     .setSelectedCols(NUMERICAL_FEATURE_COL_NAMES)
  216.             )
  217.             .add(
  218.                 new FeatureHasher()
  219.                     .setSelectedCols(ArrayUtils.addAll(FEATURE_COL_NAMES, HIGH_FREQ_FEATURE_COL_NAMES))
  220.                     .setCategoricalCols(ArrayUtils.addAll(CATEGORY_FEATURE_COL_NAMES, HIGH_FREQ_FEATURE_COL_NAMES))
  221.                     .setOutputCol(vecColName)
  222.                     .setNumFeatures(numHashFeatures).setReservedCols("click")
  223.             ).add(
  224.                 new LogisticRegression()
  225.                     .setVectorCol("vec")
  226.                     .setLabelCol("click")
  227.                     .setPredictionCol("pred")
  228.                     .setModelStreamFilePath("/tmp/ftrl_models")
  229.                     .setPredictionDetailCol("detail")
  230.                     .setMaxIter(10))
  231.             .fit(trainBatch);
  233.         pipelineModel.save().link(new AkSinkBatchOp().setOverwriteSink(true).setFilePath("/tmp/lr_pipeline.ak"));
  234.         BatchOperator.execute();
  235.     }
  237.     @Test
  238.     public void localPredictor() throws Exception {
  239.         LocalPredictor predictor = new LocalPredictor("/tmp/lr_pipeline.ak",
  240.             TableUtil.schema2SchemaStr(getSmallBatchSet().getSchema()));
  241.         System.out.println(TableUtil.schema2SchemaStr(getSmallBatchSet().getSchema()));
  242.         for (int i = 0; i < Integer.MAX_VALUE; ++i) {
  243.             System.out.println(predictor.map(
  244.                 Row.of("220869541682524752", "0", 14, 10, 21, 2, "1005", 0, "1fbe01fe", "f3845767",
  245.                     "28905ebd", "ecad2386", "7801e8d9", "07d7df22", "a99f214a", "af1c0727", "a0f5f879", "1", "0",
  246.                     15703, 320, 50, 1722, 0, 35, -1, 79)));
  248.             Thread.sleep(5000);
  249.         }
  250.     }
  252.     public static class SplitDataTime extends TableFunction <Row> {
  254.         private Integer parseInt(String s) {
  255.             if ('0' == s.charAt(0)) {
  256.                 return Integer.parseInt(s.substring(1));
  257.             } else {
  258.                 return Integer.parseInt(s);
  259.             }
  260.         }
  262.         public void eval(String str) {
  263.             collect(Row.of(
  264.                 parseInt(str.substring(0, 2)),
  265.                 parseInt(str.substring(2, 4)),
  266.                 parseInt(str.substring(4, 6)),
  267.                 parseInt(str.substring(6, 8))
  268.             ));
  269.         }
  271.         @Override
  272.         public TypeInformation <Row> getResultType() {
  273.             return new RowTypeInfo(AlinkTypes.INT, AlinkTypes.INT, AlinkTypes.INT, AlinkTypes.INT);
  274.         }
  276.     }
  278.     private static Tuple2 <StreamOperator, StreamOperator> getStreamTrainTestData() {
  279.         StringBuilder sbd = new StringBuilder();
  280.         for (int i = 0; i < ORIGIN_COL_NAMES.length; i++) {
  281.             if (i > 0) {
  282.                 sbd.append(",");
  283.             }
  284.             sbd.append(ORIGIN_COL_NAMES[i]).append(" ").append(ORIGIN_COL_TYPES[i]);
  285.         }
  286.         StreamOperator <?> source = new CsvSourceStreamOp()
  287.             .setFilePath(DATA_DIR + FtrlRebaseTest.LARGE_FILE)
  288.             .setSchemaStr(sbd.toString())
  289.             .setIgnoreFirstLine(true)
  290.             .udtf("dt", new String[] {"dt_year", "dt_month", "dt_day", "dt_hour"}, new SplitDataTime())
  291.             .select(COL_NAMES);
  293.         SplitStreamOp splitter = new SplitStreamOp().setFraction(0.5);
  295.         source.link(splitter);
  297.         return new Tuple2 <>(splitter, splitter.getSideOutput(0));
  298.     }
  300.     private static BatchOperator <?> getSmallBatchSet() {
  301.         StringBuilder sbd = new StringBuilder();
  302.         for (int i = 0; i < ORIGIN_COL_NAMES.length; i++) {
  303.             if (i > 0) {
  304.                 sbd.append(",");
  305.             }
  306.             sbd.append(ORIGIN_COL_NAMES[i]).append(" ").append(ORIGIN_COL_TYPES[i]);
  307.         }
  309.         return new CsvSourceBatchOp()
  310.             .setFilePath(DATA_DIR + FtrlRebaseTest.SMALL_FILE)
  311.             .setSchemaStr(sbd.toString())
  312.             .setIgnoreFirstLine(true)
  313.             .udtf("dt", new String[] {"dt_year", "dt_month", "dt_day", "dt_hour"}, new SplitDataTime())
  314.             .select(COL_NAMES);
  315.     }
  316. }

Demo 功能介绍

该demo使用Ftrl 算法对Avazu 数据（https://www.kaggle.com/c/avazu-ctr-prediction/data）进行实时训练并生成模型流，并将模型流实时加载到推理服务中。另外我们还增加了模型rebase 的示例代码，能够很容易的完成用一个批模型定时重新拉回模型，防止模型跑偏。最后还提供了一个模型训练+预测+评估的示例代码。

函数说明

函数 任务类型 说明
trainFeatureModel() 批任务 训练特征工程模型，这个模型将对训练、预测、推理数据进行特征编码
�savePipelineModel() 批任务 训练PipelineModel，该模型是部署到线上服务的模型
�onlineTrainAndSave() 流任务 使用Ftrl实时训练模型，并定时将模型写出到指定目录
�BatchTrainAndSaveRebaseModel() 批任务 训练用来重新初始化的模型，用来拉回模型，防止模型跑偏
� localPredict() 本地推理 本地搭建一个服务，对同一条样本预测，用来验证模型更新
�onlineTrainAndEval() 流任务 Ftrl 训练模型，并对模型进行预测评估

执行步骤

  • 首先我们需要执行 trainFeatureModel() 函数，生成特征工程模型，并存储到目录“/tmp/feature_model.csv�”，后面的函数都需要该模型
  • 第二步，执行savePipelineModel() 函数，生成部署到线上服务的模型，该模型会通过 setModelStreamFilePath(“/tmp/ftrl_models”) 设置模型流的目录，设置完再部署这个模型时，在推理的同时会实时监控这个目录，当有新模型产生时会自动加载新模型，用最新模型进行推理。
  • 第三步，执行localPredict() 函数，将第二步产生的模型部署成本地服务。
  • 第四步，执行�onlineTrainAndSave()，用Ftrl算法实时训练在线模型，并以固定频率输出到目录”/tmp/ftrl_models”，这个目录与第二步的目录是同一个目录。另外这一步还要设置一个目录setModelStreamFilePath(“/tmp/rebase_ftrl_models”)，这个目录是用来做模型rebase的，在线学习过程中会实时监控这个目录，当有新的模型出现在这个目录中，会重新加载这个模型作为base模型继续进行训练。
  • 第五步，隔段时间（1小时 or 1天）执行函数BatchTrainAndSaveRebaseModel()，这个函数将训练一个新的批模型，并将其写入到rebase目录”/tmp/rebase_ftrl_models”，这样，第四步中的函数就会监测到这个模型，并进行模型 rebase。

  • onlineTrainAndEval() 函数是一个独立的函数，用来评估在线学习算法生成的模型怎么样，并打印评估结果。

    备注

  • 特征工程这里我们对数据做标准化和FeatureHash，其实这里可以使用任何其他Alink 的特征工程算法，类似OneHot编码，GBDT编码，多热编码，归一化、分桶算法等

  • 实时加载在线训练的模型是通过一个给定的文件系统的目录（”/tmp/ftrl_models”）完成的，这个目录可以是本地目录，也可以是网络文件系统的目录，类似OSS 和HDSF等。
  • 模型重新初始化和在线模型的加载是类似的，也是通过一个给定的文件系统的目录（”/tmp/rebase_ftrl_models”）完成的，同样，这个目录可以是本地目录，也可以是网络文件系统的目录，类似OSS 和HDSF等。