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 代码
from pyalink.alink import *
import pandas as pd
useLocalEnv(1)
df = pd.DataFrame([
[1.0, "A", 0, 0, 0, 1.0],
[2.0, "B", 1, 1, 0, 2.0],
[3.0, "C", 2, 2, 1, 3.0],
[4.0, "D", 3, 3, 1, 4.0]
])
input = BatchOperator.fromDataframe(df, schemaStr='f0 double, f1 string, f2 int, f3 int, label int, reg_label double')
rfOp = RandomForestTrainBatchOp()\
.setLabelCol("reg_label")\
.setFeatureCols(["f0", "f1", "f2", "f3"])\
.setFeatureSubsamplingRatio(0.5)\
.setSubsamplingRatio(1.0)\
.setNumTreesOfInfoGain(1)\
.setNumTreesOfInfoGain(1)\
.setNumTreesOfInfoGainRatio(1)\
.setCategoricalCols(["f1"])
modelStream = AppendModelStreamFileSinkBatchOp()\
.setFilePath("/tmp/random_forest_model_stream")\
.setNumKeepModel(10)
rfOp.linkFrom(input).link(modelStream)
BatchOperator.execute()
Java 代码
import org.apache.flink.types.Row;
import com.alibaba.alink.operator.batch.BatchOperator;
import com.alibaba.alink.operator.batch.classification.RandomForestTrainBatchOp;
import com.alibaba.alink.operator.batch.sink.AppendModelStreamFileSinkBatchOp;
import com.alibaba.alink.operator.batch.source.MemSourceBatchOp;
import org.junit.Test;
import java.util.Arrays;
public class AppendModelStreamFileSinkBatchOpTest {
@Test
public void testAppendModelStreamFileSinkBatchOp() throws Exception {
Row[] rows = new Row[] {
Row.of(1.0, "A", 0L, 0, 0, 1.0),
Row.of(2.0, "B", 1L, 1, 0, 2.0),
Row.of(3.0, "C", 2L, 2, 1, 3.0),
Row.of(4.0, "D", 3L, 3, 1, 4.0)
};
String[] colNames = new String[] {"f0", "f1", "f2", "f3", "label", "reg_label"};
String labelColName = colNames[4];
MemSourceBatchOp input = new MemSourceBatchOp(
Arrays.asList(rows), new String[] {"f0", "f1", "f2", "f3", "label", "reg_label"}
);
RandomForestTrainBatchOp rfOp = new RandomForestTrainBatchOp()
.setLabelCol(labelColName)
.setFeatureCols(colNames[0], colNames[1], colNames[2], colNames[3])
.setFeatureSubsamplingRatio(0.5)
.setSubsamplingRatio(1.0)
.setNumTreesOfInfoGain(1)
.setNumTreesOfInfoGain(1)
.setNumTreesOfInfoGainRatio(1)
.setCategoricalCols(colNames[1]);
rfOp.linkFrom(input).link(
new AppendModelStreamFileSinkBatchOp()
.setFilePath("/tmp/random_forest_model_stream")
.setNumKeepModel(10)
);
BatchOperator.execute();
}
}