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

Java 代码

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