@Test public void testAggregate1() throws Exception { INDArray updates1 = Nd4j.create(1000).assign(1.0); INDArray updates2 = Nd4j.create(1000).assign(2.0); INDArray expUpdates = Nd4j.create(1000).assign(3.0); SharedTrainingResult result1 = SharedTrainingResult.builder().updaterStateArray(updates1).aggregationsCount(1) .scoreSum(1.0).build(); SharedTrainingResult result2 = SharedTrainingResult.builder().updaterStateArray(updates2).aggregationsCount(1) .scoreSum(2.0).build(); SharedTrainingAggregateFunction aggregateFunction = new SharedTrainingAggregateFunction(); SharedTrainingAccumulationTuple tuple1 = aggregateFunction.call(null, result1); SharedTrainingAccumulationTuple tuple2 = aggregateFunction.call(tuple1, result2); assertEquals(2, tuple2.getAggregationsCount()); assertEquals(3.0, tuple2.getScoreSum(), 0.001); assertEquals(expUpdates, tuple2.getUpdaterStateArray()); }
@Override public SharedTrainingAccumulationTuple call(SharedTrainingAccumulationTuple tuple, SharedTrainingResult result) throws Exception { if (tuple == null) { ThresholdAlgorithmReducer tar = null; if(result.getThresholdAlgorithm() != null){ tar = result.getThresholdAlgorithm().newReducer(); tar.add(result.getThresholdAlgorithm()); } return SharedTrainingAccumulationTuple.builder().updaterStateArray(result.getUpdaterStateArray()) .scoreSum(result.getScoreSum()).listenerStaticInfo(result.getListenerStaticInfo()) .listenerUpdates(result.getListenerUpdates()).listenerMetaData(result.getListenerMetaData()) .sparkTrainingStats(result.getSparkTrainingStats()) .aggregationsCount(result.getAggregationsCount()) .minibatchesPerExecutor(result.getMinibatchesPerExecutor()) .thresholdAlgorithmReducer(tar) .build(); } INDArray updaterStateSum = null; int aggregationsCount = 0; double score = 0.0; if (tuple.getUpdaterStateArray() != null) { if (result.getUpdaterStateArray() != null) { updaterStateSum = tuple.getUpdaterStateArray().addi(result.getUpdaterStateArray()); aggregationsCount = tuple.getAggregationsCount() + 1; score = tuple.getScoreSum() + result.getScoreSum(); } } else { if (result.getUpdaterStateArray() != null) { updaterStateSum = result.getUpdaterStateArray(); aggregationsCount = 1; score = result.getScoreSum(); } } SparkTrainingStats stats = tuple.getSparkTrainingStats(); if (result.getSparkTrainingStats() != null) { if (stats == null) stats = result.getSparkTrainingStats(); else stats.addOtherTrainingStats(result.getSparkTrainingStats()); } Nd4j.getExecutioner().commit(); Collection<StorageMetaData> listenerMetaData = tuple.getListenerMetaData(); if (listenerMetaData == null) listenerMetaData = result.getListenerMetaData(); else { Collection<StorageMetaData> newMeta = result.getListenerMetaData(); if (newMeta != null) listenerMetaData.addAll(newMeta); } Collection<Persistable> listenerStaticInfo = tuple.getListenerStaticInfo(); if (listenerStaticInfo == null) listenerStaticInfo = result.getListenerStaticInfo(); else { Collection<Persistable> newStatic = result.getListenerStaticInfo(); if (newStatic != null) listenerStaticInfo.addAll(newStatic); } Collection<Persistable> listenerUpdates = tuple.getListenerUpdates(); if (listenerUpdates == null) listenerUpdates = result.getListenerUpdates(); else { Collection<Persistable> listenerUpdates2 = result.getListenerUpdates(); if (listenerUpdates2 != null) listenerUpdates.addAll(listenerUpdates2); } Map<String,Integer> minibatchesPerExecutor = new HashMap<>(); if(tuple.getMinibatchesPerExecutor() != null) { for (Map.Entry<String, Integer> e : tuple.getMinibatchesPerExecutor().entrySet()){ minibatchesPerExecutor.put(e.getKey(), e.getValue()); } } if(result.getMinibatchesPerExecutor() != null){ for (Map.Entry<String, Integer> e : result.getMinibatchesPerExecutor().entrySet()){ if(minibatchesPerExecutor.containsKey(e.getKey())){ minibatchesPerExecutor.put(e.getKey(), minibatchesPerExecutor.get(e.getKey()) + e.getValue()); } else { minibatchesPerExecutor.put(e.getKey(), e.getValue()); } } } ThresholdAlgorithmReducer thresholdAlgorithmReducer = tuple.getThresholdAlgorithmReducer(); if(thresholdAlgorithmReducer == null && result.getThresholdAlgorithm() != null){ thresholdAlgorithmReducer = result.getThresholdAlgorithm().newReducer(); } if(thresholdAlgorithmReducer != null){ thresholdAlgorithmReducer.add(result.getThresholdAlgorithm()); } return SharedTrainingAccumulationTuple.builder().scoreSum(score).updaterStateArray(updaterStateSum) .aggregationsCount(aggregationsCount).sparkTrainingStats(stats) .listenerMetaData(listenerMetaData).listenerUpdates(listenerUpdates) .listenerStaticInfo(listenerStaticInfo) .minibatchesPerExecutor(minibatchesPerExecutor) .thresholdAlgorithmReducer(thresholdAlgorithmReducer) .build(); }
SharedTrainingAggregateFunction implements
                Function2<SharedTrainingAccumulationTuple, SharedTrainingResult, SharedTrainingAccumulationTuple> { @Override public SharedTrainingAccumulationTuple call(SharedTrainingAccumulationTuple tuple, SharedTrainingResult result) throws Exception { if (tuple == null) { ThresholdAlgorithmReducer tar = null; if(result.getThresholdAlgorithm() != null){ tar = result.getThresholdAlgorithm().newReducer(); tar.add(result.getThresholdAlgorithm()); } return SharedTrainingAccumulationTuple.builder().updaterStateArray(result.getUpdaterStateArray()) .scoreSum(result.getScoreSum()).listenerStaticInfo(result.getListenerStaticInfo()) .listenerUpdates(result.getListenerUpdates()).listenerMetaData(result.getListenerMetaData()) .sparkTrainingStats(result.getSparkTrainingStats()) .aggregationsCount(result.getAggregationsCount()) .minibatchesPerExecutor(result.getMinibatchesPerExecutor()) .thresholdAlgorithmReducer(tar) .build(); } INDArray updaterStateSum = null; int aggregationsCount = 0; double score = 0.0; if (tuple.getUpdaterStateArray() != null) { if (result.getUpdaterStateArray() != null) { updaterStateSum = tuple.getUpdaterStateArray().addi(result.getUpdaterStateArray()); aggregationsCount = tuple.getAggregationsCount() + 1; score = tuple.getScoreSum() + result.getScoreSum(); } } else { if (result.getUpdaterStateArray() != null) { updaterStateSum = result.getUpdaterStateArray(); aggregationsCount = 1; score = result.getScoreSum(); } } SparkTrainingStats stats = tuple.getSparkTrainingStats(); if (result.getSparkTrainingStats() != null) { if (stats == null) stats = result.getSparkTrainingStats(); else stats.addOtherTrainingStats(result.getSparkTrainingStats()); } Nd4j.getExecutioner().commit(); Collection<StorageMetaData> listenerMetaData = tuple.getListenerMetaData(); if (listenerMetaData == null) listenerMetaData = result.getListenerMetaData(); else { Collection<StorageMetaData> newMeta = result.getListenerMetaData(); if (newMeta != null) listenerMetaData.addAll(newMeta); } Collection<Persistable> listenerStaticInfo = tuple.getListenerStaticInfo(); if (listenerStaticInfo == null) listenerStaticInfo = result.getListenerStaticInfo(); else { Collection<Persistable> newStatic = result.getListenerStaticInfo(); if (newStatic != null) listenerStaticInfo.addAll(newStatic); } Collection<Persistable> listenerUpdates = tuple.getListenerUpdates(); if (listenerUpdates == null) listenerUpdates = result.getListenerUpdates(); else { Collection<Persistable> listenerUpdates2 = result.getListenerUpdates(); if (listenerUpdates2 != null) listenerUpdates.addAll(listenerUpdates2); } Map<String,Integer> minibatchesPerExecutor = new HashMap<>(); if(tuple.getMinibatchesPerExecutor() != null) { for (Map.Entry<String, Integer> e : tuple.getMinibatchesPerExecutor().entrySet()){ minibatchesPerExecutor.put(e.getKey(), e.getValue()); } } if(result.getMinibatchesPerExecutor() != null){ for (Map.Entry<String, Integer> e : result.getMinibatchesPerExecutor().entrySet()){ if(minibatchesPerExecutor.containsKey(e.getKey())){ minibatchesPerExecutor.put(e.getKey(), minibatchesPerExecutor.get(e.getKey()) + e.getValue()); } else { minibatchesPerExecutor.put(e.getKey(), e.getValue()); } } } ThresholdAlgorithmReducer thresholdAlgorithmReducer = tuple.getThresholdAlgorithmReducer(); if(thresholdAlgorithmReducer == null && result.getThresholdAlgorithm() != null){ thresholdAlgorithmReducer = result.getThresholdAlgorithm().newReducer(); } if(thresholdAlgorithmReducer != null){ thresholdAlgorithmReducer.add(result.getThresholdAlgorithm()); } return SharedTrainingAccumulationTuple.builder().scoreSum(score).updaterStateArray(updaterStateSum) .aggregationsCount(aggregationsCount).sparkTrainingStats(stats) .listenerMetaData(listenerMetaData).listenerUpdates(listenerUpdates) .listenerStaticInfo(listenerStaticInfo) .minibatchesPerExecutor(minibatchesPerExecutor) .thresholdAlgorithmReducer(thresholdAlgorithmReducer) .build(); } }
