多热编码预测 (MultiHotPredictStreamOp)

举例

输入列为col_0
1. 如果没有填写discreteThresholds, 假设模型中a,b,c,d,e对应的token index为0,1,2,3,4

1.1 handleInvalid为keep

| col_0 | Encode为VECTOR的输出 |
| — | ——————- |
| “a b”| $6$0:1.0 1:1.0|
| “b c”| $6$1:1.0 2:1.0 |
| “c d” | $6$3:1.0 3:1.0|
| “a d” | $6$0:1.0 3:1.0 |
| “d e” | $6$0:3.0 4:1.0 |
| NULL | NULL |

1.2 handleInvalid为skip

1.3 handleInvalid为error: 直接报错

参数说明

代码示例

Python 代码

from pyalink.alink import *

import pandas as pd

useLocalEnv(1)

df = pd.DataFrame([
    ["a b", 1],
    ["b c", 1],
    ["c d", 1],
    ["a d", 2],
    ["d e", 2],
    [None, 1]
])

# load data

inOp = BatchOperator.fromDataframe(df, schemaStr='query string, weight long')
streamOp = StreamOperator.fromDataframe(df, schemaStr='query string, weight long')

# multi hot train
multi_hot = MultiHotTrainBatchOp().setSelectedCols(["query"])
model = inOp.link(multi_hot)
model.print()

# batch predict
predictor = MultiHotPredictStreamOp(model).setSelectedCols(["query"]).setOutputCols(["output"])
predictor.linkFrom(streamOp).print()
StreamOperator.execute()

Java 代码

import org.apache.flink.types.Row;

import com.alibaba.alink.operator.batch.BatchOperator;
import com.alibaba.alink.operator.batch.feature.MultiHotTrainBatchOp;
import com.alibaba.alink.operator.batch.source.MemSourceBatchOp;
import com.alibaba.alink.operator.stream.StreamOperator;
import com.alibaba.alink.operator.stream.feature.MultiHotPredictStreamOp;
import com.alibaba.alink.operator.stream.source.MemSourceStreamOp;
import org.junit.Test;

import java.util.Arrays;
import java.util.List;

public class MultiHotPredictStreamOpTest {
	@Test
	public void testMultiHotPredictStreamOp() throws Exception {
		List <Row> df = Arrays.asList(
			Row.of("a b", 1),
			Row.of("b c", 1),
			Row.of("c d", 1),
			Row.of("a d", 2),
			Row.of("d e", 2),
			Row.of(null, 1)
		);
		BatchOperator <?> bOp = new MemSourceBatchOp(df, "query string, weight int");
		StreamOperator <?> sOp = new MemSourceStreamOp(df, "query string, weight int");
		BatchOperator <?> multi_hot = new MultiHotTrainBatchOp().setSelectedCols("query");
		BatchOperator <?> model = bOp.link(multi_hot);
		StreamOperator <?> predictor = new MultiHotPredictStreamOp(model).setSelectedCols("query").setOutputCols(
			"output");
		predictor.linkFrom(sOp).print();
		StreamOperator.execute();
	}
}

运行结果