Method and system for distributed machine learning
Method, system, and programs for distributed machine learning on a cluster including a plurality of nodes are disclosed. A machine learning process is performed in each of the plurality of nodes based on a respective subset of training data to calculate a local parameter. The training data is partitioned over the plurality of nodes. A plurality of operation nodes are determined from the plurality of nodes based on a status of the machine learning process performed in each of the plurality of nodes. The plurality of operation nodes are connected to form a network topology. An aggregated parameter is generated by merging local parameters calculated in each of the plurality of operation nodes in accordance with the network topology.
1. A method, implemented on at least one machine each of which has at least one processor, storage, and a communication platform connected to a network for distributed machine learning on a cluster including a plurality of nodes, the method comprising:
determining a plurality of operation nodes from the plurality of nodes in the cluster, wherein each of the plurality of operation nodes is selected from a plurality of competing nodes of the cluster based on performance of each of the plurality of competing nodes on an identical subset of training data associated with a machine learning job;
forming a network by connecting the plurality of operation nodes;
distributing the machine learning job to the network, where each of the plurality of operation nodes performs a machine learning sub job on a respective subset of the training data;
computing, in each of the plurality of operation nodes, a local parameter related to the machine learning sub-job; and
generating an aggregated parameter related to the machine learning job based on the computed local parameters.
2. The method of claim 1 , wherein the step of generating an aggregated parameter comprises:
calculating an initial aggregated parameter by merging initial local parameters calculated in each of the plurality of operation nodes;
transmitting the initial aggregated parameter to each of the plurality of operation nodes; and
calculating an updated aggregated parameter by merging updated local parameters calculated in each of the plurality of operation nodes, each updated local parameter being calculated based on the initial aggregated parameter and the subset of the training data in each of the plurality of operation nodes.
3. The method of claim 2 , wherein
a stochastic gradient descent process is performed in each of the plurality of operation nodes for calculating the initial local parameter; and
a batch gradient descent process is performed in each of the plurality of operation nodes for calculating the updated local parameter.
4. The method of claim 1 , further comprising:
selecting each of the plurality of operation nodes from the plurality of competing nodes based on a processing speed of each of the plurality of competing nodes.
5. The method of claim 1 , wherein the determining a plurality of operation nodes further comprises:
dynamically detecting a slow operation node based on a processing speed of each of the plurality of operation nodes;
moving the subset of the training data and the local parameter of the slow operation node to a backup node of the cluster; and
replacing the slow operation node with the backup node in the network.
6. The method of claim 1 , wherein each of the plurality of operation nodes is selected by:
performing a training process on each of the plurality of competing nodes;
determining a competing node that first finishes the training process;
selecting the determined competing node as the operation node; and
stopping the training process on each of the other competing nodes in the plurality of competing nodes.
7. The method of claim 1 , wherein each of the plurality of operation nodes is selected by:
performing a training process on each of the plurality of competing nodes;
inquiring a status of each competing node after performing the training process for a predetermined time period; and
selecting one of the plurality of competing nodes as the operation node based on the inquired status from each competing node.
8. A system for distributed machine learning, the system comprising:
a plurality of operation nodes, wherein each operation node is configured to perform a machine learning sub job on a respective subset of training data associated with a machine learning job to calculate a local parameter, wherein the training data is partitioned over the plurality of operation nodes; and
a coordination node operatively coupled to the plurality of operation nodes, configured to:
determine the plurality of operation nodes from a plurality of nodes, wherein each of the plurality of operation nodes is selected from a plurality of competing nodes based on a performance of each of the plurality of competing nodes on an identical subset of the training data associated with the machine learning job; and
form a network by connecting the plurality of operation nodes,
wherein the plurality of operation nodes are further configured to generate an aggregated parameter related to the machine learning job by merging local parameters calculated in each of the plurality of operation nodes.
9. The system of claim 8 , wherein the plurality of operation nodes are configured to:
calculate an initial aggregated parameter by merging initial local parameters calculated in each of the plurality of operation nodes;
transmit the initial aggregated parameter to each of the plurality of operation nodes; and
calculate an updated aggregated parameter by merging updated local parameters calculated in each of the plurality of operation nodes, each updated local parameter being calculated based on the initial aggregated parameter and the subset of the training data in each of the plurality of operation nodes.
10. The system of claim 9 , wherein
a stochastic gradient descent process is performed in each of the plurality of operation nodes for calculating the initial local parameter; and
a batch gradient descent process is performed in each of the plurality of operation nodes for calculating the updated local parameter.
11. The system of claim 8 ,
wherein each of the plurality of operation nodes is selected from the plurality of competing nodes based on a processing speed of the plurality of competing nodes.
12. The system of claim 8 , wherein the coordination node is further configured to:
dynamically detect a slow operation node based on a processing speed of each of the plurality of operation nodes;
move the subset of the training data and the local parameter of the slow operation node to a backup node of the cluster; and
replace the slow operation node with the backup node in the network.
13. A machine-readable tangible and non-transitory medium having information for distributed machine learning on a cluster including a plurality of nodes recorded thereon, wherein the information, when read by the machine, causes the machine to perform at least the following:
determining a plurality of operation nodes from the plurality of nodes in the cluster, wherein each of the plurality of operation nodes is selected from a plurality of competing nodes of the cluster based on performance of each of the plurality of competing nodes on an identical subset of the training data associated with the machine learning job;
forming a network by connecting the plurality of operation nodes;
distributing the machine learning job to the network, where each of the plurality of operation nodes performs a machine learning sub job on a respective subset of the training data;
computing in each of the plurality of operation nodes, a local parameter related to the machine learning sub-job; and
generating an aggregated parameter related to the machine learning job based on the computed local parameters.
14. The medium of claim 13 , wherein the step of generating an aggregated parameter comprises:
calculating an initial aggregated parameter by merging initial local parameters calculated in each of the plurality of operation nodes;
transmitting the initial aggregated parameter to each of the plurality of operation nodes; and
calculating an updated aggregated parameter by merging updated local parameters calculated in each of the plurality of operation nodes, each updated local parameter being calculated based on the initial aggregated parameter and the subset of the training data in each of the plurality of operation nodes.
15. The medium of claim 14 , wherein
a stochastic gradient descent process is performed in each of the plurality of operation nodes for calculating the initial local parameter; and
a batch gradient descent process is performed in each of the plurality of operation nodes for calculating the updated local parameter.
16. The medium of claim 13 , wherein the information, when read by the machine, causes the machine to further perform the following:
selecting each of the plurality of operation nodes from the plurality of competing nodes based on a processing speed of each of the plurality of competing nodes.
17. The medium of claim 13 , wherein the determining a plurality of operation nodes comprises
dynamically detecting a slow operation node based on a processing speed of each of the plurality of operation nodes;
moving the subset of the training data and the local parameter of the slow operation node to a backup node of the cluster; and
replacing the slow operation node with the backup node in the network.
18. A method, implemented on at least one machine each of which has at least one processor, storage, and a communication platform connected to a network for distributed machine learning on a cluster including a plurality of nodes, the method comprising the steps of:
storing a subset of training data that is partitioned over the plurality of nodes;
performing a stochastic gradient descent process based on the subset of the training data to calculate an initial local parameter;
transmitting the initial local parameter to at least one connected node in accordance with a network topology, wherein the network topology is formed by connecting a plurality of operation nodes and each of the plurality of operation nodes is selected from a plurality of competing nodes based on a processing speed of each of the plurality of competing nodes;
receiving an initial aggregated parameter from the at least one connected node, wherein the initial aggregated parameter is calculated by merging initial local parameters calculated by each of the plurality of nodes in accordance with the network topology;
performing a batch gradient descent process based on the received initial aggregated parameter and the subset of the training data to calculate an updated local parameter; and
transmitting the updated local parameter to the at least one connected node in accordance with the network topology for calculating an updated aggregated parameter.
19. The method of claim 18 , further comprising:
merging the initial local parameter with one or more initial local parameters received from one or more connected nodes in accordance with the network topology; and
merging the updated local parameter with one or more updated local parameters received from the one or more connected nodes in accordance with the network topology.
20. An apparatus comprising:
a storage configured to store a subset of training data that is partitioned over the plurality of nodes;
an AllReducing module configured to:
transmit a local parameter to at least one connected node in accordance with a network topology, wherein the network topology is formed by connecting a plurality of operation nodes and each of the plurality of operation nodes is selected from a plurality of competing nodes based on a processing speed of each of the plurality of competing nodes, and
receive an initial aggregated parameter from the at least one connected node, wherein the initial aggregated parameter is calculated by merging initial local parameters calculated by each of the plurality of nodes in accordance with the network topology; and
a machine learning module configured to:
perform a stochastic gradient descent process based on the subset of the training data to calculate the initial local parameter, and
perform a batch gradient descent process based on the initial aggregated parameter and the subset of the training data to calculate an updated local parameter, wherein the updated local parameter is transmitted to the at least one connected node for calculating an updated aggregated parameter.
21. The apparatus of claim 20 , wherein the AllReducing module is further configured to:
merge the initial local parameter with one or more initial local parameters received from one or more connected nodes in accordance with the network topology; and
merge the updated local parameter with one or more updated local parameters received from the one or more connected nodes in accordance with the network topology.
22. A machine-readable tangible and non-transitory medium having information for distributed machine learning on a cluster including a plurality of nodes recorded thereon, wherein the information, when read by the machine, causes the machine to perform the following:
storing a subset of training data that is partitioned over the plurality of nodes;
performing a stochastic gradient descent process based on the subset of the training data to calculate an initial local parameter;
transmitting the initial local parameter to at least one connected node in accordance with a network topology, wherein the network topology is formed by connecting a plurality of operation nodes and each of the plurality of operation nodes is selected from a plurality of competing nodes based on a processing speed of each of the plurality of competing nodes;
receiving an initial aggregated parameter from the at least one connected node, wherein the initial aggregated parameter is calculated by merging initial local parameters calculated by each of the plurality of nodes in accordance with the network topology;
performing a batch gradient descent process based on the received initial aggregated parameter and the subset of the training data to calculate an updated local parameter; and
transmitting the updated local parameter to the at least one connected node in accordance with the network topology for calculating an updated aggregated parameter.
23. The medium of claim 22 , further comprising:
merging the initial local parameter with one or more initial local parameters received from one or more connected nodes in accordance with the network topology; and
merging the updated local parameter with one or more updated local parameters received from the one or more connected nodes in accordance with the network topology.