@Test
public void testICQLinearRegression1() throws Exception {

	final long start = System.currentTimeMillis();
	final int m = 10000;
	final int n = 20;
	List<Tuple2<DenseVector, Double>> data = new ArrayList<>();
	for (int i = 0; i < m; ++i) {
		DenseVector feature = DenseVector.rand(n);
		data.add(Tuple2.of(feature.append(1.0), feature.dot(DenseVector.ones(n))));
	}
	DataSet<Tuple2<DenseVector, Double>> trainData =
		MLEnvironmentFactory
			.getDefault()
			.getExecutionEnvironment()
			.fromCollection(data)
			.rebalance();
	DataSet<DenseVector> initialCoefs =
		MLEnvironmentFactory
			.getDefault()
			.getExecutionEnvironment()
			.fromCollection(Collections.singletonList(DenseVector.zeros(n + 1)));
	DataSet<Double> sampleCount = DataSetUtils
		.countElementsPerPartition(trainData)
		.sum(1)
		.map(new MapFunction<Tuple2<Integer, Long>, Double>() {
			@Override
			public Double map(Tuple2<Integer, Long> value) throws Exception {
				return value.f1.doubleValue();
			}
		});
	final double learningRate = 1.0;
	final double decayLrRate = 0.8;
	final double decayStep = 5;
	DataSet<Row> model = new IterativeComQueue()
		.setMaxIter(1000)
		.initWithPartitionedData(TRAIN_DATA, trainData)
		.initWithBroadcastData(COEFS, initialCoefs)
		.initWithBroadcastData(SAMPLE_COUNT, sampleCount)
		.add(new UpdateCoefs(learningRate, decayStep, decayLrRate))
		.add(new AllReduce(COEFS_ARRAY))
		.closeWith(new SerializeModel())
		.exec();
	List<Row> modelL = model.collect();
	System.out.println(JsonConverter.toJson(modelL));
	for (int i = 0; i < 10; ++i) {
		System.out.println(
			JsonConverter.toJson(
				Tuple2.of(
					data.get(i).f1,
					data.get(i).f0.dot((Vector) modelL.get(0).getField(0))
				)
			)
		);
	}
	final long end = System.currentTimeMillis();
	System.out.println(String.format("ICQ time: %d", end - start));
	Assert.assertEquals(data.get(0).f1,
		data.get(0).f0.dot((Vector) modelL.get(0).getField(0)),
		2.0
	);
}