ALINK(十八)：数据处理（四）缺失值处理(二)缺失值填充训练 (ImputerTrainBatchOp)

import org.apache.flink.types.Row;
import com.alibaba.alink.operator.batch.BatchOperator;
import com.alibaba.alink.operator.batch.dataproc.ImputerPredictBatchOp;
import com.alibaba.alink.operator.batch.dataproc.ImputerTrainBatchOp;
import com.alibaba.alink.operator.batch.source.MemSourceBatchOp;
import com.alibaba.alink.operator.stream.StreamOperator;
import com.alibaba.alink.operator.stream.dataproc.ImputerPredictStreamOp;
import com.alibaba.alink.operator.stream.source.MemSourceStreamOp;
import org.junit.Test;
import java.util.Arrays;
import java.util.List;
public class ImputerTrainBatchOpTest {
    @Test
    public void testImputerTrainBatchOp() throws Exception {
        List <Row> df_data = Arrays.asList(
            Row.of("a", 10.0, 100),
            Row.of("b", -2.5, 9),
            Row.of("c", 100.2, 1),
            Row.of("d", -99.9, 100),
            Row.of("a", 1.4, 1),
            Row.of("b", -2.2, 9),
            Row.of("c", 100.9, 1),
            Row.of(null, null, null)
        );
        String[] selectedColNames = new String[] {"col2", "col3"};
        BatchOperator <?> inOp = new MemSourceBatchOp(df_data, "col1 string, col2 double, col3 int");
        BatchOperator <?> trainOp = new ImputerTrainBatchOp()
            .setSelectedCols(selectedColNames);
        BatchOperator model = trainOp.linkFrom(inOp);
        BatchOperator <?> predictOp = new ImputerPredictBatchOp();
        predictOp.linkFrom(model, inOp).print();
        StreamOperator <?> sinOp = new MemSourceStreamOp(df_data, "col1 string, col2 double, col3 int");
        StreamOperator <?> predictStreamOp = new ImputerPredictStreamOp(model);
        predictStreamOp.linkFrom(sinOp).print();
        StreamOperator.execute();
    }
}