Method and system for distributed deep machine learning

1. A method implemented on at least one computing device each of which has at least one processor, storage, and a communication platform connected to a network for estimating one or more parameters of a machine learning model, the method comprising:

receiving a request for estimating one or more parameters in a vector associated with the machine learning model;

dividing a set of data into a plurality of sub-sets of data, each of which is allocated to corresponding one of a plurality of nodes; and

receiving, from one of the plurality of nodes, estimated values of the one or more parameters, wherein each of the plurality of nodes is configured to

estimate, by a plurality of processing units in the node, values of the one or more parameters based on a sub-set of data allocated to the node, wherein one of the plurality of processing units in the node is preselected as a representative processing unit, wherein the preselected processing unit is a graphical processing unit (GPU),

aggregate, by the representative processing unit of the node, values of the one or more parameters estimated by the plurality of processing units in the node to generate an estimated vector,

divide, by the representative processing unit of the node, the estimated vector into a plurality of portions, each portion of the plurality of portions being designated to one of the plurality of nodes,

collect, by the representative processing unit of the node, estimates of a portion of the vector designated to the node, from representative processing units of all other nodes of the plurality of nodes, to generate a first estimate of the portion of the vector, wherein the first estimate is generated based on the plurality of sub-sets of data,

broadcast, by the representative processing unit of the node, the first estimate of the portion of the vector to the representative processing units of the all other nodes of the plurality of nodes,

receive, by the representative processing unit of the node, from the representative processing units of the all other nodes of the plurality of nodes, second estimates of corresponding portions of the vector designated to the all other nodes, and

broadcast, by the representative processing unit of the node, the first estimate and the received second estimates, to all other processing units of the plurality of processing units included in the node.

2. The method of claim 1 , wherein each of the plurality of nodes performs the following:

obtaining training data to be used to estimate the one or more parameters;

generating updated parameter estimates based on a corresponding sub-set of data;

exchanging the updated parameter estimates with other nodes of the plurality of nodes to generate a state of the one or more parameters that is shared by the plurality of nodes; and

repeating the steps of generating and exchanging until a predetermined condition is met to generate the estimated values of the one or more parameters.

3. The method of claim 1 , further comprising:

detecting a failure at one of the plurality of nodes; and

instructing remaining nodes in the plurality of nodes to continue estimating the one or more parameters.

4. The method of claim 1 , further comprising:

determining a number of the plurality of nodes in accordance with the request; and

determining a location of the set of data for estimating the one or more parameters associated with the machine learning model based on the request.

5. The method of claim 1 , further comprising:

instructing one of the plurality of nodes to store a snapshot of a state of the one or more parameters of the machine learning model in a storage outside the plurality of nodes.

6. The method of claim 5 , further comprising:

detecting a failure with respect to estimating the one or more parameters;

instructing the plurality of nodes to retrieve the snapshot of the state of the one or more parameters of the machine learning model from the storage; and

instructing the plurality of nodes to continue estimating the one or more parameters based on the last state of the one or more parameters.

7. The method claim 1 , wherein the plurality of nodes are synchronized via a Message Passing Interface (MPI) AllReduce based Application Program Interface (API) using a network interface implemented on each of the plurality of nodes, wherein the network interface includes an Ethernet interface, an Infiniband interface, or the Ethernet interface and the Infiniband interface.

8. The method of claim 1 , wherein:

the set of data for estimating the one or more parameters is stored on a Hadoop Distributed File System (HDFS); and

the plurality of nodes are implemented in a Spark framework.

9. The method of claim 1 , further comprising:

selecting a number of processing units in the node for estimating the one or more parameters of the machine learning model based on a workload of each processing unit in the node.

10. A system, having at least one processor, storage, and a communication platform connected to a network for estimating one or more parameters of a machine learning model, the system comprising:

a configuration information identifier implemented by the at least one processor and configured for receiving a request for estimating one or more parameters in a vector associated with the machine learning model;

a training data distributor implemented by the at least one processor and configured for:

dividing a set of data into a plurality of sub-sets of data, each of which corresponds to one of the plurality of nodes; and

a training model determiner implemented by the at least one processor and configured for:

receiving, from one of the plurality of nodes, estimated values of the one or more parameters, wherein each of the plurality of nodes is configured to

estimate, by a plurality of processing units in the node, values of the one or more parameters based on a sub-set of data allocated to the node, wherein one of the plurality of processing units in the node is preselected as a representative processing unit, wherein the preselected processing unit is a graphical processing unit (GPU),

aggregate, by the representative processing unit of the node, values of the one or more parameters estimated by the plurality of processing units in the node to generate an estimated vector,

divide, by the representative processing unit of the node, the estimated vector into a plurality of portions, each portion of the plurality of portions being designated to one of the plurality of nodes,

collect, by the representative processing unit of the node, estimates of a portion of the vector designated to the node, from representative processing units of all other nodes of the plurality of nodes, to generate a first estimate of the portion of the vector, wherein the first estimate is generated based on the plurality of sub-sets of data,

