shifuml / guagua Goto Github PK

So far the policy is to detect whether one worker is over threshold three times. Guagua will kill worker and make it run in another machine.

In some cases it does not work well in a busy Hadoop cluster. Some times a worker is very slow but never over threshold which cause bad performance.

Consider this policy:
In each iteration, master receives all running time of workers, if the running time is over std, should be better than original policy.

No doc for guagua-site.xml, ComputableMonitor, ...

1. Add GUAGUA_CLASSPATH env.
2. Add guagua-env.sh in conf folder
3. Add GUAGUA_CONF_DIR env
4. ADD GUAGUA_OPTS to support CLI java opts

Should use maven profile to support that.

2014-09-15 20:43:30,734 ERROR [main] ml.shifu.guagua.mapreduce.GuaguaMapper: Error in guagua main run method.
java.lang.NumberFormatException: For input string: "split"
at java.lang.NumberFormatException.forInputString(NumberFormatException.java:65)
at java.lang.Integer.parseInt(Integer.java:492)
at java.lang.Integer.valueOf(Integer.java:582)
at ml.shifu.guagua.worker.AbstractWorkerCoordinator$FailOverCoordinatorCommand$1.compare(AbstractWorkerCoordinator.java:126)
at ml.shifu.guagua.worker.AbstractWorkerCoordinator$FailOverCoordinatorCommand$1.compare(AbstractWorkerCoordinator.java:123)
at java.util.TimSort.countRunAndMakeAscending(TimSort.java:324)
at java.util.TimSort.sort(TimSort.java:189)
at java.util.TimSort.sort(TimSort.java:173)
at java.util.Arrays.sort(Arrays.java:659)
at java.util.Collections.sort(Collections.java:217)
at ml.shifu.guagua.worker.AbstractWorkerCoordinator$FailOverCoordinatorCommand.doExecute(AbstractWorkerCoordinator.java:123)
at ml.shifu.guagua.BasicCoordinator$BasicCoordinatorCommand.execute(BasicCoordinator.java:461)
at ml.shifu.guagua.worker.SyncWorkerCoordinator.preApplication(SyncWorkerCoordinator.java:61)
at ml.shifu.guagua.worker.GuaguaWorkerService.start(GuaguaWorkerService.java:160)
at ml.shifu.guagua.mapreduce.GuaguaMapper.setup(GuaguaMapper.java:99)
at ml.shifu.guagua.mapreduce.GuaguaMapper.run(GuaguaMapper.java:133)
at org.apache.hadoop.mapred.MapTask.runNewMapper(MapTask.java:764)
at org.apache.hadoop.mapred.MapTask.run(MapTask.java:340)
at org.apache.hadoop.mapred.YarnChild$2.run(YarnChild.java:167)
at java.security.AccessController.doPrivileged(Native Method)
at javax.security.auth.Subject.doAs(Subject.java:415)
at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1557)
at org.apache.hadoop.mapred.YarnChild.main(YarnChild.java:162)

In current solution, each iteration, every worker is sucessful, then master can go on next iteration. Although there are two parameters to define the percentage of successful workers in each iteration, but at last iteration, all mapper tasks are done and successful, the job can be successful.

If we have 10 workers, 9 worker at last iteration are done successful, how about just terminate the job with successful state.

This should be easy in YARN, but not easy in MapReduce implemenataion.

Try to implement by using SGD.

From this paper, MapReduce is used, check if we can implement them into guagua,

http://www.csie.ntu.edu.tw/~cjlin/talks/icmr2012.pdf

1. First K points
   In 1st iteration, choose random k points and set to workers
2. Worker Iteration
   Compare each point with K points, find closet one and tag each point. At the same time, For each category, count and sum value and then send them to master
3. Master Computation
   Recompute new K points and send to workers
4. How to save all points with tag
   Add one Worker Interceptor: Save all data to one part hdfs file. Data set to Context#props and used in interceptor#postApplication
5. Think about fail-over (tag???)???

So far each worker each iteration we use a zookeeper znode. Investigate on whether we can use one znode for all iterations.

By using some data Shifu used, set NN setting without hidden layer or just one layer.
For spark, test SparkHDFSLR in 2.2.0 cluster by using the same data.

Compare the performance.

The same as proto-buf

Not like metromone, to be sure all algorithms are right not only for speed but for performance

