@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);
}