broadcast, by the representative processing unit of the node, the first estimate of the portion of the vector to the representative processing units of the all other nodes of the plurality of nodes,

receive, by the representative processing unit of the node, from the representative processing units of the all other nodes of the plurality of nodes, second estimates of corresponding portions of the vector designated to the all other nodes, and

broadcast, by the representative processing unit of the node, the first estimate and the received second estimates, to all other processing units of the plurality of processing units included in the node.

11. The system of claim 10 , wherein each of the plurality of nodes performs the following:

obtaining training data to be used to estimate the one or more parameters;

generating updated parameter estimates based on a corresponding sub-set of data;

exchanging the updated parameter estimates with other nodes of the plurality of nodes to generate a state of the one or more parameters that is shared by the plurality of nodes; and

repeating the steps of generating and exchanging until a predetermined condition is met to generate the estimated values of the one or more parameters.

12. The system of claim 10 , further comprising:

a node failure detector implemented by the at least one processor and configured for detecting a failure at one of the plurality of nodes; and

a training instruction generator implemented by the at least one processor and configured for instructing the remaining nodes in the plurality of nodes to continue the estimating the one or more parameters.

13. The system of claim 10 , further comprising:

an operation node selector implemented by the at least one processor and configured for determining a number of the plurality of nodes in accordance with the request; and

the training data locator implemented by the at least one processor and configured for determining a location of the set of data for estimating the one or more parameters associated with the machine learning model based on the request.

14. The system of claim 10 , further comprising:

a training instruction generator implemented by the at least one processor and configured for instructing one of the plurality of nodes to store a snapshot of a state of the one or more parameters of the machine learning model in a storage outside the plurality of nodes.

15. The system of claim 14 , further comprising:

a node failure detector implemented by the at least one processor and configured for detecting a failure with respect to estimating the one or more parameters; and

the training instruction generator further configured for:

instructing the plurality of nodes to retrieve the snapshot of the state of the one or more parameters of the machine learning model from the storage, and

instructing the plurality of nodes to continue estimating the one or more parameters based on the last state of the one or more parameters.

16. The system of claim 10 , wherein the plurality of nodes are synchronized via Message Passing Interface (MPI) AllReduce based Application Program Interface (API) using a network interface implemented on each of the plurality of nodes, wherein the network interface includes an Ethernet interface, an Infiniband interface, or the Ethernet interface and the Infiniband interface.

17. A machine-readable tangible and non-transitory medium having information for estimating one or more parameters of a machine learning model, wherein the information, when read by the machine, causes the machine to perform the following:

receiving a request for estimating one or more parameters in a vector associated with the machine learning model;

dividing a set of data into a plurality of sub-sets of data, each of which is allocated to corresponding one of a plurality of nodes; and

receiving, from one of the plurality of nodes, estimated values of the one or more parameters, wherein each of the plurality of nodes is configured to

estimate, by a plurality of processing units in the node, values of the one or more parameters based on a sub-set of data allocated to the node, wherein one of the plurality of processing units in the node is preselected as a representative processing unit, wherein the preselected processing unit is a graphical processing unit (GPU),

aggregate, by the representative processing unit of the node, values of the one or more parameters estimated by the plurality of processing units in the node to generate an estimated vector,

divide, by the representative processing unit of the node, the estimated vector into a plurality of portions, each portion of the plurality of portions being designated to one of the plurality of nodes,

collect, by the representative processing unit of the node, estimates of a portion of the vector designated to the node, from representative processing units of all other nodes of the plurality of nodes, to generate a first estimate of the portion of the vector, wherein the first estimate is generated based on the plurality of sub-sets of data,

broadcast, by the representative processing unit of the node, the first estimate of the portion of the vector to the representative processing units of the all other nodes of the plurality of nodes,

receive, by the representative processing unit of the node, from the representative processing units of the all other nodes of the plurality of nodes, second estimates of corresponding portions of the vector designated to the all other nodes, and

broadcast, by the representative processing unit of the node, the first estimate and the received second estimates, to all other processing units of the plurality of processing units included in the node.

18. The method of claim 1 , wherein the machine learning model includes a cluster, the cluster being a single cluster that enables a deep machine learning to be performed along with a non-deep machine learning and other data processing in the single cluster, with a single program.

19. The method of claim 18 , wherein:

some data processing is performed in the cluster for producing input datasets for the deep machine learning;

the deep machine learning is performed in the cluster for extracting features;

the non-deep machine learning is performed in the cluster to generate a classification model based on the features; and

the other data processing in the cluster includes applying the classification model against a big data set.

20. The method of claim 1 , wherein:

the cluster is an Apache Hadoop cluster on top of Apache Spark; and

the plurality of nodes in the cluster communicate with each other via an Application Program Interface (API) similar to Message Passing Interface (MPI) AllReduce.