Real interceptor chain should be cancelled, in each interceptor user may use context.byPass() to stop next execution for all interceptors and even computable. Investigate whether it is doable in guagua interceptor chain.

1. Java doc change
2. guagua shell change
3. add kmeans example

ZooKeeper supports in-memory set-up. This can be used for in memory coordinator. Maybe a choice for embeded zookeeper server.

We store some jars and files into hdfs tmp folder for yarn container usage. Should be cleaned when guagua application is done.

This optimization is for large model.. In master, so far worker results is set as a list, but for large models and more workers, master memory may be a bottleneck. To optimize master, should set worker results not a list but implement as Iteratable to load worker result in memory one by one.

guagua-common: hadoop related common features like InputFormat, guagua-mapreduce and guagua-yarn should depend on it.

guagua-examples: merge guagua-mapreduce-examples and guagua-yarn-examples together.

So far, user can only set guagua parameters by using cmd-line '-Dkey=value'. We'd like to support xml configuration file:
User can configure guagua parameters by setting in guagua-site.xml

Check

1. http://deeplearning4j.org/overview.html
2. https://github.com/jpatanooga/Metronome
3. https://github.com/agibsonccc/java-deeplearning

Looks like we should change a little for deep learning neural network support.

Currently, we load data firstly in a collection, then in each iteration iterating the collection. Should have a good abstraction to make data loading and collection iterating together, the user only needs to write the only function like map in MapReduce to process each record.

Not sure the senario, but it should be a good abstraction.

How to do it?

Follow Spark implementation:
http://spark.apache.org/docs/latest/mllib-linear-methods.html#linear-support-vector-machine-svm

Code:
https://github.com/apache/spark/tree/branch-1.0/mllib/src/main/scala/org/apache/spark

Our input data are located into memory and only one Hadoop job is used. While in Mahout each iteration is a Hadoop job.

Now on worker, gradient computing is one by one, should be run in parallel. Looks like vector abstraction are working on it, check other vector implementation and learn from them.

Or just use multi-threads work on it.

If user doesn't set total iteration number, add a default number.

This is useful to make user set less parameter.

If we can support speculative feature, we can add a backup for each master or worker task to make sure their velocity.

Looks very hard on Yarn, but doable in MapReduce

No Neural Network support in MLlib, try to add such feature to Spark MLLib, contribute to Guagua's competetor.

So far, Bytable results are stored into ZooKeeper znodes, because of size limit of znode(although multiple znodes can be used to store one big Bytable result), another option like HDFS should be supported to support bigger Bytable results.

To use HDFS, read and write are slow, think about an good in-memory solution to both support bigger model and gain good speed.

User don't need to specify zookeeper for some cases. Like Giraph, to start a zookeeper process in master.

A big improvement including embeded zookeeper, time out computable, iteratable worker results instead of list (save memory efficiently in master).

It's time to release 0.5.0.

In current implementation, k is configured, it is not good.

We'd like to: Use the first iteration to choose K centrods.

1. Each worker randomly choose k points and send them to master
2. Master sort all m * k points and choose K discrete points as the first centroids

In Hadoop 2.2.0, even in MRAppMaster there is no fault tolerance, of course GuaguaAppMaster no fault tolerance.

Check Hadoop 2.4.1 to see if there is fault tolerance in MRAppMaster, if yes, learn and make it in GuaguaAppMaster.

To support protobuf as guagua input.

Let user to set maximal time for one iteration of 'compute' function.

User can set time out and after time out null will be returned so master or worker result got in context should do null-checking.

For Big Model, zookeeper is a bottleneck. Sometimes we need big zookeeper heap(10G) and big disk(500G). Consider only using zookeeper for coordination, but transferring results through TCP server/client.

One big issue, fault tolerance doesn't work well. Should consider to persistent master result to HDFS or zookeeper.

Follow Spark MLLib, add L-BFGS, starting from logistic regression.

jar should be first, and then common parameters, then -D parameters

So far, gradients results are stored into one zookeeper znode, if it is over 1MB, other secondary znodes will be used, but a big model like 20M (may need 20 znodes) are still not tested.

Test this case with big data on our hadoop cluster.

So far, splitter are saved in job.xml configuration file. It should be saved into splitter files with format like MapReduce.

Improve readme like Shifu readme doc. This is urgent, to make sure users can get the doc and start from Guagua.