SharedTrainingAggregateFunction implements
                Function2<SharedTrainingAccumulationTuple, SharedTrainingResult, SharedTrainingAccumulationTuple> { @Override public SharedTrainingAccumulationTuple call(SharedTrainingAccumulationTuple tuple, SharedTrainingResult result) throws Exception { if (tuple == null) { ThresholdAlgorithmReducer tar = null; if(result.getThresholdAlgorithm() != null){ tar = result.getThresholdAlgorithm().newReducer(); tar.add(result.getThresholdAlgorithm()); } return SharedTrainingAccumulationTuple.builder().updaterStateArray(result.getUpdaterStateArray()) .scoreSum(result.getScoreSum()).listenerStaticInfo(result.getListenerStaticInfo()) .listenerUpdates(result.getListenerUpdates()).listenerMetaData(result.getListenerMetaData()) .sparkTrainingStats(result.getSparkTrainingStats()) .aggregationsCount(result.getAggregationsCount()) .minibatchesPerExecutor(result.getMinibatchesPerExecutor()) .thresholdAlgorithmReducer(tar) .build(); } INDArray updaterStateSum = null; int aggregationsCount = 0; double score = 0.0; if (tuple.getUpdaterStateArray() != null) { if (result.getUpdaterStateArray() != null) { updaterStateSum = tuple.getUpdaterStateArray().addi(result.getUpdaterStateArray()); aggregationsCount = tuple.getAggregationsCount() + 1; score = tuple.getScoreSum() + result.getScoreSum(); } } else { if (result.getUpdaterStateArray() != null) { updaterStateSum = result.getUpdaterStateArray(); aggregationsCount = 1; score = result.getScoreSum(); } } SparkTrainingStats stats = tuple.getSparkTrainingStats(); if (result.getSparkTrainingStats() != null) { if (stats == null) stats = result.getSparkTrainingStats(); else stats.addOtherTrainingStats(result.getSparkTrainingStats()); } Nd4j.getExecutioner().commit(); Collection<StorageMetaData> listenerMetaData = tuple.getListenerMetaData(); if (listenerMetaData == null) listenerMetaData = result.getListenerMetaData(); else { Collection<StorageMetaData> newMeta = result.getListenerMetaData(); if (newMeta != null) listenerMetaData.addAll(newMeta); } Collection<Persistable> listenerStaticInfo = tuple.getListenerStaticInfo(); if (listenerStaticInfo == null) listenerStaticInfo = result.getListenerStaticInfo(); else { Collection<Persistable> newStatic = result.getListenerStaticInfo(); if (newStatic != null) listenerStaticInfo.addAll(newStatic); } Collection<Persistable> listenerUpdates = tuple.getListenerUpdates(); if (listenerUpdates == null) listenerUpdates = result.getListenerUpdates(); else { Collection<Persistable> listenerUpdates2 = result.getListenerUpdates(); if (listenerUpdates2 != null) listenerUpdates.addAll(listenerUpdates2); } Map<String,Integer> minibatchesPerExecutor = new HashMap<>(); if(tuple.getMinibatchesPerExecutor() != null) { for (Map.Entry<String, Integer> e : tuple.getMinibatchesPerExecutor().entrySet()){ minibatchesPerExecutor.put(e.getKey(), e.getValue()); } } if(result.getMinibatchesPerExecutor() != null){ for (Map.Entry<String, Integer> e : result.getMinibatchesPerExecutor().entrySet()){ if(minibatchesPerExecutor.containsKey(e.getKey())){ minibatchesPerExecutor.put(e.getKey(), minibatchesPerExecutor.get(e.getKey()) + e.getValue()); } else { minibatchesPerExecutor.put(e.getKey(), e.getValue()); } } } ThresholdAlgorithmReducer thresholdAlgorithmReducer = tuple.getThresholdAlgorithmReducer(); if(thresholdAlgorithmReducer == null && result.getThresholdAlgorithm() != null){ thresholdAlgorithmReducer = result.getThresholdAlgorithm().newReducer(); } if(thresholdAlgorithmReducer != null){ thresholdAlgorithmReducer.add(result.getThresholdAlgorithm()); } return SharedTrainingAccumulationTuple.builder().scoreSum(score).updaterStateArray(updaterStateSum) .aggregationsCount(aggregationsCount).sparkTrainingStats(stats) .listenerMetaData(listenerMetaData).listenerUpdates(listenerUpdates) .listenerStaticInfo(listenerStaticInfo) .minibatchesPerExecutor(minibatchesPerExecutor) .thresholdAlgorithmReducer(thresholdAlgorithmReducer) .build(); }  }
