Java 类名：com.alibaba.alink.operator.batch.sink.AppendModelStreamFileSinkBatchOp
Python 类名：AppendModelStreamFileSinkBatchOp

功能介绍

将模型按照给定的时间戳，插入模型流。

参数说明

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

| filePath | 文件路径 | 文件路径 | String | ✓ | | |

| modelTime | 批模型时间戳 | 模型时间戳。默认当前时间。 使用yyyy-mm-dd hh:mm:ss.fffffffff格式，详见Timestamp.valueOf(String s) | String | | | null |

| numFiles | 文件数目 | 文件数目 | Integer | | | 1 |

| numKeepModel | 保存模型的数目 | 实时写出模型的数目上限 | Integer | | | 2147483647 |

代码示例

Python 代码

  1. from pyalink.alink import *
  3. import pandas as pd
  5. useLocalEnv(1)
  7. df = pd.DataFrame([
  8.     [1.0, "A", 0, 0, 0, 1.0],
  9.     [2.0, "B", 1, 1, 0, 2.0],
  10.     [3.0, "C", 2, 2, 1, 3.0],
  11.     [4.0, "D", 3, 3, 1, 4.0]
  12. ])
  14. input = BatchOperator.fromDataframe(df, schemaStr='f0 double, f1 string, f2 int, f3 int, label int, reg_label double')
  16. rfOp = RandomForestTrainBatchOp()\
  17.     .setLabelCol("reg_label")\
  18.     .setFeatureCols(["f0", "f1", "f2", "f3"])\
  19.     .setFeatureSubsamplingRatio(0.5)\
  20.     .setSubsamplingRatio(1.0)\
  21.     .setNumTreesOfInfoGain(1)\
  22.     .setNumTreesOfInfoGain(1)\
  23.     .setNumTreesOfInfoGainRatio(1)\
  24.     .setCategoricalCols(["f1"])
  26. modelStream = AppendModelStreamFileSinkBatchOp()\
  27.     .setFilePath("/tmp/random_forest_model_stream")\
  28.     .setNumKeepModel(10)
  30. rfOp.linkFrom(input).link(modelStream)
  32. BatchOperator.execute()

Java 代码

  1. import org.apache.flink.types.Row;
  3. import com.alibaba.alink.operator.batch.BatchOperator;
  4. import com.alibaba.alink.operator.batch.classification.RandomForestTrainBatchOp;
  5. import com.alibaba.alink.operator.batch.sink.AppendModelStreamFileSinkBatchOp;
  6. import com.alibaba.alink.operator.batch.source.MemSourceBatchOp;
  7. import org.junit.Test;
  9. import java.util.Arrays;
  11. public class AppendModelStreamFileSinkBatchOpTest {
  12.     @Test
  13.     public void testAppendModelStreamFileSinkBatchOp() throws Exception {
  14.         Row[] rows = new Row[] {
  15.             Row.of(1.0, "A", 0L, 0, 0, 1.0),
  16.             Row.of(2.0, "B", 1L, 1, 0, 2.0),
  17.             Row.of(3.0, "C", 2L, 2, 1, 3.0),
  18.             Row.of(4.0, "D", 3L, 3, 1, 4.0)
  19.         };
  21.         String[] colNames = new String[] {"f0", "f1", "f2", "f3", "label", "reg_label"};
  23.         String labelColName = colNames[4];
  25.         MemSourceBatchOp input = new MemSourceBatchOp(
  26.             Arrays.asList(rows), new String[] {"f0", "f1", "f2", "f3", "label", "reg_label"}
  27.         );
  29.         RandomForestTrainBatchOp rfOp = new RandomForestTrainBatchOp()
  30.             .setLabelCol(labelColName)
  31.             .setFeatureCols(colNames[0], colNames[1], colNames[2], colNames[3])
  32.             .setFeatureSubsamplingRatio(0.5)
  33.             .setSubsamplingRatio(1.0)
  34.             .setNumTreesOfInfoGain(1)
  35.             .setNumTreesOfInfoGain(1)
  36.             .setNumTreesOfInfoGainRatio(1)
  37.             .setCategoricalCols(colNames[1]);
  39.         rfOp.linkFrom(input).link(
  40.             new AppendModelStreamFileSinkBatchOp()
  41.                 .setFilePath("/tmp/random_forest_model_stream")
  42.                 .setNumKeepModel(10)
  43.         );
  45.         BatchOperator.execute();
  46.     }
  47. }