@Test
public void testBucketizer() throws Exception{

    Bucketizer op = new Bucketizer()
        .setSelectedCols(new String[]{"features1", "features2"})
        .setOutputCols(new String[]{"bucket1", "bucket2"})
        .setCutsArray(cutsArray);
    Table res = op.transform(data);
    List<Long> list = MLEnvironmentFactory.getDefault().getBatchTableEnvironment().toDataSet(res.select("bucket1"), Long.class).collect();
    Assert.assertArrayEquals(list.toArray(new Long[0]), new Long[]{0L, 0L, 1L, 1L, 2L, 3L});
    res = op.transform(dataStream);
    DataStreamConversionUtil.fromTable(MLEnvironmentFactory.DEFAULT_ML_ENVIRONMENT_ID,res).print();
    MLEnvironmentFactory.getDefault().getStreamExecutionEnvironment().execute();
}