SharedTrainingAggregateFunction implements
                Function2<SharedTrainingAccumulationTuple, SharedTrainingResult, SharedTrainingAccumulationTuple> { @Override public SharedTrainingAccumulationTuple call(SharedTrainingAccumulationTuple tuple, SharedTrainingResult result) throws Exception { if (tuple == null) { ThresholdAlgorithmReducer tar = null; if(result.getThresholdAlgorithm() != null){ tar = result.getThresholdAlgorithm().newReducer(); tar.add(result.getThresholdAlgorithm()); } return SharedTrainingAccumulationTuple.builder().updaterStateArray(result.getUpdaterStateArray()) .scoreSum(result.getScoreSum()).listenerStaticInfo(result.getListenerStaticInfo()) .listenerUpdates(result.getListenerUpdates()).listenerMetaData(result.getListenerMetaData()) .sparkTrainingStats(result.getSparkTrainingStats()) .aggregationsCount(result.getAggregationsCount()) .minibatchesPerExecutor(result.getMinibatchesPerExecutor()) .thresholdAlgorithmReducer(tar) .build(); } INDArray updaterStateSum = null; int aggregationsCount = 0; double score = 0.0; if (tuple.getUpdaterStateArray() != null) { if (result.getUpdaterStateArray() != null) { updaterStateSum = tuple.getUpdaterStateArray().addi(result.getUpdaterStateArray()); aggregationsCount = tuple.getAggregationsCount() + 1; score = tuple.getScoreSum() + result.getScoreSum(); } } else { if (result.getUpdaterStateArray() != null) { updaterStateSum = result.getUpdaterStateArray(); aggregationsCount = 1; score = result.getScoreSum(); } } SparkTrainingStats stats = tuple.getSparkTrainingStats(); if (result.getSparkTrainingStats() != null) { if (stats == null) stats = result.getSparkTrainingStats(); else stats.addOtherTrainingStats(result.getSparkTrainingStats()); } Nd4j.getExecutioner().commit(); Collection<StorageMetaData> listenerMetaData = tuple.getListenerMetaData(); if (listenerMetaData == null) listenerMetaData = result.getListenerMetaData(); else { Collection<StorageMetaData> newMeta = result.getListenerMetaData(); if (newMeta != null) listenerMetaData.addAll(newMeta); } Collection<Persistable> listenerStaticInfo = tuple.getListenerStaticInfo(); if (listenerStaticInfo == null) listenerStaticInfo = result.getListenerStaticInfo(); else { Collection<Persistable> newStatic = result.getListenerStaticInfo(); if (newStatic != null) listenerStaticInfo.addAll(newStatic); } Collection<Persistable> listenerUpdates = tuple.getListenerUpdates(); if (listenerUpdates == null) listenerUpdates = result.getListenerUpdates(); else { Collection<Persistable> listenerUpdates2 = result.getListenerUpdates(); if (listenerUpdates2 != null) listenerUpdates.addAll(listenerUpdates2); } Map<String,Integer> minibatchesPerExecutor = new HashMap<>(); if(tuple.getMinibatchesPerExecutor() != null) { for (Map.Entry<String, Integer> e : tuple.getMinibatchesPerExecutor().entrySet()){ minibatchesPerExecutor.put(e.getKey(), e.getValue()); } } if(result.getMinibatchesPerExecutor() != null){ for (Map.Entry<String, Integer> e : result.getMinibatchesPerExecutor().entrySet()){ if(minibatchesPerExecutor.containsKey(e.getKey())){ minibatchesPerExecutor.put(e.getKey(), minibatchesPerExecutor.get(e.getKey()) + e.getValue()); } else { minibatchesPerExecutor.put(e.getKey(), e.getValue()); } } } ThresholdAlgorithmReducer thresholdAlgorithmReducer = tuple.getThresholdAlgorithmReducer(); if(thresholdAlgorithmReducer == null && result.getThresholdAlgorithm() != null){ thresholdAlgorithmReducer = result.getThresholdAlgorithm().newReducer(); } if(thresholdAlgorithmReducer != null){ thresholdAlgorithmReducer.add(result.getThresholdAlgorithm()); } return SharedTrainingAccumulationTuple.builder().scoreSum(score).updaterStateArray(updaterStateSum) .aggregationsCount(aggregationsCount).sparkTrainingStats(stats) .listenerMetaData(listenerMetaData).listenerUpdates(listenerUpdates) .listenerStaticInfo(listenerStaticInfo) .minibatchesPerExecutor(minibatchesPerExecutor) .thresholdAlgorithmReducer(thresholdAlgorithmReducer) .build(); }  @Override SharedTrainingAccumulationTuple call(SharedTrainingAccumulationTuple tuple, SharedTrainingResult result); }
