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(); } AlsTopKPredictBatchOp();  AlsTopKPredictBatchOp(Params params); @Override AlsTopKPredictBatchOp linkFrom(BatchOperator<?>... inputs); Table recommendForUsers(BatchOperator model, BatchOperator data);  }
@Test public void testRecommendForUsers() throws Exception { BatchOperator data = BatchOperator.fromTable( MLEnvironmentFactory.getDefault().createBatchTable(rows1, new String[]{"user", "item", "rating"})); BatchOperator predData = BatchOperator.fromTable( MLEnvironmentFactory.getDefault().createBatchTable(rows2, new String[]{"user"})); ALS als = new ALS() .setUserCol("user") .setItemCol("item") .setRateCol("rating") .setLambda(0.01) .setRank(10) .setNumIter(10) .setPredictionCol("recommendations"); ALSModel model = als.fit(data); BatchOperator modelData = BatchOperator.fromTable(model.getModelData()); BatchOperator pred = new AlsTopKPredictBatchOp() .setUserCol("user").setTopK(2).setPredictionCol("topk").linkFrom(modelData, predData); pred.print(); Assert.assertEquals(pred.count(), 4); }