@Test public void testAccumulation1() 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); SharedTrainingAccumulationTuple tuple1 = SharedTrainingAccumulationTuple.builder().updaterStateArray(updates1) .scoreSum(1.0).aggregationsCount(1).build(); SharedTrainingAccumulationTuple tuple2 = SharedTrainingAccumulationTuple.builder().updaterStateArray(updates2) .scoreSum(2.0).aggregationsCount(1).build(); SharedTrainingAccumulationFunction accumulationFunction = new SharedTrainingAccumulationFunction(); SharedTrainingAccumulationTuple tupleE = accumulationFunction.call(null, tuple1); assertEquals(1, tupleE.getAggregationsCount()); assertEquals(1.0, tupleE.getScoreSum(), 0.01); assertEquals(updates1, tupleE.getUpdaterStateArray()); SharedTrainingAccumulationTuple tupleResult = accumulationFunction.call(tuple1, tuple2); assertEquals(2, tupleResult.getAggregationsCount()); assertEquals(3.0, tupleResult.getScoreSum(), 0.01); assertEquals(expUpdates, tupleResult.getUpdaterStateArray()); }
@Override public SharedTrainingAccumulationTuple call(SharedTrainingAccumulationTuple tuple1, SharedTrainingAccumulationTuple tuple2) throws Exception { if (tuple1 == null) return tuple2; else if (tuple2 == null) return tuple1; double score = 0.0; INDArray stateView = null; int aggregationsCount = 0; if (tuple1.getUpdaterStateArray() != null && tuple2.getUpdaterStateArray() != null) { stateView = tuple1.getUpdaterStateArray().addi(tuple2.getUpdaterStateArray()); } else if (tuple1.getUpdaterStateArray() != null || tuple2.getUpdaterStateArray() != null) { stateView = tuple1.getUpdaterStateArray() != null ? tuple1.getUpdaterStateArray() : tuple2.getUpdaterStateArray(); } aggregationsCount = tuple1.getAggregationsCount() + tuple2.getAggregationsCount(); score = tuple1.getScoreSum() + tuple2.getScoreSum(); SparkTrainingStats stats = tuple1.getSparkTrainingStats(); if (tuple2.getSparkTrainingStats() != null) { if (stats == null) stats = tuple2.getSparkTrainingStats(); else stats.addOtherTrainingStats(tuple2.getSparkTrainingStats()); } Nd4j.getExecutioner().commit(); Collection<StorageMetaData> listenerMetaData = tuple1.getListenerMetaData(); if (listenerMetaData == null) listenerMetaData = tuple2.getListenerMetaData(); else { Collection<StorageMetaData> newMeta = tuple2.getListenerMetaData(); if (newMeta != null) listenerMetaData.addAll(newMeta); } Collection<Persistable> listenerStaticInfo = tuple1.getListenerStaticInfo(); if (listenerStaticInfo == null) listenerStaticInfo = tuple2.getListenerStaticInfo(); else { Collection<Persistable> newStatic = tuple2.getListenerStaticInfo(); if (newStatic != null) listenerStaticInfo.addAll(newStatic); } Collection<Persistable> listenerUpdates = tuple1.getListenerUpdates(); if (listenerUpdates == null) listenerUpdates = tuple2.getListenerUpdates(); else { Collection<Persistable> listenerUpdates2 = tuple2.getListenerUpdates(); if (listenerUpdates2 != null) listenerUpdates.addAll(listenerUpdates2); } Map<String,Integer> minibatchesPerExecutor = new HashMap<>(); if(tuple1.getMinibatchesPerExecutor() != null) { for (Map.Entry<String, Integer> e : tuple1.getMinibatchesPerExecutor().entrySet()){ minibatchesPerExecutor.put(e.getKey(), e.getValue()); } } if(tuple2.getMinibatchesPerExecutor() != null){ for (Map.Entry<String, Integer> e : tuple2.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 = null; if(tuple1.getThresholdAlgorithmReducer() != null){ thresholdAlgorithmReducer = tuple1.getThresholdAlgorithmReducer(); } if(tuple2.getThresholdAlgorithmReducer() != null){ if(thresholdAlgorithmReducer == null){ thresholdAlgorithmReducer = tuple2.getThresholdAlgorithmReducer(); } else { thresholdAlgorithmReducer = thresholdAlgorithmReducer.merge(tuple2.getThresholdAlgorithmReducer()); } } return SharedTrainingAccumulationTuple.builder().scoreSum(score).updaterStateArray(stateView) .aggregationsCount(aggregationsCount).sparkTrainingStats(stats) .listenerMetaData(listenerMetaData).listenerUpdates(listenerUpdates) .listenerStaticInfo(listenerStaticInfo) .minibatchesPerExecutor(minibatchesPerExecutor) .thresholdAlgorithmReducer(thresholdAlgorithmReducer) .build(); }
SharedTrainingAccumulationFunction implements
                Function2<SharedTrainingAccumulationTuple, SharedTrainingAccumulationTuple, SharedTrainingAccumulationTuple> { @Override public SharedTrainingAccumulationTuple call(SharedTrainingAccumulationTuple tuple1, SharedTrainingAccumulationTuple tuple2) throws Exception { if (tuple1 == null) return tuple2; else if (tuple2 == null) return tuple1; double score = 0.0; INDArray stateView = null; int aggregationsCount = 0; if (tuple1.getUpdaterStateArray() != null && tuple2.getUpdaterStateArray() != null) { stateView = tuple1.getUpdaterStateArray().addi(tuple2.getUpdaterStateArray()); } else if (tuple1.getUpdaterStateArray() != null || tuple2.getUpdaterStateArray() != null) { stateView = tuple1.getUpdaterStateArray() != null ? tuple1.getUpdaterStateArray() : tuple2.getUpdaterStateArray(); } aggregationsCount = tuple1.getAggregationsCount() + tuple2.getAggregationsCount(); score = tuple1.getScoreSum() + tuple2.getScoreSum(); SparkTrainingStats stats = tuple1.getSparkTrainingStats(); if (tuple2.getSparkTrainingStats() != null) { if (stats == null) stats = tuple2.getSparkTrainingStats(); else stats.addOtherTrainingStats(tuple2.getSparkTrainingStats()); } Nd4j.getExecutioner().commit(); Collection<StorageMetaData> listenerMetaData = tuple1.getListenerMetaData(); if (listenerMetaData == null) listenerMetaData = tuple2.getListenerMetaData(); else { Collection<StorageMetaData> newMeta = tuple2.getListenerMetaData(); if (newMeta != null) listenerMetaData.addAll(newMeta); } Collection<Persistable> listenerStaticInfo = tuple1.getListenerStaticInfo(); if (listenerStaticInfo == null) listenerStaticInfo = tuple2.getListenerStaticInfo(); else { Collection<Persistable> newStatic = tuple2.getListenerStaticInfo(); if (newStatic != null) listenerStaticInfo.addAll(newStatic); } Collection<Persistable> listenerUpdates = tuple1.getListenerUpdates(); if (listenerUpdates == null) listenerUpdates = tuple2.getListenerUpdates(); else { Collection<Persistable> listenerUpdates2 = tuple2.getListenerUpdates(); if (listenerUpdates2 != null) listenerUpdates.addAll(listenerUpdates2); } Map<String,Integer> minibatchesPerExecutor = new HashMap<>(); if(tuple1.getMinibatchesPerExecutor() != null) { for (Map.Entry<String, Integer> e : tuple1.getMinibatchesPerExecutor().entrySet()){ minibatchesPerExecutor.put(e.getKey(), e.getValue()); } } if(tuple2.getMinibatchesPerExecutor() != null){ for (Map.Entry<String, Integer> e : tuple2.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 = null; if(tuple1.getThresholdAlgorithmReducer() != null){ thresholdAlgorithmReducer = tuple1.getThresholdAlgorithmReducer(); } if(tuple2.getThresholdAlgorithmReducer() != null){ if(thresholdAlgorithmReducer == null){ thresholdAlgorithmReducer = tuple2.getThresholdAlgorithmReducer(); } else { thresholdAlgorithmReducer = thresholdAlgorithmReducer.merge(tuple2.getThresholdAlgorithmReducer()); } } return SharedTrainingAccumulationTuple.builder().scoreSum(score).updaterStateArray(stateView) .aggregationsCount(aggregationsCount).sparkTrainingStats(stats) .listenerMetaData(listenerMetaData).listenerUpdates(listenerUpdates) .listenerStaticInfo(listenerStaticInfo) .minibatchesPerExecutor(minibatchesPerExecutor) .thresholdAlgorithmReducer(thresholdAlgorithmReducer) .build(); } }
SharedTrainingAccumulationFunction implements
                Function2<SharedTrainingAccumulationTuple, SharedTrainingAccumulationTuple, SharedTrainingAccumulationTuple> { @Override public SharedTrainingAccumulationTuple call(SharedTrainingAccumulationTuple tuple1, SharedTrainingAccumulationTuple tuple2) throws Exception { if (tuple1 == null) return tuple2; else if (tuple2 == null) return tuple1; double score = 0.0; INDArray stateView = null; int aggregationsCount = 0; if (tuple1.getUpdaterStateArray() != null && tuple2.getUpdaterStateArray() != null) { stateView = tuple1.getUpdaterStateArray().addi(tuple2.getUpdaterStateArray()); } else if (tuple1.getUpdaterStateArray() != null || tuple2.getUpdaterStateArray() != null) { stateView = tuple1.getUpdaterStateArray() != null ? tuple1.getUpdaterStateArray() : tuple2.getUpdaterStateArray(); } aggregationsCount = tuple1.getAggregationsCount() + tuple2.getAggregationsCount(); score = tuple1.getScoreSum() + tuple2.getScoreSum(); SparkTrainingStats stats = tuple1.getSparkTrainingStats(); if (tuple2.getSparkTrainingStats() != null) { if (stats == null) stats = tuple2.getSparkTrainingStats(); else stats.addOtherTrainingStats(tuple2.getSparkTrainingStats()); } Nd4j.getExecutioner().commit(); Collection<StorageMetaData> listenerMetaData = tuple1.getListenerMetaData(); if (listenerMetaData == null) listenerMetaData = tuple2.getListenerMetaData(); else { Collection<StorageMetaData> newMeta = tuple2.getListenerMetaData(); if (newMeta != null) listenerMetaData.addAll(newMeta); } Collection<Persistable> listenerStaticInfo = tuple1.getListenerStaticInfo(); if (listenerStaticInfo == null) listenerStaticInfo = tuple2.getListenerStaticInfo(); else { Collection<Persistable> newStatic = tuple2.getListenerStaticInfo(); if (newStatic != null) listenerStaticInfo.addAll(newStatic); } Collection<Persistable> listenerUpdates = tuple1.getListenerUpdates(); if (listenerUpdates == null) listenerUpdates = tuple2.getListenerUpdates(); else { Collection<Persistable> listenerUpdates2 = tuple2.getListenerUpdates(); if (listenerUpdates2 != null) listenerUpdates.addAll(listenerUpdates2); } Map<String,Integer> minibatchesPerExecutor = new HashMap<>(); if(tuple1.getMinibatchesPerExecutor() != null) { for (Map.Entry<String, Integer> e : tuple1.getMinibatchesPerExecutor().entrySet()){ minibatchesPerExecutor.put(e.getKey(), e.getValue()); } } if(tuple2.getMinibatchesPerExecutor() != null){ for (Map.Entry<String, Integer> e : tuple2.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 = null; if(tuple1.getThresholdAlgorithmReducer() != null){ thresholdAlgorithmReducer = tuple1.getThresholdAlgorithmReducer(); } if(tuple2.getThresholdAlgorithmReducer() != null){ if(thresholdAlgorithmReducer == null){ thresholdAlgorithmReducer = tuple2.getThresholdAlgorithmReducer(); } else { thresholdAlgorithmReducer = thresholdAlgorithmReducer.merge(tuple2.getThresholdAlgorithmReducer()); } } return SharedTrainingAccumulationTuple.builder().scoreSum(score).updaterStateArray(stateView) .aggregationsCount(aggregationsCount).sparkTrainingStats(stats) .listenerMetaData(listenerMetaData).listenerUpdates(listenerUpdates) .listenerStaticInfo(listenerStaticInfo) .minibatchesPerExecutor(minibatchesPerExecutor) .thresholdAlgorithmReducer(thresholdAlgorithmReducer) .build(); }  }
SharedTrainingAccumulationFunction implements
                Function2<SharedTrainingAccumulationTuple, SharedTrainingAccumulationTuple, SharedTrainingAccumulationTuple> { @Override public SharedTrainingAccumulationTuple call(SharedTrainingAccumulationTuple tuple1, SharedTrainingAccumulationTuple tuple2) throws Exception { if (tuple1 == null) return tuple2; else if (tuple2 == null) return tuple1; double score = 0.0; INDArray stateView = null; int aggregationsCount = 0; if (tuple1.getUpdaterStateArray() != null && tuple2.getUpdaterStateArray() != null) { stateView = tuple1.getUpdaterStateArray().addi(tuple2.getUpdaterStateArray()); } else if (tuple1.getUpdaterStateArray() != null || tuple2.getUpdaterStateArray() != null) { stateView = tuple1.getUpdaterStateArray() != null ? tuple1.getUpdaterStateArray() : tuple2.getUpdaterStateArray(); } aggregationsCount = tuple1.getAggregationsCount() + tuple2.getAggregationsCount(); score = tuple1.getScoreSum() + tuple2.getScoreSum(); SparkTrainingStats stats = tuple1.getSparkTrainingStats(); if (tuple2.getSparkTrainingStats() != null) { if (stats == null) stats = tuple2.getSparkTrainingStats(); else stats.addOtherTrainingStats(tuple2.getSparkTrainingStats()); } Nd4j.getExecutioner().commit(); Collection<StorageMetaData> listenerMetaData = tuple1.getListenerMetaData(); if (listenerMetaData == null) listenerMetaData = tuple2.getListenerMetaData(); else { Collection<StorageMetaData> newMeta = tuple2.getListenerMetaData(); if (newMeta != null) listenerMetaData.addAll(newMeta); } Collection<Persistable> listenerStaticInfo = tuple1.getListenerStaticInfo(); if (listenerStaticInfo == null) listenerStaticInfo = tuple2.getListenerStaticInfo(); else { Collection<Persistable> newStatic = tuple2.getListenerStaticInfo(); if (newStatic != null) listenerStaticInfo.addAll(newStatic); } Collection<Persistable> listenerUpdates = tuple1.getListenerUpdates(); if (listenerUpdates == null) listenerUpdates = tuple2.getListenerUpdates(); else { Collection<Persistable> listenerUpdates2 = tuple2.getListenerUpdates(); if (listenerUpdates2 != null) listenerUpdates.addAll(listenerUpdates2); } Map<String,Integer> minibatchesPerExecutor = new HashMap<>(); if(tuple1.getMinibatchesPerExecutor() != null) { for (Map.Entry<String, Integer> e : tuple1.getMinibatchesPerExecutor().entrySet()){ minibatchesPerExecutor.put(e.getKey(), e.getValue()); } } if(tuple2.getMinibatchesPerExecutor() != null){ for (Map.Entry<String, Integer> e : tuple2.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 = null; if(tuple1.getThresholdAlgorithmReducer() != null){ thresholdAlgorithmReducer = tuple1.getThresholdAlgorithmReducer(); } if(tuple2.getThresholdAlgorithmReducer() != null){ if(thresholdAlgorithmReducer == null){ thresholdAlgorithmReducer = tuple2.getThresholdAlgorithmReducer(); } else { thresholdAlgorithmReducer = thresholdAlgorithmReducer.merge(tuple2.getThresholdAlgorithmReducer()); } } return SharedTrainingAccumulationTuple.builder().scoreSum(score).updaterStateArray(stateView) .aggregationsCount(aggregationsCount).sparkTrainingStats(stats) .listenerMetaData(listenerMetaData).listenerUpdates(listenerUpdates) .listenerStaticInfo(listenerStaticInfo) .minibatchesPerExecutor(minibatchesPerExecutor) .thresholdAlgorithmReducer(thresholdAlgorithmReducer) .build(); }  @Override SharedTrainingAccumulationTuple call(SharedTrainingAccumulationTuple tuple1,
                    SharedTrainingAccumulationTuple tuple2); }
SharedTrainingAccumulationFunction implements
                Function2<SharedTrainingAccumulationTuple, SharedTrainingAccumulationTuple, SharedTrainingAccumulationTuple> { @Override public SharedTrainingAccumulationTuple call(SharedTrainingAccumulationTuple tuple1, SharedTrainingAccumulationTuple tuple2) throws Exception { if (tuple1 == null) return tuple2; else if (tuple2 == null) return tuple1; double score = 0.0; INDArray stateView = null; int aggregationsCount = 0; if (tuple1.getUpdaterStateArray() != null && tuple2.getUpdaterStateArray() != null) { stateView = tuple1.getUpdaterStateArray().addi(tuple2.getUpdaterStateArray()); } else if (tuple1.getUpdaterStateArray() != null || tuple2.getUpdaterStateArray() != null) { stateView = tuple1.getUpdaterStateArray() != null ? tuple1.getUpdaterStateArray() : tuple2.getUpdaterStateArray(); } aggregationsCount = tuple1.getAggregationsCount() + tuple2.getAggregationsCount(); score = tuple1.getScoreSum() + tuple2.getScoreSum(); SparkTrainingStats stats = tuple1.getSparkTrainingStats(); if (tuple2.getSparkTrainingStats() != null) { if (stats == null) stats = tuple2.getSparkTrainingStats(); else stats.addOtherTrainingStats(tuple2.getSparkTrainingStats()); } Nd4j.getExecutioner().commit(); Collection<StorageMetaData> listenerMetaData = tuple1.getListenerMetaData(); if (listenerMetaData == null) listenerMetaData = tuple2.getListenerMetaData(); else { Collection<StorageMetaData> newMeta = tuple2.getListenerMetaData(); if (newMeta != null) listenerMetaData.addAll(newMeta); } Collection<Persistable> listenerStaticInfo = tuple1.getListenerStaticInfo(); if (listenerStaticInfo == null) listenerStaticInfo = tuple2.getListenerStaticInfo(); else { Collection<Persistable> newStatic = tuple2.getListenerStaticInfo(); if (newStatic != null) listenerStaticInfo.addAll(newStatic); } Collection<Persistable> listenerUpdates = tuple1.getListenerUpdates(); if (listenerUpdates == null) listenerUpdates = tuple2.getListenerUpdates(); else { Collection<Persistable> listenerUpdates2 = tuple2.getListenerUpdates(); if (listenerUpdates2 != null) listenerUpdates.addAll(listenerUpdates2); } Map<String,Integer> minibatchesPerExecutor = new HashMap<>(); if(tuple1.getMinibatchesPerExecutor() != null) { for (Map.Entry<String, Integer> e : tuple1.getMinibatchesPerExecutor().entrySet()){ minibatchesPerExecutor.put(e.getKey(), e.getValue()); } } if(tuple2.getMinibatchesPerExecutor() != null){ for (Map.Entry<String, Integer> e : tuple2.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 = null; if(tuple1.getThresholdAlgorithmReducer() != null){ thresholdAlgorithmReducer = tuple1.getThresholdAlgorithmReducer(); } if(tuple2.getThresholdAlgorithmReducer() != null){ if(thresholdAlgorithmReducer == null){ thresholdAlgorithmReducer = tuple2.getThresholdAlgorithmReducer(); } else { thresholdAlgorithmReducer = thresholdAlgorithmReducer.merge(tuple2.getThresholdAlgorithmReducer()); } } return SharedTrainingAccumulationTuple.builder().scoreSum(score).updaterStateArray(stateView) .aggregationsCount(aggregationsCount).sparkTrainingStats(stats) .listenerMetaData(listenerMetaData).listenerUpdates(listenerUpdates) .listenerStaticInfo(listenerStaticInfo) .minibatchesPerExecutor(minibatchesPerExecutor) .thresholdAlgorithmReducer(thresholdAlgorithmReducer) .build(); }  @Override SharedTrainingAccumulationTuple call(SharedTrainingAccumulationTuple tuple1,
                    SharedTrainingAccumulationTuple tuple2);  }