SharedTrainingAggregateFunction implements
                Function2<SharedTrainingAccumulationTuple, SharedTrainingResult, SharedTrainingAccumulationTuple> { @Override public SharedTrainingAccumulationTuple call(SharedTrainingAccumulationTuple tuple, SharedTrainingResult result) throws Exception { if (tuple == null) { ThresholdAlgorithmReducer tar = null; if(result.getThresholdAlgorithm() != null){ tar = result.getThresholdAlgorithm().newReducer(); tar.add(result.getThresholdAlgorithm()); } return SharedTrainingAccumulationTuple.builder().updaterStateArray(result.getUpdaterStateArray()) .scoreSum(result.getScoreSum()).listenerStaticInfo(result.getListenerStaticInfo()) .listenerUpdates(result.getListenerUpdates()).listenerMetaData(result.getListenerMetaData()) .sparkTrainingStats(result.getSparkTrainingStats()) .aggregationsCount(result.getAggregationsCount()) .minibatchesPerExecutor(result.getMinibatchesPerExecutor()) .thresholdAlgorithmReducer(tar) .build(); } INDArray updaterStateSum = null; int aggregationsCount = 0; double score = 0.0; if (tuple.getUpdaterStateArray() != null) { if (result.getUpdaterStateArray() != null) { updaterStateSum = tuple.getUpdaterStateArray().addi(result.getUpdaterStateArray()); aggregationsCount = tuple.getAggregationsCount() + 1; score = tuple.getScoreSum() + result.getScoreSum(); } } else { if (result.getUpdaterStateArray() != null) { updaterStateSum = result.getUpdaterStateArray(); aggregationsCount = 1; score = result.getScoreSum(); } } SparkTrainingStats stats = tuple.getSparkTrainingStats(); if (result.getSparkTrainingStats() != null) { if (stats == null) stats = result.getSparkTrainingStats(); else stats.addOtherTrainingStats(result.getSparkTrainingStats()); } Nd4j.getExecutioner().commit(); Collection<StorageMetaData> listenerMetaData = tuple.getListenerMetaData(); if (listenerMetaData == null) listenerMetaData = result.getListenerMetaData(); else { Collection<StorageMetaData> newMeta = result.getListenerMetaData(); if (newMeta != null) listenerMetaData.addAll(newMeta); } Collection<Persistable> listenerStaticInfo = tuple.getListenerStaticInfo(); if (listenerStaticInfo == null) listenerStaticInfo = result.getListenerStaticInfo(); else { Collection<Persistable> newStatic = result.getListenerStaticInfo(); if (newStatic != null) listenerStaticInfo.addAll(newStatic); } Collection<Persistable> listenerUpdates = tuple.getListenerUpdates(); if (listenerUpdates == null) listenerUpdates = result.getListenerUpdates(); else { Collection<Persistable> listenerUpdates2 = result.getListenerUpdates(); if (listenerUpdates2 != null) listenerUpdates.addAll(listenerUpdates2); } Map<String,Integer> minibatchesPerExecutor = new HashMap<>(); if(tuple.getMinibatchesPerExecutor() != null) { for (Map.Entry<String, Integer> e : tuple.getMinibatchesPerExecutor().entrySet()){ minibatchesPerExecutor.put(e.getKey(), e.getValue()); } } if(result.getMinibatchesPerExecutor() != null){ for (Map.Entry<String, Integer> e : result.getMinibatchesPerExecutor().entrySet()){ if(minibatchesPerExecutor.containsKey(e.getKey())){ minibatchesPerExecutor.put(e.getKey(), minibatchesPerExecutor.get(e.getKey()) + e.getValue()); } else { minibatchesPerExecutor.put(e.getKey(), e.getValue()); } } } ThresholdAlgorithmReducer thresholdAlgorithmReducer = tuple.getThresholdAlgorithmReducer(); if(thresholdAlgorithmReducer == null && result.getThresholdAlgorithm() != null){ thresholdAlgorithmReducer = result.getThresholdAlgorithm().newReducer(); } if(thresholdAlgorithmReducer != null){ thresholdAlgorithmReducer.add(result.getThresholdAlgorithm()); } return SharedTrainingAccumulationTuple.builder().scoreSum(score).updaterStateArray(updaterStateSum) .aggregationsCount(aggregationsCount).sparkTrainingStats(stats) .listenerMetaData(listenerMetaData).listenerUpdates(listenerUpdates) .listenerStaticInfo(listenerStaticInfo) .minibatchesPerExecutor(minibatchesPerExecutor) .thresholdAlgorithmReducer(thresholdAlgorithmReducer) .build(); }  @Override SharedTrainingAccumulationTuple call(SharedTrainingAccumulationTuple tuple, SharedTrainingResult result);  }