class AlsTopKPredictBatchOp extends BatchOperator<AlsTopKPredictBatchOp> implements AlsTopKPredictParams<AlsTopKPredictBatchOp> {

    public Table recommendForUsers(BatchOperator model, BatchOperator data) {
        String userColName = getUserCol();
        String predResultColName = getPredictionCol();
        int topk = getTopK();

        data = data.select("`" + userColName + "`");
        DataSet<Tuple1<Long>> users = data.getDataSet()
            .map(new MapFunction<Row, Tuple1<Long>>() {
                @Override
                public Tuple1<Long> map(Row value) throws Exception {
                    return Tuple1.of(((Number) value.getField(0)).longValue());
                }
            });

        DataSet<Tuple2<Long, float[]>> userFactors = getFactors(model, 0);
        DataSet<Tuple2<Long, float[]>> itemFactors = getFactors(model, 1);

        DataSet<Tuple2<Long, String>> recommend = AlsPredict.recommendForUsers(userFactors, itemFactors, users, topk);

        DataSet<Row> output = recommend
            .map(new MapFunction<Tuple2<Long, String>, Row>() {
                @Override
                public Row map(Tuple2<Long, String> value) throws Exception {
                    return Row.of(value.f0, value.f1);
                }
            });
        setOutput(output, new String[]{userColName, predResultColName},
            new TypeInformation[]{Types.LONG, Types.STRING});

        return this.getOutputTable();
    }

    public  AlsTopKPredictBatchOp();
    public  AlsTopKPredictBatchOp(Params params);

}