IterativeComQueue extends BaseComQueue <IterativeComQueue> { @Override public IterativeComQueue setMaxIter(int maxIter) { return super.setMaxIter(maxIter); } IterativeComQueue(); @Override IterativeComQueue setCompareCriterionOfNode0(CompareCriterionFunction compareCriterion); @Override IterativeComQueue setMaxIter(int maxIter);  }
@Test public void testPI() throws Exception { DataSet<Row> result = new IterativeComQueue() .add(new ComputeFunction() { @Override public void calc(ComContext context) { if (1 == context.getStepNo()) { context.putObj("cnt", new int[]{0}); } int[] cnt = context.getObj("cnt"); double x = Math.random(); double y = Math.random(); cnt[0] += ((x * x + y * y < 1) ? 1 : 0); } }) .closeWith(new CompleteResultFunction() { @Override public List<Row> calc(ComContext context) { int[] cnt = context.getObj("cnt"); return Collections.singletonList(Row.of(4.0 * cnt[0] / 1000)); } }) .setMaxIter(1000) .exec(); Assert.assertEquals(3.0, (double) result.collect().get(0).getField(0), 0.5); }
@Test public void testICQLinearRegression() throws Exception { final int m = 10000; final int n = 3; 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); 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; DataSet<Row> model = new IterativeComQueue() .setMaxIter(100) .initWithPartitionedData(TRAIN_DATA, trainData) .initWithBroadcastData(COEFS, initialCoefs) .initWithBroadcastData(SAMPLE_COUNT, sampleCount) .add(new ComputeFunction() { @Override public void calc(ComContext context) { List<Tuple2<DenseVector, Double>> trainData = context.getObj(TRAIN_DATA); List<DenseVector> coefs = context.getObj(COEFS); double[] grads = context.getObj("grads"); if (grads == null) { grads = new double[coefs.get(0).size()]; context.putObj("grads", grads); } Arrays.fill(grads, 0.0); DenseVector gradsWrapper = new DenseVector(grads); for (Tuple2<DenseVector, Double> sample : trainData) { gradsWrapper.plusScaleEqual(sample.f0, sample.f1 - sample.f0.dot(coefs.get(0))); } } }) .add(new AllReduce("grads")) .add(new ComputeFunction() { @Override public void calc(ComContext context) { List<DenseVector> coefs = context.getObj(COEFS); double[] grads = context.getObj("grads"); List<Double> sampleCount = context.getObj(SAMPLE_COUNT); coefs.get(0).plusScaleEqual(new DenseVector(grads), learningRate / sampleCount.get(0)); } }) .closeWith(new CompleteResultFunction() { @Override public List<Row> calc(ComContext context) { if (context.getTaskId() == 0) { List<DenseVector> coefs = context.getObj(COEFS); return Collections.singletonList(Row.of(coefs.get(0))); } else { return null; } } }) .exec(); List<Row> modelL = model.collect(); Assert.assertEquals(data.get(0).f1, data.get(0).f0.dot((Vector) modelL.get(0).getField(0)), 2.0 ); }
@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 ); }
@Test public void testICQ() { try { DataSet<Row> ret = new IterativeComQueue() .setMaxIter(1) .add(new ComputeFunction() { @Override public void calc(ComContext context) { context.putObj("allReduce", new double[]{1.0, 1.0, 1.0}); } }) .add(new AllReduce("allReduce")) .add(new ComputeFunction() { @Override public void calc(ComContext context) { double[] result = context.getObj("allReduce"); System.out.println(JsonConverter.toJson(result)); } }) .closeWith(new CompleteResultFunction() { @Override public List<Row> calc(ComContext context) { return null; } }) .exec(); ret.collect(); } catch (Exception e) { Assert.fail("should not throw exception"); } }