You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ch...@apache.org on 2018/01/09 16:45:03 UTC
[2/4] ignite git commit: IGNITE-7322: Return authorship of 7322 and
6783, step 1.
http://git-wip-us.apache.org/repos/asf/ignite/blob/0234ee3b/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/GroupTrainer.java
----------------------------------------------------------------------
diff --git a/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/GroupTrainer.java b/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/GroupTrainer.java
deleted file mode 100644
index fb34bf7..0000000
--- a/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/GroupTrainer.java
+++ /dev/null
@@ -1,208 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.ml.trainers.group;
-
-import java.io.Serializable;
-import java.util.List;
-import java.util.UUID;
-import java.util.stream.Stream;
-import org.apache.ignite.Ignite;
-import org.apache.ignite.IgniteCache;
-import org.apache.ignite.ml.Model;
-import org.apache.ignite.ml.math.functions.IgniteFunction;
-import org.apache.ignite.ml.math.functions.IgniteSupplier;
-import org.apache.ignite.ml.trainers.Trainer;
-import org.apache.ignite.ml.trainers.group.chain.ComputationsChain;
-import org.apache.ignite.ml.trainers.group.chain.EntryAndContext;
-import org.apache.ignite.ml.trainers.group.chain.HasTrainingUUID;
-
-/**
- * Class encapsulating synchronous distributed group training.
- * Training is performed by following scheme:
- * 1. For specified set of keys distributed initialization is done. For each key some initialization result is returned.
- * 2. All initialization results are processed locally and reduced into some object of type I.
- * 3. While 'shouldContinue' condition is true, training loop step is executed.
- * 4. After loop is finished, data from each key from final key set is collected.
- * 5. Data collected on previous step is transformed into a model which is returned as final result.
- * Note that all methods returning functions, suppliers etc should return values with minimal dependencies because they are serialized
- * with all dependent objects.
- *
- * @param <LC> Type of local context of the training.
- * @param <K> Type of data in {@link GroupTrainerCacheKey} keys on which the training is done.
- * @param <V> Type of cache values on which the training is done.
- * @param <IN> Type of data returned after initializing of distributed context.
- * @param <R> Type of result returned after training from each node.
- * @param <I> Type of data which is fed into each training loop step and returned from it.
- * @param <M> Type of model returned after training.
- * @param <T> Type of input to this trainer.
- * @param <G> Type of distributed context which is needed for forming final result which is send from each node to trainer for final model creation.
- */
-abstract class GroupTrainer<LC extends HasTrainingUUID, K, V, IN extends Serializable, R extends Serializable, I extends Serializable, M extends Model, T extends GroupTrainerInput<K>, G> implements Trainer<M, T> {
- /**
- * Cache on which training is performed. For example it can be cache of neural networks.
- */
- protected IgniteCache<GroupTrainerCacheKey<K>, V> cache;
-
- /**
- * Ignite instance.
- */
- protected Ignite ignite;
-
- /**
- * Construct an instance of this class.
- *
- * @param cache Cache on which training is performed.
- * @param ignite Ignite instance.
- */
- GroupTrainer(
- IgniteCache<GroupTrainerCacheKey<K>, V> cache,
- Ignite ignite) {
- this.cache = cache;
- this.ignite = ignite;
- }
-
- /** {@inheritDoc} */
- @Override public final M train(T data) {
- UUID trainingUUID = UUID.randomUUID();
- LC locCtx = initialLocalContext(data, trainingUUID);
-
- GroupTrainingContext<K, V, LC> ctx = new GroupTrainingContext<>(locCtx, cache, ignite);
- ComputationsChain<LC, K, V, T, T> chain = (i, c) -> i;
- IgniteFunction<GroupTrainerCacheKey<K>, ResultAndUpdates<IN>> distributedInitializer
- = distributedInitializer(data);
-
- init(data, trainingUUID);
-
- M res = chain.
- thenDistributedForKeys(distributedInitializer, (t, lc) -> data.initialKeys(trainingUUID),
- reduceDistributedInitData()).
- thenLocally(this::locallyProcessInitData).
- thenWhile(this::shouldContinue, trainingLoopStep()).
- thenDistributedForEntries(this::extractContextForFinalResultCreation, finalResultsExtractor(),
- this::finalResultKeys, finalResultsReducer()).
- thenLocally(this::mapFinalResult).
- process(data, ctx);
-
- cleanup(locCtx);
-
- return res;
- }
-
- /**
- * Create initial local context from data given as input to trainer.
- *
- * @param data Data given as input to this trainer.
- * @param trainingUUID UUID of this training.
- * @return Initial local context.
- */
- protected abstract LC initialLocalContext(T data, UUID trainingUUID);
-
- /** Override in subclasses if needed. */
- protected void init(T data, UUID trainingUUID) {
- }
-
- /**
- * Get function for initialization for each of keys specified in initial key set.
- *
- * @param data Data given to this trainer as input.
- * @return Function for initialization for each of keys specified in initial key set.
- */
- protected abstract IgniteFunction<GroupTrainerCacheKey<K>, ResultAndUpdates<IN>> distributedInitializer(T data);
-
- /**
- * Get reducer to reduce data collected from initialization of each key specified in initial key set.
- *
- * @return Reducer to reduce data collected from initialization of each key specified in initial key set.
- */
- protected abstract IgniteFunction<List<IN>, IN> reduceDistributedInitData();
-
- /**
- * Transform data from initialization step into data which is fed as input to first step of training loop.
- *
- * @param data Data from initialization step.
- * @param locCtx Local context.
- * @return Data which is fed as input to first step of training loop.
- */
- protected abstract I locallyProcessInitData(IN data, LC locCtx);
-
- /**
- * Training loop step.
- *
- * @return Result of training loop step.
- */
- protected abstract ComputationsChain<LC, K, V, I, I> trainingLoopStep();
-
- /**
- * Condition specifying if training loop should continue.
- *
- * @param data First time, data returned by locallyProcessInitData then data returned by last step of loop.
- * @param locCtx Local context.
- * @return Boolean value indicating if training loop should continue.
- */
- protected abstract boolean shouldContinue(I data, LC locCtx);
-
- /**
- * Extract context for final result creation. Each key from the final keys set will be processed with
- * finalResultsExtractor. While entry data (i.e. key and value) for each key varies, some data can be common for all
- * processed entries. This data is called context.
- *
- * @param data Data returned from last training loop step.
- * @param locCtx Local context.
- * @return Context.
- */
- protected abstract IgniteSupplier<G> extractContextForFinalResultCreation(I data, LC locCtx);
-
- /**
- * Keys for final result creation.
- *
- * @param data Data returned from the last training loop step.
- * @param locCtx Local context.
- * @return Stream of keys for final result creation.
- */
- protected abstract IgniteSupplier<Stream<GroupTrainerCacheKey<K>>> finalResultKeys(I data, LC locCtx);
-
- /**
- * Get function for extracting final result from each key specified in finalResultKeys.
- *
- * @return Function for extracting final result from each key specified in finalResultKeys.
- */
- protected abstract IgniteFunction<EntryAndContext<K, V, G>, ResultAndUpdates<R>> finalResultsExtractor();
-
- /**
- * Get function for reducing final results.
- *
- * @return Function for reducing final results.
- */
- protected abstract IgniteFunction<List<R>, R> finalResultsReducer();
-
- /**
- * Map final result to model which is returned by trainer.
- *
- * @param res Final result.
- * @param locCtx Local context.
- * @return Model resulted from training.
- */
- protected abstract M mapFinalResult(R res, LC locCtx);
-
- /**
- * Performs cleanups of temporary objects created by this trainer.
- *
- * @param locCtx Local context.
- */
- protected abstract void cleanup(LC locCtx);
-}
http://git-wip-us.apache.org/repos/asf/ignite/blob/0234ee3b/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/GroupTrainerBaseProcessorTask.java
----------------------------------------------------------------------
diff --git a/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/GroupTrainerBaseProcessorTask.java b/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/GroupTrainerBaseProcessorTask.java
deleted file mode 100644
index 755f200..0000000
--- a/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/GroupTrainerBaseProcessorTask.java
+++ /dev/null
@@ -1,143 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.ml.trainers.group;
-
-import java.io.Serializable;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.UUID;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
-import org.apache.ignite.Ignite;
-import org.apache.ignite.IgniteException;
-import org.apache.ignite.cache.affinity.Affinity;
-import org.apache.ignite.cluster.ClusterNode;
-import org.apache.ignite.compute.ComputeJob;
-import org.apache.ignite.compute.ComputeJobResult;
-import org.apache.ignite.compute.ComputeTaskAdapter;
-import org.apache.ignite.ml.math.functions.IgniteFunction;
-import org.apache.ignite.ml.math.functions.IgniteSupplier;
-import org.jetbrains.annotations.Nullable;
-
-/**
- * Base task for group trainer.
- *
- * @param <K> Type of cache keys of cache used for training.
- * @param <V> Type of cache values of cache used for training.
- * @param <C> Type of context (common part of data needed for computation).
- * @param <T> Type of arguments of workers.
- * @param <R> Type of computation result.
- */
-public abstract class GroupTrainerBaseProcessorTask<K, V, C, T, R extends Serializable> extends ComputeTaskAdapter<Void, R> {
- /**
- * Context supplier.
- */
- protected final IgniteSupplier<C> ctxSupplier;
-
- /**
- * UUID of training.
- */
- protected final UUID trainingUUID;
-
- /**
- * Worker.
- */
- protected IgniteFunction<T, ResultAndUpdates<R>> worker;
-
- /**
- * Reducer used for reducing of computations on specified keys.
- */
- protected final IgniteFunction<List<R>, R> reducer;
-
- /**
- * Name of cache on which training is done.
- */
- protected final String cacheName;
-
- /**
- * Supplier of keys on which worker should be executed.
- */
- protected final IgniteSupplier<Stream<GroupTrainerCacheKey<K>>> keysSupplier;
-
- /**
- * Ignite instance.
- */
- protected final Ignite ignite;
-
- /**
- * Construct an instance of this class with specified parameters.
- *
- * @param trainingUUID UUID of training.
- * @param ctxSupplier Supplier of context.
- * @param worker Function calculated on each of specified keys.
- * @param keysSupplier Supplier of keys on which training is done.
- * @param reducer Reducer used for reducing results of computation performed on each of specified keys.
- * @param cacheName Name of cache on which training is done.
- * @param ignite Ignite instance.
- */
- public GroupTrainerBaseProcessorTask(UUID trainingUUID,
- IgniteSupplier<C> ctxSupplier,
- IgniteFunction<T, ResultAndUpdates<R>> worker,
- IgniteSupplier<Stream<GroupTrainerCacheKey<K>>> keysSupplier,
- IgniteFunction<List<R>, R> reducer,
- String cacheName,
- Ignite ignite) {
- this.trainingUUID = trainingUUID;
- this.ctxSupplier = ctxSupplier;
- this.worker = worker;
- this.keysSupplier = keysSupplier;
- this.reducer = reducer;
- this.cacheName = cacheName;
- this.ignite = ignite;
- }
-
- /** {@inheritDoc} */
- @Nullable @Override public Map<? extends ComputeJob, ClusterNode> map(List<ClusterNode> subgrid,
- @Nullable Void arg) throws IgniteException {
- Map<ComputeJob, ClusterNode> res = new HashMap<>();
-
- for (ClusterNode node : subgrid) {
- BaseLocalProcessorJob<K, V, T, R> job = createJob();
- res.put(job, node);
- }
-
- return res;
- }
-
- /** {@inheritDoc} */
- @Nullable @Override public R reduce(List<ComputeJobResult> results) throws IgniteException {
- return reducer.apply(results.stream().map(res -> (R)res.getData()).collect(Collectors.toList()));
- }
-
- /**
- * Create job for execution on subgrid.
- *
- * @return Job for execution on subgrid.
- */
- protected abstract BaseLocalProcessorJob<K, V, T, R> createJob();
-
- /**
- * Get affinity function of cache on which training is done.
- *
- * @return Affinity function of cache on which training is done.
- */
- protected Affinity<GroupTrainerCacheKey> affinity() {
- return ignite.affinity(cacheName);
- }
-}
http://git-wip-us.apache.org/repos/asf/ignite/blob/0234ee3b/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/GroupTrainerCacheKey.java
----------------------------------------------------------------------
diff --git a/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/GroupTrainerCacheKey.java b/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/GroupTrainerCacheKey.java
deleted file mode 100644
index 5e4cb76..0000000
--- a/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/GroupTrainerCacheKey.java
+++ /dev/null
@@ -1,125 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.ml.trainers.group;
-
-import java.util.UUID;
-import org.apache.ignite.cache.affinity.AffinityKeyMapped;
-
-/**
- * Class used as a key for caches on which {@link GroupTrainer} works.
- * Structurally it is a triple: (nodeLocalEntityIndex, trainingUUID, data);
- * nodeLocalEntityIndex is used to map key to node;
- * trainingUUID is id of training;
- * data is some custom data stored in this key, for example if we want to store three neural networks on one node
- * for training with training UUID == trainingUUID, we can use keys
- * (1, trainingUUID, networkIdx1), (1, trainingUUID, networkIdx2), (1, trainingUUID, networkIdx3).
- *
- * @param <K> Type of data part of this key.
- */
-public class GroupTrainerCacheKey<K> {
- /**
- * Part of key for key-to-node affinity.
- */
- @AffinityKeyMapped
- private Long nodeLocEntityIdx;
-
- /**
- * UUID of training.
- */
- private UUID trainingUUID;
-
- /**
- * Data.
- */
- K data;
-
- /**
- * Construct instance of this class.
- *
- * @param nodeLocEntityIdx Part of key for key-to-node affinity.
- * @param data Data.
- * @param trainingUUID Training UUID.
- */
- public GroupTrainerCacheKey(long nodeLocEntityIdx, K data, UUID trainingUUID) {
- this.nodeLocEntityIdx = nodeLocEntityIdx;
- this.trainingUUID = trainingUUID;
- this.data = data;
- }
-
- /**
- * Construct instance of this class.
- *
- * @param nodeLocEntityIdx Part of key for key-to-node affinity.
- * @param data Data.
- * @param trainingUUID Training UUID.
- */
- public GroupTrainerCacheKey(int nodeLocEntityIdx, K data, UUID trainingUUID) {
- this((long)nodeLocEntityIdx, data, trainingUUID);
- }
-
- /**
- * Get part of key used for key-to-node affinity.
- *
- * @return Part of key used for key-to-node affinity.
- */
- public Long nodeLocalEntityIndex() {
- return nodeLocEntityIdx;
- }
-
- /**
- * Get UUID of training.
- *
- * @return UUID of training.
- */
- public UUID trainingUUID() {
- return trainingUUID;
- }
-
- /**
- * Get data.
- *
- * @return Data.
- */
- public K data() {
- return data;
- }
-
- /** {@inheritDoc} */
- @Override public boolean equals(Object o) {
- if (this == o)
- return true;
- if (o == null || getClass() != o.getClass())
- return false;
-
- GroupTrainerCacheKey<?> key = (GroupTrainerCacheKey<?>)o;
-
- if (nodeLocEntityIdx != null ? !nodeLocEntityIdx.equals(key.nodeLocEntityIdx) : key.nodeLocEntityIdx != null)
- return false;
- if (trainingUUID != null ? !trainingUUID.equals(key.trainingUUID) : key.trainingUUID != null)
- return false;
- return data != null ? data.equals(key.data) : key.data == null;
- }
-
- /** {@inheritDoc} */
- @Override public int hashCode() {
- int res = nodeLocEntityIdx != null ? nodeLocEntityIdx.hashCode() : 0;
- res = 31 * res + (trainingUUID != null ? trainingUUID.hashCode() : 0);
- res = 31 * res + (data != null ? data.hashCode() : 0);
- return res;
- }
-}
http://git-wip-us.apache.org/repos/asf/ignite/blob/0234ee3b/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/GroupTrainerEntriesProcessorTask.java
----------------------------------------------------------------------
diff --git a/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/GroupTrainerEntriesProcessorTask.java b/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/GroupTrainerEntriesProcessorTask.java
deleted file mode 100644
index daa396f..0000000
--- a/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/GroupTrainerEntriesProcessorTask.java
+++ /dev/null
@@ -1,64 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.ml.trainers.group;
-
-import java.io.Serializable;
-import java.util.List;
-import java.util.UUID;
-import java.util.stream.Stream;
-import org.apache.ignite.Ignite;
-import org.apache.ignite.ml.math.functions.IgniteFunction;
-import org.apache.ignite.ml.math.functions.IgniteSupplier;
-import org.apache.ignite.ml.trainers.group.chain.EntryAndContext;
-
-/**
- * Task for processing entries of cache used for training.
- *
- * @param <K> Type of cache keys of cache used for training.
- * @param <V> Type of cache values of cache used for training.
- * @param <C> Type of context (common part of data needed for computation).
- * @param <R> Type of computation result.
- */
-public class GroupTrainerEntriesProcessorTask<K, V, C, R extends Serializable>
- extends GroupTrainerBaseProcessorTask<K, V, C, EntryAndContext<K, V, C>, R> {
- /**
- * Construct instance of this class with given parameters.
- *
- * @param trainingUUID UUID of training.
- * @param ctxSupplier Supplier of context.
- * @param worker Function calculated on each of specified keys.
- * @param keysSupplier Supplier of keys on which training is done.
- * @param reducer Reducer used for reducing results of computation performed on each of specified keys.
- * @param cacheName Name of cache on which training is done.
- * @param ignite Ignite instance.
- */
- public GroupTrainerEntriesProcessorTask(UUID trainingUUID,
- IgniteSupplier<C> ctxSupplier,
- IgniteFunction<EntryAndContext<K, V, C>, ResultAndUpdates<R>> worker,
- IgniteSupplier<Stream<GroupTrainerCacheKey<K>>> keysSupplier,
- IgniteFunction<List<R>, R> reducer,
- String cacheName,
- Ignite ignite) {
- super(trainingUUID, ctxSupplier, worker, keysSupplier, reducer, cacheName, ignite);
- }
-
- /** {@inheritDoc} */
- @Override protected BaseLocalProcessorJob<K, V, EntryAndContext<K, V, C>, R> createJob() {
- return new LocalEntriesProcessorJob<>(ctxSupplier, worker, keysSupplier, reducer, trainingUUID, cacheName);
- }
-}
http://git-wip-us.apache.org/repos/asf/ignite/blob/0234ee3b/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/GroupTrainerInput.java
----------------------------------------------------------------------
diff --git a/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/GroupTrainerInput.java b/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/GroupTrainerInput.java
deleted file mode 100644
index ae75f16..0000000
--- a/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/GroupTrainerInput.java
+++ /dev/null
@@ -1,37 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.ml.trainers.group;
-
-import java.util.UUID;
-import java.util.stream.Stream;
-import org.apache.ignite.ml.math.functions.IgniteSupplier;
-
-/**
- * Interface for {@link GroupTrainer} inputs.
- *
- * @param <K> Types of cache keys used for group training.
- */
-public interface GroupTrainerInput<K> {
- /**
- * Get supplier of stream of keys used for initialization of {@link GroupTrainer}.
- *
- * @param trainingUUID UUID of training.
- * @return Supplier of stream of keys used for initialization of {@link GroupTrainer}.
- */
- IgniteSupplier<Stream<GroupTrainerCacheKey<K>>> initialKeys(UUID trainingUUID);
-}
http://git-wip-us.apache.org/repos/asf/ignite/blob/0234ee3b/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/GroupTrainerKeysProcessorTask.java
----------------------------------------------------------------------
diff --git a/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/GroupTrainerKeysProcessorTask.java b/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/GroupTrainerKeysProcessorTask.java
deleted file mode 100644
index 7ac18f8..0000000
--- a/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/GroupTrainerKeysProcessorTask.java
+++ /dev/null
@@ -1,62 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.ml.trainers.group;
-
-import java.io.Serializable;
-import java.util.List;
-import java.util.UUID;
-import java.util.stream.Stream;
-import org.apache.ignite.Ignite;
-import org.apache.ignite.ml.math.functions.IgniteFunction;
-import org.apache.ignite.ml.math.functions.IgniteSupplier;
-import org.apache.ignite.ml.trainers.group.chain.KeyAndContext;
-
-/**
- * Task for processing entries of cache used for training.
- *
- * @param <K> Type of cache keys of cache used for training.
- * @param <C> Type of context (common part of data needed for computation).
- * @param <R> Type of computation result.
- */
-public class GroupTrainerKeysProcessorTask<K, C, R extends Serializable> extends GroupTrainerBaseProcessorTask<K, Object, C, KeyAndContext<K, C>, R> {
- /**
- * Construct instance of this class with specified parameters.
- *
- * @param trainingUUID UUID of training.
- * @param ctxSupplier Context supplier.
- * @param worker Function calculated on each of specified keys.
- * @param keysSupplier Supplier of keys on which computations should be done.
- * @param reducer Reducer used for reducing results of computation performed on each of specified keys.
- * @param cacheName Name of cache on which training is done.
- * @param ignite Ignite instance.
- */
- public GroupTrainerKeysProcessorTask(UUID trainingUUID,
- IgniteSupplier<C> ctxSupplier,
- IgniteFunction<KeyAndContext<K, C>, ResultAndUpdates<R>> worker,
- IgniteSupplier<Stream<GroupTrainerCacheKey<K>>> keysSupplier,
- IgniteFunction<List<R>, R> reducer,
- String cacheName,
- Ignite ignite) {
- super(trainingUUID, ctxSupplier, worker, keysSupplier, reducer, cacheName, ignite);
- }
-
- /** {@inheritDoc} */
- @Override protected BaseLocalProcessorJob<K, Object, KeyAndContext<K, C>, R> createJob() {
- return new LocalKeysProcessorJob<>(ctxSupplier, worker, keysSupplier, reducer, trainingUUID, cacheName);
- }
-}
http://git-wip-us.apache.org/repos/asf/ignite/blob/0234ee3b/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/GroupTrainingContext.java
----------------------------------------------------------------------
diff --git a/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/GroupTrainingContext.java b/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/GroupTrainingContext.java
deleted file mode 100644
index cbd04b2..0000000
--- a/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/GroupTrainingContext.java
+++ /dev/null
@@ -1,98 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.ml.trainers.group;
-
-import org.apache.ignite.Ignite;
-import org.apache.ignite.IgniteCache;
-import org.apache.ignite.ml.trainers.group.chain.HasTrainingUUID;
-
-/**
- * Context for group training.
- *
- * @param <K> Type of keys of cache used for group training.
- * @param <V> Type of values of cache used for group training.
- * @param <L> Type of local context used for training.
- */
-public class GroupTrainingContext<K, V, L extends HasTrainingUUID> {
- /**
- * Local context.
- */
- private L locCtx;
-
- /**
- * Cache used for training.
- */
- private IgniteCache<GroupTrainerCacheKey<K>, V> cache;
-
- /**
- * Ignite instance.
- */
- private Ignite ignite;
-
- /**
- * Construct instance of this class.
- *
- * @param locCtx Local context.
- * @param cache Information about cache used for training.
- * @param ignite Ignite instance.
- */
- public GroupTrainingContext(L locCtx, IgniteCache<GroupTrainerCacheKey<K>, V> cache, Ignite ignite) {
- this.locCtx = locCtx;
- this.cache = cache;
- this.ignite = ignite;
- }
-
- /**
- * Construct new training context with same parameters but with new cache.
- *
- * @param newCache New cache.
- * @param <K1> Type of keys of new cache.
- * @param <V1> Type of values of new cache.
- * @return New training context with same parameters but with new cache.
- */
- public <K1, V1> GroupTrainingContext<K1, V1, L> withCache(IgniteCache<GroupTrainerCacheKey<K1>, V1> newCache) {
- return new GroupTrainingContext<>(locCtx, newCache, ignite);
- }
-
- /**
- * Get local context.
- *
- * @return Local context.
- */
- public L localContext() {
- return locCtx;
- }
-
- /**
- * Get cache used for training.
- *
- * @return Cache used for training.
- */
- public IgniteCache<GroupTrainerCacheKey<K>, V> cache() {
- return cache;
- }
-
- /**
- * Get ignite instance.
- *
- * @return Ignite instance.
- */
- public Ignite ignite() {
- return ignite;
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/0234ee3b/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/LocalEntriesProcessorJob.java
----------------------------------------------------------------------
diff --git a/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/LocalEntriesProcessorJob.java b/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/LocalEntriesProcessorJob.java
deleted file mode 100644
index d035aa5..0000000
--- a/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/LocalEntriesProcessorJob.java
+++ /dev/null
@@ -1,85 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.ml.trainers.group;
-
-import java.io.Serializable;
-import java.util.List;
-import java.util.Map;
-import java.util.Objects;
-import java.util.Set;
-import java.util.UUID;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
-import org.apache.ignite.ml.math.functions.IgniteFunction;
-import org.apache.ignite.ml.math.functions.IgniteSupplier;
-import org.apache.ignite.ml.trainers.group.chain.EntryAndContext;
-
-/**
- * {@link BaseLocalProcessorJob} specified to entry processing.
- *
- * @param <K> Type of cache used for group training.
- * @param <V> Type of values used for group training.
- * @param <C> Type of context.
- * @param <R> Type of result returned by worker.
- */
-public class LocalEntriesProcessorJob<K, V, C, R extends Serializable> extends BaseLocalProcessorJob<K, V, EntryAndContext<K, V, C>, R> {
- /**
- * Supplier of context for worker.
- */
- private final IgniteSupplier<C> ctxSupplier;
-
- /**
- * Construct an instance of this class.
- *
- * @param ctxSupplier Supplier for context for worker.
- * @param worker Worker.
- * @param keySupplier Supplier of keys.
- * @param reducer Reducer.
- * @param trainingUUID UUID for training.
- * @param cacheName Name of cache used for training.
- */
- public LocalEntriesProcessorJob(IgniteSupplier<C> ctxSupplier,
- IgniteFunction<EntryAndContext<K, V, C>, ResultAndUpdates<R>> worker,
- IgniteSupplier<Stream<GroupTrainerCacheKey<K>>> keySupplier,
- IgniteFunction<List<R>, R> reducer,
- UUID trainingUUID, String cacheName) {
- super(worker, keySupplier, reducer, trainingUUID, cacheName);
- this.ctxSupplier = ctxSupplier;
- }
-
- /** {@inheritDoc} */
- @Override protected Stream<EntryAndContext<K, V, C>> toProcess() {
- C ctx = ctxSupplier.get();
-
- return selectLocalEntries().map(e -> new EntryAndContext<>(e, ctx));
- }
-
- /**
- * Select entries for processing by worker.
- *
- * @return Entries for processing by worker.
- */
- private Stream<Map.Entry<GroupTrainerCacheKey<K>, V>> selectLocalEntries() {
- Set<GroupTrainerCacheKey<K>> keys = keySupplier.get().
- filter(k -> Objects.requireNonNull(affinity().mapKeyToNode(k)).isLocal()).
- filter(k -> k.trainingUUID().equals(trainingUUID)).
- collect(Collectors.toSet());
-
- return cache().getAll(keys).entrySet().stream();
- }
-}
http://git-wip-us.apache.org/repos/asf/ignite/blob/0234ee3b/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/LocalKeysProcessorJob.java
----------------------------------------------------------------------
diff --git a/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/LocalKeysProcessorJob.java b/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/LocalKeysProcessorJob.java
deleted file mode 100644
index cad53c9..0000000
--- a/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/LocalKeysProcessorJob.java
+++ /dev/null
@@ -1,78 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.ml.trainers.group;
-
-import java.io.Serializable;
-import java.util.List;
-import java.util.Objects;
-import java.util.UUID;
-import java.util.stream.Stream;
-import org.apache.ignite.ml.math.functions.IgniteFunction;
-import org.apache.ignite.ml.math.functions.IgniteSupplier;
-import org.apache.ignite.ml.trainers.group.chain.KeyAndContext;
-
-/**
- * {@link BaseLocalProcessorJob} specified to keys processing.
- *
- * @param <K> Type of cache used for group training.
- * @param <V> Type of values used for group training.
- * @param <C> Type of context.
- * @param <R> Type of result returned by worker.
- */
-public class LocalKeysProcessorJob<K, V, C, R extends Serializable> extends BaseLocalProcessorJob<K, V, KeyAndContext<K, C>, R> {
- /**
- * Supplier of worker context.
- */
- private final IgniteSupplier<C> ctxSupplier;
-
- /**
- * Construct instance of this class with given arguments.
- *
- * @param worker Worker.
- * @param keySupplier Supplier of keys.
- * @param reducer Reducer.
- * @param trainingUUID UUID of training.
- * @param cacheName Name of cache used for training.
- */
- public LocalKeysProcessorJob(IgniteSupplier<C> ctxSupplier,
- IgniteFunction<KeyAndContext<K, C>, ResultAndUpdates<R>> worker,
- IgniteSupplier<Stream<GroupTrainerCacheKey<K>>> keySupplier,
- IgniteFunction<List<R>, R> reducer,
- UUID trainingUUID, String cacheName) {
- super(worker, keySupplier, reducer, trainingUUID, cacheName);
- this.ctxSupplier = ctxSupplier;
- }
-
- /** {@inheritDoc} */
- @Override protected Stream<KeyAndContext<K, C>> toProcess() {
- C ctx = ctxSupplier.get();
-
- return selectLocalKeys().map(k -> new KeyAndContext<>(k, ctx));
- }
-
- /**
- * Get subset of keys provided by keySupplier which are mapped to node on which code is executed.
- *
- * @return Subset of keys provided by keySupplier which are mapped to node on which code is executed.
- */
- private Stream<GroupTrainerCacheKey<K>> selectLocalKeys() {
- return keySupplier.get().
- filter(k -> Objects.requireNonNull(affinity().mapKeyToNode(k)).isLocal()).
- filter(k -> k.trainingUUID().equals(trainingUUID));
- }
-}
http://git-wip-us.apache.org/repos/asf/ignite/blob/0234ee3b/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/Metaoptimizer.java
----------------------------------------------------------------------
diff --git a/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/Metaoptimizer.java b/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/Metaoptimizer.java
deleted file mode 100644
index 0ab6d32..0000000
--- a/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/Metaoptimizer.java
+++ /dev/null
@@ -1,93 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.ml.trainers.group;
-
-import java.util.List;
-import org.apache.ignite.ml.math.functions.IgniteFunction;
-
-/**
- * Class encapsulating data transformations in group training in {@link MetaoptimizerGroupTrainer}, which is adapter of
- * {@link GroupTrainer}.
- *
- * @param <LC> Local context of {@link GroupTrainer}.
- * @param <X> Type of data which is processed in training loop step.
- * @param <Y> Type of data returned by training loop step data processor.
- * @param <I> Type of data to which data returned by distributed initialization is mapped.
- * @param <D> Type of data returned by initialization.
- * @param <O> Type of data to which data returned by data processor is mapped.
- */
-public interface Metaoptimizer<LC, X, Y, I, D, O> {
- /**
- * Get function used to reduce distributed initialization results.
- *
- * @return Function used to reduce distributed initialization results.
- */
- IgniteFunction<List<D>, D> initialReducer();
-
- /**
- * Maps data returned by distributed initialization to data consumed by training loop step.
- *
- * @param data Data returned by distributed initialization.
- * @param locCtx Local context.
- * @return Mapping of data returned by distributed initialization to data consumed by training loop step.
- */
- I locallyProcessInitData(D data, LC locCtx);
-
- /**
- * Preprocess data for {@link MetaoptimizerGroupTrainer#dataProcessor()}.
- *
- * @return Preprocessed data for {@link MetaoptimizerGroupTrainer#dataProcessor()}.
- */
- default IgniteFunction<X, X> distributedPreprocessor() {
- return x -> x;
- }
-
- /**
- * Get function used to map values returned by {@link MetaoptimizerGroupTrainer#dataProcessor()}.
- *
- * @return Function used to map values returned by {@link MetaoptimizerGroupTrainer#dataProcessor()}.
- */
- IgniteFunction<Y, O> distributedPostprocessor();
-
- /**
- * Get binary operator used for reducing results returned by distributedPostprocessor.
- *
- * @return Binary operator used for reducing results returned by distributedPostprocessor.
- */
- IgniteFunction<List<O>, O> postProcessReducer();
-
- /**
- * Transform data returned by distributed part of training loop step into input fed into distributed part of training
- * loop step.
- *
- * @param input Type of output of distributed part of training loop step.
- * @param locCtx Local context.
- * @return Result of transform data returned by distributed part of training loop step into input fed into distributed part of training
- * loop step.
- */
- I localProcessor(O input, LC locCtx);
-
- /**
- * Returns value of predicate 'should training loop continue given previous step output and local context'.
- *
- * @param input Input of previous step.
- * @param locCtx Local context.
- * @return Value of predicate 'should training loop continue given previous step output and local context'.
- */
- boolean shouldContinue(I input, LC locCtx);
-}
http://git-wip-us.apache.org/repos/asf/ignite/blob/0234ee3b/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/MetaoptimizerDistributedStep.java
----------------------------------------------------------------------
diff --git a/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/MetaoptimizerDistributedStep.java b/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/MetaoptimizerDistributedStep.java
deleted file mode 100644
index 67dcf7f..0000000
--- a/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/MetaoptimizerDistributedStep.java
+++ /dev/null
@@ -1,88 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.ml.trainers.group;
-
-import java.io.Serializable;
-import java.util.List;
-import java.util.stream.Stream;
-import org.apache.ignite.ml.math.functions.IgniteFunction;
-import org.apache.ignite.ml.math.functions.IgniteSupplier;
-import org.apache.ignite.ml.trainers.group.chain.DistributedEntryProcessingStep;
-import org.apache.ignite.ml.trainers.group.chain.EntryAndContext;
-import org.apache.ignite.ml.trainers.group.chain.HasTrainingUUID;
-
-/**
- * Distributed step
- * TODO: IGNITE-7350: add full description.
- */
-class MetaoptimizerDistributedStep<L extends HasTrainingUUID, K, V, G, I extends Serializable, O extends Serializable,
- X, Y, D extends Serializable> implements DistributedEntryProcessingStep<L, K, V, G, I, O> {
- /**
- * {@link Metaoptimizer}.
- */
- private final Metaoptimizer<L, X, Y, I, D, O> metaoptimizer;
-
- /**
- * {@link MetaoptimizerGroupTrainer} for which this distributed step is used.
- */
- private final MetaoptimizerGroupTrainer<L, K, V, D, ?, I, ?, ?, G, O, X, Y> trainer;
-
- /**
- * Construct instance of this class with given parameters.
- *
- * @param metaoptimizer Metaoptimizer.
- * @param trainer {@link MetaoptimizerGroupTrainer} for which this distributed step is used.
- */
- public MetaoptimizerDistributedStep(Metaoptimizer<L, X, Y, I, D, O> metaoptimizer,
- MetaoptimizerGroupTrainer<L, K, V, D, ?, I, ?, ?, G, O, X, Y> trainer) {
- this.metaoptimizer = metaoptimizer;
- this.trainer = trainer;
- }
-
- /** {@inheritDoc} */
- @Override public IgniteSupplier<G> remoteContextSupplier(I input, L locCtx) {
- return trainer.remoteContextExtractor(input, locCtx);
- }
-
- /** {@inheritDoc} */
- @Override public IgniteFunction<EntryAndContext<K, V, G>, ResultAndUpdates<O>> worker() {
- IgniteFunction<X, ResultAndUpdates<Y>> dataProcessor = trainer.dataProcessor();
- IgniteFunction<X, X> preprocessor = metaoptimizer.distributedPreprocessor();
- IgniteFunction<Y, O> postprocessor = metaoptimizer.distributedPostprocessor();
- IgniteFunction<EntryAndContext<K, V, G>, X> ctxExtractor = trainer.trainingLoopStepDataExtractor();
-
- return entryAndCtx -> {
- X apply = ctxExtractor.apply(entryAndCtx);
- preprocessor.apply(apply);
- ResultAndUpdates<Y> res = dataProcessor.apply(apply);
- O postprocessRes = postprocessor.apply(res.result());
-
- return ResultAndUpdates.of(postprocessRes).setUpdates(res.updates());
- };
- }
-
- /** {@inheritDoc} */
- @Override public IgniteSupplier<Stream<GroupTrainerCacheKey<K>>> keys(I input, L locCtx) {
- return trainer.keysToProcessInTrainingLoop(locCtx);
- }
-
- /** {@inheritDoc} */
- @Override public IgniteFunction<List<O>, O> reducer() {
- return metaoptimizer.postProcessReducer();
- }
-}
http://git-wip-us.apache.org/repos/asf/ignite/blob/0234ee3b/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/MetaoptimizerGroupTrainer.java
----------------------------------------------------------------------
diff --git a/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/MetaoptimizerGroupTrainer.java b/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/MetaoptimizerGroupTrainer.java
deleted file mode 100644
index bebfe3e..0000000
--- a/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/MetaoptimizerGroupTrainer.java
+++ /dev/null
@@ -1,132 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.ml.trainers.group;
-
-import java.io.Serializable;
-import java.util.List;
-import java.util.UUID;
-import java.util.stream.Stream;
-import org.apache.ignite.Ignite;
-import org.apache.ignite.IgniteCache;
-import org.apache.ignite.ml.Model;
-import org.apache.ignite.ml.math.functions.IgniteFunction;
-import org.apache.ignite.ml.math.functions.IgniteSupplier;
-import org.apache.ignite.ml.trainers.group.chain.Chains;
-import org.apache.ignite.ml.trainers.group.chain.ComputationsChain;
-import org.apache.ignite.ml.trainers.group.chain.EntryAndContext;
-import org.apache.ignite.ml.trainers.group.chain.HasTrainingUUID;
-
-/**
- * Group trainer using {@link Metaoptimizer}.
- * Main purpose of this trainer is to extract various transformations (normalizations for example) of data which is processed
- * in the training loop step into distinct entity called metaoptimizer and only fix the main part of logic in
- * trainers extending this class. This way we'll be able to quickly switch between this transformations by using different metaoptimizers
- * without touching main logic.
- *
- * @param <LC> Type of local context.
- * @param <K> Type of data in {@link GroupTrainerCacheKey} keys on which the training is done.
- * @param <V> Type of values of cache used in group training.
- * @param <IN> Data type which is returned by distributed initializer.
- * @param <R> Type of final result returned by nodes on which training is done.
- * @param <I> Type of data which is fed into each training loop step and returned from it.
- * @param <M> Type of model returned after training.
- * @param <T> Type of input of this trainer.
- * @param <G> Type of distributed context which is needed for forming final result which is send from each node to trainer for final model creation.
- * @param <O> Type of output of postprocessor.
- * @param <X> Type of data which is processed by dataProcessor.
- * @param <Y> Type of data which is returned by postprocessor.
- */
-public abstract class MetaoptimizerGroupTrainer<LC extends HasTrainingUUID, K, V, IN extends Serializable,
- R extends Serializable, I extends Serializable,
- M extends Model, T extends GroupTrainerInput<K>,
- G, O extends Serializable, X, Y> extends
- GroupTrainer<LC, K, V, IN, R, I, M, T, G> {
- /**
- * Metaoptimizer.
- */
- private Metaoptimizer<LC, X, Y, I, IN, O> metaoptimizer;
-
- /**
- * Construct instance of this class.
- *
- * @param cache Cache on which group trainer is done.
- * @param ignite Ignite instance.
- */
- public MetaoptimizerGroupTrainer(Metaoptimizer<LC, X, Y, I, IN, O> metaoptimizer,
- IgniteCache<GroupTrainerCacheKey<K>, V> cache,
- Ignite ignite) {
- super(cache, ignite);
- this.metaoptimizer = metaoptimizer;
- }
-
- /**
- * Get function used to map EntryAndContext to type which is processed by dataProcessor.
- *
- * @return Function used to map EntryAndContext to type which is processed by dataProcessor.
- */
- protected abstract IgniteFunction<EntryAndContext<K, V, G>, X> trainingLoopStepDataExtractor();
-
- /**
- * Get supplier of keys which should be processed by training loop.
- *
- * @param locCtx Local text.
- * @return Supplier of keys which should be processed by training loop.
- */
- protected abstract IgniteSupplier<Stream<GroupTrainerCacheKey<K>>> keysToProcessInTrainingLoop(LC locCtx);
-
- /**
- * Get supplier of context used in training loop step.
- *
- * @param input Input.
- * @param ctx Local context.
- * @return Supplier of context used in training loop step.
- */
- protected abstract IgniteSupplier<G> remoteContextExtractor(I input, LC ctx);
-
- /** {@inheritDoc} */
- @Override protected void init(T data, UUID trainingUUID) {
- }
-
- /**
- * Get function used to process data in training loop step.
- *
- * @return Function used to process data in training loop step.
- */
- protected abstract IgniteFunction<X, ResultAndUpdates<Y>> dataProcessor();
-
- /** {@inheritDoc} */
- @Override protected ComputationsChain<LC, K, V, I, I> trainingLoopStep() {
- ComputationsChain<LC, K, V, I, O> chain = Chains.create(new MetaoptimizerDistributedStep<>(metaoptimizer, this));
- return chain.thenLocally(metaoptimizer::localProcessor);
- }
-
- /** {@inheritDoc} */
- @Override protected I locallyProcessInitData(IN data, LC locCtx) {
- return metaoptimizer.locallyProcessInitData(data, locCtx);
- }
-
- /** {@inheritDoc} */
- @Override protected boolean shouldContinue(I data, LC locCtx) {
- return metaoptimizer.shouldContinue(data, locCtx);
- }
-
- /** {@inheritDoc} */
- @Override protected IgniteFunction<List<IN>, IN> reduceDistributedInitData() {
- return metaoptimizer.initialReducer();
- }
-}
http://git-wip-us.apache.org/repos/asf/ignite/blob/0234ee3b/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/ResultAndUpdates.java
----------------------------------------------------------------------
diff --git a/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/ResultAndUpdates.java b/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/ResultAndUpdates.java
deleted file mode 100644
index 2e3f457..0000000
--- a/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/ResultAndUpdates.java
+++ /dev/null
@@ -1,177 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.ml.trainers.group;
-
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.stream.Collectors;
-import org.apache.ignite.Ignite;
-import org.apache.ignite.IgniteCache;
-import org.apache.ignite.ml.math.functions.IgniteFunction;
-
-/**
- * Class containing result of computation and updates which should be made for caches.
- * Purpose of this class is mainly performance optimization: suppose we have multiple computations which run in parallel
- * and do some updates to caches. It is more efficient to collect all changes from all this computations and perform them
- * in batch.
- *
- * @param <R> Type of computation result.
- */
-public class ResultAndUpdates<R> {
- /**
- * Result of computation.
- */
- private R res;
-
- /**
- * Updates in the form cache name -> (key -> new value).
- */
- private Map<String, Map> updates = new ConcurrentHashMap<>();
-
- /**
- * Construct an instance of this class.
- *
- * @param res Computation result.
- */
- public ResultAndUpdates(R res) {
- this.res = res;
- }
-
- /**
- * Construct an instance of this class.
- *
- * @param res Computation result.
- * @param updates Map of updates in the form cache name -> (key -> new value).
- */
- ResultAndUpdates(R res, Map<String, Map> updates) {
- this.res = res;
- this.updates = updates;
- }
-
- /**
- * Construct an empty result.
- *
- * @param <R> Result type.
- * @return Empty result.
- */
- public static <R> ResultAndUpdates<R> empty() {
- return new ResultAndUpdates<>(null);
- }
-
- /**
- * Construct {@link ResultAndUpdates} object from given result.
- *
- * @param res Result of computation.
- * @param <R> Type of result of computation.
- * @return ResultAndUpdates object.
- */
- public static <R> ResultAndUpdates<R> of(R res) {
- return new ResultAndUpdates<>(res);
- }
-
- /**
- * Add a cache update to this object.
- *
- * @param cache Cache to be updated.
- * @param key Key of cache to be updated.
- * @param val New value.
- * @param <K> Type of key of cache to be updated.
- * @param <V> New value.
- * @return This object.
- */
- @SuppressWarnings("unchecked")
- public <K, V> ResultAndUpdates<R> updateCache(IgniteCache<K, V> cache, K key, V val) {
- String name = cache.getName();
-
- updates.computeIfAbsent(name, s -> new ConcurrentHashMap());
- updates.get(name).put(key, val);
-
- return this;
- }
-
- /**
- * Get result of computation.
- *
- * @return Result of computation.
- */
- public R result() {
- return res;
- }
-
- /**
- * Sum collection of ResultAndUpdate into one: results are reduced by specified binary operator and updates are merged.
- *
- * @param reducer Reducer used to combine computation results.
- * @param resultsAndUpdates ResultAndUpdates to be combined with.
- * @param <R> Type of computation result.
- * @return Sum of collection ResultAndUpdate objects.
- */
- @SuppressWarnings("unchecked")
- static <R> ResultAndUpdates<R> sum(IgniteFunction<List<R>, R> reducer,
- Collection<ResultAndUpdates<R>> resultsAndUpdates) {
- Map<String, Map> allUpdates = new HashMap<>();
-
- for (ResultAndUpdates<R> ru : resultsAndUpdates) {
- for (String cacheName : ru.updates.keySet()) {
- allUpdates.computeIfAbsent(cacheName, s -> new HashMap());
-
- allUpdates.get(cacheName).putAll(ru.updates.get(cacheName));
- }
- }
-
- List<R> results = resultsAndUpdates.stream().map(ResultAndUpdates::result).collect(Collectors.toList());
-
- return new ResultAndUpdates<>(reducer.apply(results), allUpdates);
- }
-
- /**
- * Get updates map.
- *
- * @return Updates map.
- */
- public Map<String, Map> updates() {
- return updates;
- }
-
- /**
- * Set updates map.
- *
- * @param updates New updates map.
- * @return This object.
- */
- ResultAndUpdates<R> setUpdates(Map<String, Map> updates) {
- this.updates = updates;
- return this;
- }
-
- /**
- * Apply updates to caches.
- *
- * @param ignite Ignite instance.
- */
- void applyUpdates(Ignite ignite) {
- for (Map.Entry<String, Map> entry : updates.entrySet()) {
- IgniteCache<Object, Object> cache = ignite.getOrCreateCache(entry.getKey());
-
- cache.putAll(entry.getValue());
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/ignite/blob/0234ee3b/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/chain/Chains.java
----------------------------------------------------------------------
diff --git a/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/chain/Chains.java b/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/chain/Chains.java
deleted file mode 100644
index db4f13f..0000000
--- a/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/chain/Chains.java
+++ /dev/null
@@ -1,56 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.ml.trainers.group.chain;
-
-import java.io.Serializable;
-
-/**
- * Class containing methods creating {@link ComputationsChain}.
- */
-public class Chains {
- /**
- * Create computation chain consisting of one returning its input as output.
- *
- * @param <L> Type of local context of created chain.
- * @param <K> Type of keys of cache used in computation chain.
- * @param <V> Type of values of cache used in computation chain.
- * @param <I> Type of input to computation chain.
- * @return Computation chain consisting of one returning its input as output.
- */
- public static <L extends HasTrainingUUID, K, V, I> ComputationsChain<L, K, V, I, I> create() {
- return (input, context) -> input;
- }
-
- /**
- * Create {@link ComputationsChain} from {@link DistributedEntryProcessingStep}.
- *
- * @param step Distributed chain step.
- * @param <L> Type of local context of created chain.
- * @param <K> Type of keys of cache used in computation chain.
- * @param <V> Type of values of cache used in computation chain.
- * @param <C> Type of context used by worker in {@link DistributedEntryProcessingStep}.
- * @param <I> Type of input to computation chain.
- * @param <O> Type of output of computation chain.
- * @return Computation created from {@link DistributedEntryProcessingStep}.
- */
- public static <L extends HasTrainingUUID, K, V, C, I, O extends Serializable> ComputationsChain<L, K, V, I, O> create(
- DistributedEntryProcessingStep<L, K, V, C, I, O> step) {
- ComputationsChain<L, K, V, I, I> chain = create();
- return chain.thenDistributedForEntries(step);
- }
-}
http://git-wip-us.apache.org/repos/asf/ignite/blob/0234ee3b/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/chain/ComputationsChain.java
----------------------------------------------------------------------
diff --git a/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/chain/ComputationsChain.java b/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/chain/ComputationsChain.java
deleted file mode 100644
index 534b5f9..0000000
--- a/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/chain/ComputationsChain.java
+++ /dev/null
@@ -1,246 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.ml.trainers.group.chain;
-
-import java.io.Serializable;
-import java.util.List;
-import java.util.UUID;
-import java.util.stream.Stream;
-import javax.cache.processor.EntryProcessor;
-import org.apache.ignite.Ignite;
-import org.apache.ignite.cluster.ClusterGroup;
-import org.apache.ignite.lang.IgniteBiPredicate;
-import org.apache.ignite.ml.math.functions.Functions;
-import org.apache.ignite.ml.math.functions.IgniteBiFunction;
-import org.apache.ignite.ml.math.functions.IgniteFunction;
-import org.apache.ignite.ml.math.functions.IgniteSupplier;
-import org.apache.ignite.ml.trainers.group.GroupTrainerCacheKey;
-import org.apache.ignite.ml.trainers.group.GroupTrainerEntriesProcessorTask;
-import org.apache.ignite.ml.trainers.group.GroupTrainerKeysProcessorTask;
-import org.apache.ignite.ml.trainers.group.GroupTrainingContext;
-import org.apache.ignite.ml.trainers.group.ResultAndUpdates;
-
-/**
- * This class encapsulates convenient way for creating computations chain for distributed model training.
- * Chain is meant in the sense that output of each non-final computation is fed as input to next computation.
- * Chain is basically a bi-function from context and input to output, context is separated from input
- * because input is specific to each individual step and context is something which is convenient to have access to in each of steps.
- * Context is separated into two parts: local context and remote context.
- * There are two kinds of computations: local and distributed.
- * Local steps are just functions from two arguments: input and local context.
- * Distributed steps are more sophisticated, but basically can be thought as functions of form
- * localContext -> (function of remote context -> output), locally we fix local context and get function
- * (function of remote context -> output) which is executed distributed.
- * Chains are composable through 'then' method.
- *
- * @param <L> Type of local context.
- * @param <K> Type of cache keys.
- * @param <V> Type of cache values.
- * @param <I> Type of input of this chain.
- * @param <O> Type of output of this chain.
- * // TODO: IGNITE-7350 check if it is possible to integrate with {@link EntryProcessor}.
- */
-@FunctionalInterface
-public interface ComputationsChain<L extends HasTrainingUUID, K, V, I, O> {
- /**
- * Process given input and {@link GroupTrainingContext}.
- *
- * @param input Computation chain input.
- * @param ctx {@link GroupTrainingContext}.
- * @return Result of processing input and context.
- */
- O process(I input, GroupTrainingContext<K, V, L> ctx);
-
- /**
- * Add a local step to this chain.
- *
- * @param locStep Local step.
- * @param <O1> Output of local step.
- * @return Composition of this chain and local step.
- */
- default <O1> ComputationsChain<L, K, V, I, O1> thenLocally(IgniteBiFunction<O, L, O1> locStep) {
- ComputationsChain<L, K, V, O, O1> nextStep = (input, context) -> locStep.apply(input, context.localContext());
- return then(nextStep);
- }
-
- /**
- * Add a distributed step which works in the following way:
- * 1. apply local context and input to local context extractor and keys supplier to get corresponding suppliers;
- * 2. on each node_n
- * 2.1. get context object.
- * 2.2. for each entry_i e located on node_n with key_i from keys stream compute worker((context, entry_i)) and get
- * (cachesUpdates_i, result_i).
- * 2.3. for all i on node_n merge cacheUpdates_i and apply them.
- * 2.4. for all i on node_n, reduce result_i into result_n.
- * 3. get all result_n, reduce them into result and return result.
- *
- * @param <O1> Type of worker output.
- * @param <G> Type of context used by worker.
- * @param workerCtxExtractor Extractor of context for worker.
- * @param worker Function computed on each entry of cache used for training. Second argument is context:
- * common part of data which is independent from key.
- * @param ks Function from chain input and local context to supplier of keys for worker.
- * @param reducer Function used for reducing results of worker.
- * @return Combination of this chain and distributed step specified by given parameters.
- */
- default <O1 extends Serializable, G> ComputationsChain<L, K, V, I, O1> thenDistributedForEntries(
- IgniteBiFunction<O, L, IgniteSupplier<G>> workerCtxExtractor,
- IgniteFunction<EntryAndContext<K, V, G>, ResultAndUpdates<O1>> worker,
- IgniteBiFunction<O, L, IgniteSupplier<Stream<GroupTrainerCacheKey<K>>>> ks,
- IgniteFunction<List<O1>, O1> reducer) {
- ComputationsChain<L, K, V, O, O1> nextStep = (input, context) -> {
- L locCtx = context.localContext();
- IgniteSupplier<Stream<GroupTrainerCacheKey<K>>> keysSupplier = ks.apply(input, locCtx);
-
- Ignite ignite = context.ignite();
- UUID trainingUUID = context.localContext().trainingUUID();
- String cacheName = context.cache().getName();
- ClusterGroup grp = ignite.cluster().forDataNodes(cacheName);
-
- // Apply first two arguments locally because it is common for all nodes.
- IgniteSupplier<G> extractor = Functions.curry(workerCtxExtractor).apply(input).apply(locCtx);
-
- return ignite.compute(grp).execute(new GroupTrainerEntriesProcessorTask<>(trainingUUID, extractor, worker, keysSupplier, reducer, cacheName, ignite), null);
- };
- return then(nextStep);
- }
-
- /**
- * Add a distributed step which works in the following way:
- * 1. apply local context and input to local context extractor and keys supplier to get corresponding suppliers;
- * 2. on each node_n
- * 2.1. get context object.
- * 2.2. for each key_i from keys stream such that key_i located on node_n compute worker((context, entry_i)) and get
- * (cachesUpdates_i, result_i).
- * 2.3. for all i on node_n merge cacheUpdates_i and apply them.
- * 2.4. for all i on node_n, reduce result_i into result_n.
- * 3. get all result_n, reduce them into result and return result.
- *
- * @param <O1> Type of worker output.
- * @param <G> Type of context used by worker.
- * @param workerCtxExtractor Extractor of context for worker.
- * @param worker Function computed on each entry of cache used for training. Second argument is context:
- * common part of data which is independent from key.
- * @param keysSupplier Function from chain input and local context to supplier of keys for worker.
- * @param reducer Function used for reducing results of worker.
- * @return Combination of this chain and distributed step specified by given parameters.
- */
- default <O1 extends Serializable, G> ComputationsChain<L, K, V, I, O1> thenDistributedForKeys(
- IgniteBiFunction<O, L, IgniteSupplier<G>> workerCtxExtractor,
- IgniteFunction<KeyAndContext<K, G>, ResultAndUpdates<O1>> worker,
- IgniteBiFunction<O, L, IgniteSupplier<Stream<GroupTrainerCacheKey<K>>>> keysSupplier,
- IgniteFunction<List<O1>, O1> reducer) {
- ComputationsChain<L, K, V, O, O1> nextStep = (input, context) -> {
- L locCtx = context.localContext();
- IgniteSupplier<Stream<GroupTrainerCacheKey<K>>> ks = keysSupplier.apply(input, locCtx);
-
- Ignite ignite = context.ignite();
- UUID trainingUUID = context.localContext().trainingUUID();
- String cacheName = context.cache().getName();
- ClusterGroup grp = ignite.cluster().forDataNodes(cacheName);
-
- // Apply first argument locally because it is common for all nodes.
- IgniteSupplier<G> extractor = Functions.curry(workerCtxExtractor).apply(input).apply(locCtx);
-
- return ignite.compute(grp).execute(new GroupTrainerKeysProcessorTask<>(trainingUUID, extractor, worker, ks, reducer, cacheName, ignite), null);
- };
- return then(nextStep);
- }
-
- /**
- * Add a distributed step specified by {@link DistributedEntryProcessingStep}.
- *
- * @param step Distributed step.
- * @param <O1> Type of output of distributed step.
- * @param <G> Type of context of distributed step.
- * @return Combination of this chain and distributed step specified by input.
- */
- default <O1 extends Serializable, G> ComputationsChain<L, K, V, I, O1> thenDistributedForEntries(
- DistributedEntryProcessingStep<L, K, V, G, O, O1> step) {
- return thenDistributedForEntries(step::remoteContextSupplier, step.worker(), step::keys, step.reducer());
- }
-
- /**
- * Add a distributed step specified by {@link DistributedKeyProcessingStep}.
- *
- * @param step Distributed step.
- * @param <O1> Type of output of distributed step.
- * @param <G> Type of context of distributed step.
- * @return Combination of this chain and distributed step specified by input.
- */
- default <O1 extends Serializable, G> ComputationsChain<L, K, V, I, O1> thenDistributedForKeys(
- DistributedKeyProcessingStep<L, K, G, O, O1> step) {
- return thenDistributedForKeys(step::remoteContextSupplier, step.worker(), step::keys, step.reducer());
- }
-
- /**
- * Version of 'thenDistributedForKeys' where worker does not depend on context.
- *
- * @param worker Worker.
- * @param kf Function providing supplier
- * @param reducer Function from chain input and local context to supplier of keys for worker.
- * @param <O1> Type of worker output.
- * @return Combination of this chain and distributed step specified by given parameters.
- */
- default <O1 extends Serializable> ComputationsChain<L, K, V, I, O1> thenDistributedForKeys(
- IgniteFunction<GroupTrainerCacheKey<K>, ResultAndUpdates<O1>> worker,
- IgniteBiFunction<O, L, IgniteSupplier<Stream<GroupTrainerCacheKey<K>>>> kf,
- IgniteFunction<List<O1>, O1> reducer) {
-
- return thenDistributedForKeys((o, lc) -> () -> o, (context) -> worker.apply(context.key()), kf, reducer);
- }
-
- /**
- * Combine this computation chain with other computation chain in the following way:
- * 1. perform this calculations chain and get result r.
- * 2. while 'cond(r)' is true, r = otherChain(r, context)
- * 3. return r.
- *
- * @param cond Condition checking if 'while' loop should continue.
- * @param otherChain Chain to be combined with this chain.
- * @return Combination of this chain and otherChain.
- */
- default ComputationsChain<L, K, V, I, O> thenWhile(IgniteBiPredicate<O, L> cond,
- ComputationsChain<L, K, V, O, O> otherChain) {
- ComputationsChain<L, K, V, I, O> me = this;
- return (input, context) -> {
- O res = me.process(input, context);
-
- while (cond.apply(res, context.localContext()))
- res = otherChain.process(res, context);
-
- return res;
- };
- }
-
- /**
- * Combine two this chain to other: feed this chain as input to other, pass same context as second argument to both chains
- * process method.
- *
- * @param next Next chain.
- * @param <O1> Type of next chain output.
- * @return Combined chain.
- */
- default <O1> ComputationsChain<L, K, V, I, O1> then(ComputationsChain<L, K, V, O, O1> next) {
- ComputationsChain<L, K, V, I, O> me = this;
- return (input, context) -> {
- O myRes = me.process(input, context);
- return next.process(myRes, context);
- };
- }
-}
http://git-wip-us.apache.org/repos/asf/ignite/blob/0234ee3b/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/chain/DistributedEntryProcessingStep.java
----------------------------------------------------------------------
diff --git a/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/chain/DistributedEntryProcessingStep.java b/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/chain/DistributedEntryProcessingStep.java
deleted file mode 100644
index 8fd1264..0000000
--- a/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/chain/DistributedEntryProcessingStep.java
+++ /dev/null
@@ -1,34 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.ml.trainers.group.chain;
-
-import java.io.Serializable;
-
-/**
- * {@link DistributedStep} specialized to {@link EntryAndContext}.
- *
- * @param <L> Local context.
- * @param <K> Type of keys of cache used for group training.
- * @param <V> Type of values of cache used for group training.
- * @param <C> Context used by worker.
- * @param <I> Type of input to this step.
- * @param <O> Type of output of this step.
- */
-public interface DistributedEntryProcessingStep<L, K, V, C, I, O extends Serializable> extends
- DistributedStep<EntryAndContext<K, V, C>, L, K, C, I, O> {
-}
http://git-wip-us.apache.org/repos/asf/ignite/blob/0234ee3b/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/chain/DistributedKeyProcessingStep.java
----------------------------------------------------------------------
diff --git a/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/chain/DistributedKeyProcessingStep.java b/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/chain/DistributedKeyProcessingStep.java
deleted file mode 100644
index fb8d867..0000000
--- a/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/chain/DistributedKeyProcessingStep.java
+++ /dev/null
@@ -1,33 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.ml.trainers.group.chain;
-
-import java.io.Serializable;
-
-/**
- * {@link DistributedStep} specialized to {@link KeyAndContext}.
- *
- * @param <L> Local context.
- * @param <K> Type of keys of cache used for group training.
- * @param <C> Context used by worker.
- * @param <I> Type of input to this step.
- * @param <O> Type of output of this step.
- */
-public interface DistributedKeyProcessingStep<L, K, C, I, O extends Serializable> extends
- DistributedStep<KeyAndContext<K, C>, L, K, C, I, O> {
-}
http://git-wip-us.apache.org/repos/asf/ignite/blob/0234ee3b/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/chain/DistributedStep.java
----------------------------------------------------------------------
diff --git a/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/chain/DistributedStep.java b/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/chain/DistributedStep.java
deleted file mode 100644
index 804a886..0000000
--- a/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/chain/DistributedStep.java
+++ /dev/null
@@ -1,70 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.ml.trainers.group.chain;
-
-import java.io.Serializable;
-import java.util.List;
-import java.util.stream.Stream;
-import org.apache.ignite.ml.math.functions.IgniteFunction;
-import org.apache.ignite.ml.math.functions.IgniteSupplier;
-import org.apache.ignite.ml.trainers.group.GroupTrainerCacheKey;
-import org.apache.ignite.ml.trainers.group.ResultAndUpdates;
-
-/**
- * Class encapsulating logic of distributed step in {@link ComputationsChain}.
- *
- * @param <T> Type of elements to be processed by worker.
- * @param <L> Local context.
- * @param <K> Type of keys of cache used for group training.
- * @param <C> Context used by worker.
- * @param <I> Type of input to this step.
- * @param <O> Type of output of this step.
- */
-public interface DistributedStep<T, L, K, C, I, O extends Serializable> {
- /**
- * Create supplier of context used by worker.
- *
- * @param input Input.
- * @param locCtx Local context.
- * @return Context used by worker.
- */
- IgniteSupplier<C> remoteContextSupplier(I input, L locCtx);
-
- /**
- * Get function applied to each cache element specified by keys.
- *
- * @return Function applied to each cache entry specified by keys..
- */
- IgniteFunction<T, ResultAndUpdates<O>> worker();
-
- /**
- * Get supplier of keys for worker.
- *
- * @param input Input to this step.
- * @param locCtx Local context.
- * @return Keys for worker.
- */
- IgniteSupplier<Stream<GroupTrainerCacheKey<K>>> keys(I input, L locCtx);
-
- /**
- * Get function used to reduce results returned by worker.
- *
- * @return Function used to reduce results returned by worker..
- */
- IgniteFunction<List<O>, O> reducer();
-}
http://git-wip-us.apache.org/repos/asf/ignite/blob/0234ee3b/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/chain/EntryAndContext.java
----------------------------------------------------------------------
diff --git a/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/chain/EntryAndContext.java b/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/chain/EntryAndContext.java
deleted file mode 100644
index 59c3b34..0000000
--- a/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/chain/EntryAndContext.java
+++ /dev/null
@@ -1,70 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.ml.trainers.group.chain;
-
-import java.util.Map;
-import org.apache.ignite.ml.trainers.group.GroupTrainerCacheKey;
-
-/**
- * Entry of cache used for group training and context.
- * This class is used as input for workers of distributed steps of {@link ComputationsChain}.
- *
- * @param <K> Type of cache keys used for training.
- * @param <V> Type of cache values used for training.
- * @param <C> Type of context.
- */
-public class EntryAndContext<K, V, C> {
- /**
- * Entry of cache used for training.
- */
- private Map.Entry<GroupTrainerCacheKey<K>, V> entry;
-
- /**
- * Context.
- */
- private C ctx;
-
- /**
- * Construct instance of this class.
- *
- * @param entry Entry of cache used for training.
- * @param ctx Context.
- */
- public EntryAndContext(Map.Entry<GroupTrainerCacheKey<K>, V> entry, C ctx) {
- this.entry = entry;
- this.ctx = ctx;
- }
-
- /**
- * Get entry of cache used for training.
- *
- * @return Entry of cache used for training.
- */
- public Map.Entry<GroupTrainerCacheKey<K>, V> entry() {
- return entry;
- }
-
- /**
- * Get context.
- *
- * @return Context.
- */
- public C context() {
- return ctx;
- }
-}
http://git-wip-us.apache.org/repos/asf/ignite/blob/0234ee3b/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/chain/HasTrainingUUID.java
----------------------------------------------------------------------
diff --git a/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/chain/HasTrainingUUID.java b/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/chain/HasTrainingUUID.java
deleted file mode 100644
index d855adf..0000000
--- a/modules/ml/src/main/java/org/apache/ignite/ml/trainers/group/chain/HasTrainingUUID.java
+++ /dev/null
@@ -1,32 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.ml.trainers.group.chain;
-
-import java.util.UUID;
-
-/**
- * Interface for classes which contain UUID of training.
- */
-public interface HasTrainingUUID {
- /**
- * Get training UUID.
- *
- * @return Training UUID.
- */
- UUID trainingUUID();
-}