[Elephas]Distributed Deep Learning with Keras & Spark Build Status

是 否 +2 论坛币

k人 参与回答

经管之家送您一份

应届毕业生专属福利!

求职就业群

赵安豆老师微信：zhaoandou666

经管之家联合CDA

送您一个全额奖学金名额~ !

立即领取

感谢您参与论坛问题回答

经管之家送您两个论坛币！

+2 论坛币

1. Elephas: Distributed Deep Learning with Keras & Spark Build Status
   
2. 
   
3. Elephas is an extension of Keras, which allows you to run distributed deep learning models at scale with Spark. Elephas currently supports a number of applications, including:
   
4. 
   
5. Data-parallel training of deep learning models
   
6. Distributed hyper-parameter optimization
   
7. Distributed training of ensemble models
   
8. Schematically, elephas works as follows.
   
9. 
   
10. Elephas
    
11. 
    
12. Table of content:
    
13. 
    
14. Elephas: Distributed Deep Learning with Keras & Spark
    
15. Introduction
    
16. Getting started
    
17. Installation
    
18. Basic example
    
19. Spark ML example
    
20. Usage of data-parallel models
    
21. Model updates (optimizers)
    
22. Update frequency
    
23. Update mode
    
24. Asynchronous updates with read and write locks (mode='asynchronous')
    
25. Asynchronous updates without locks (mode='hogwild')
    
26. Synchronous updates (mode='synchronous')
    
27. Degree of parallelization (number of workers)
    
28. Distributed hyper-parameter optimization
    
29. Distributed training of ensemble models
    
30. Discussion
    
31. Future work & contributions
    
32. Literature
    
33. Introduction
    
34. 
    
35. Elephas brings deep learning with Keras to Spark. Elephas intends to keep the simplicity and high usability of Keras, thereby allowing for fast prototyping of distributed models, which can be run on massive data sets. For an introductory example, see the following iPython notebook.
    
36. 
    
37. ἐλέφας is Greek for ivory and an accompanying project to κέρας, meaning horn. If this seems weird mentioning, like a bad dream, you should confirm it actually is at the Keras documentation. Elephas also means elephant, as in stuffed yellow elephant.
    
38. 
    
39. Elephas implements a class of data-parallel algorithms on top of Keras, using Spark's RDDs and data frames. Keras Models are initialized on the driver, then serialized and shipped to workers, alongside with data and broadcasted model parameters. Spark workers deserialize the model, train their chunk of data and send their gradients back to the driver. The "master" model on the driver is updated by an optimizer, which takes gradients either synchronously or asynchronously.
    
40. 
    
41. Getting started
    
42. 
    
43. Installation
    
44. 
    
45. Install elephas from PyPI with
    
46. 
    
47. pip install elephas
    
48. Depending on what OS you are using, you may need to install some prerequisite modules (LAPACK, BLAS, fortran compiler) first.
    
49. 
    
50. For example, on Ubuntu Linux:
    
51. 
    
52. sudo apt-get install liblapack-dev libblas-dev gfortran
    
53. A quick way to install Spark locally is to use homebrew on Mac
    
54. 
    
55. brew install spark
    
56. or linuxbrew on linux.
    
57. 
    
58. brew install apache-spark
    
59. The brew version of Spark may be outdated at times. To build from source, simply follow the instructions at the Spark download section or use the following commands.
    
60. 
    
61. wget http://apache.mirrors.tds.net/spark/spark-1.5.2/spark-1.5.2-bin-hadoop2.6.tgz -P ~
    
62. sudo tar zxvf ~/spark-* -C /usr/local
    
63. sudo mv /usr/local/spark-* /usr/local/spark
    
64. After that, make sure to put these path variables to your shell profile (e.g. ~/.zshrc):
    
65. 
    
66. export SPARK_HOME=/usr/local/spark
    
67. export PATH=$PATH:$SPARK_HOME/bin
    
68. Using Docker
    
69. 
    
70. Install and get Docker running by following the instructions here (https://www.docker.com/).
    
71. 
    
72. Building
    
73. 
    
74. The build takes quite a while to run the first time since many packages need to be downloaded and installed. In the same directory as the Dockerfile run the following commands
    
75. 
    
76. docker build . -t pyspark/elephas
    
77. Running
    
78. 
    
79. The following command starts a container with the Notebook server listening for HTTP connections on port 8899 (since local Jupyter notebooks use 8888) without authentication configured.
    
80. 
    
81. docker run -d -p 8899:8888 pyspark/elephas

复制代码

本帖隐藏的内容

https://github.com/maxpumperla/elephas

扫码加我 拉你入群

请注明：姓名-公司-职位

以便审核进群资格，未注明则拒绝

分享0 收藏0 回帖

关键词：distributed Learning earning tribute status

1. Basic example
   
2. 
   
3. After installing both Elephas and Spark, training a model is done schematically as follows:
   
4. 
   
5. ##Create a local pyspark context
   
6. from pyspark import SparkContext, SparkConf
   
7. conf = SparkConf().setAppName('Elephas_App').setMaster('local[8]')
   
8. sc = SparkContext(conf=conf)
   
9. 
   
10. ##Define and compile a Keras model
    
11. from keras.models import Sequential
    
12. from keras.layers.core import Dense, Dropout, Activation
    
13. from keras.optimizers import SGD
    
14. model = Sequential()
    
15. model.add(Dense(128, input_dim=784))
    
16. model.add(Activation('relu'))
    
17. model.add(Dropout(0.2))
    
18. model.add(Dense(128))
    
19. model.add(Activation('relu'))
    
20. model.add(Dropout(0.2))
    
21. model.add(Dense(10))
    
22. model.add(Activation('softmax'))
    
23. model.compile(loss='categorical_crossentropy', optimizer=SGD())
    
24. 
    
25. ##Create an RDD from numpy arrays
    
26. from elephas.utils.rdd_utils import to_simple_rdd
    
27. rdd = to_simple_rdd(sc, X_train, Y_train)
    
28. A SparkModel is defined by passing Spark context and Keras model. Additionally, one has choose an optimizer used for updating the elephas model, an update frequency, a parallelization mode and the degree of parallelism, i.e. the number of workers.
    
29. from elephas.spark_model import SparkModel
    
30. from elephas import optimizers as elephas_optimizers
    
31. 
    
32. adagrad = elephas_optimizers.Adagrad()
    
33. spark_model = SparkModel(sc,model, optimizer=adagrad, frequency='epoch', mode='asynchronous', num_workers=2)
    
34. spark_model.train(rdd, nb_epoch=20, batch_size=32, verbose=0, validation_split=0.1, num_workers=8)
    
35. 
    
36. ##Run your script using spark-submit
    
37. spark-submit --driver-memory 1G ./your_script.py
    
38. Increasing the driver memory even further may be necessary, as the set of parameters in a network may be very large and collecting them on the driver eats up a lot of resources. See the examples folder for a few working examples.

复制代码

谢谢楼主分享！

1. Spark MLlib example
   
2. 
   
3. Following up on the last example, to create an RDD of LabeledPoints for supervised training from pairs of numpy arrays, use
   
4. 
   
5. from elephas.utils.rdd_utils import to_labeled_point
   
6. lp_rdd = to_labeled_point(sc, X_train, Y_train, categorical=True)
   
7. Training a given LabeledPoint-RDD is very similar to what we've seen already
   
8. 
   
9. from elephas.spark_model import SparkMLlibModel
   
10. adadelta = elephas_optimizers.Adadelta()
    
11. spark_model = SparkMLlibModel(sc,model, optimizer=adadelta, frequency='batch', mode='hogwild', num_workers=2)
    
12. spark_model.train(lp_rdd, nb_epoch=20, batch_size=32, verbose=0, validation_split=0.1, categorical=True, nb_classes=nb_classes)

复制代码

1. Spark ML example
   
2. 
   
3. To train a model with a SparkML estimator on a data frame, use the following syntax.
   
4. 
   
5. df = to_data_frame(sc, X_train, Y_train, categorical=True)
   
6. test_df = to_data_frame(sc, X_test, Y_test, categorical=True)
   
7. 
   
8. adadelta = elephas_optimizers.Adadelta()
   
9. estimator = ElephasEstimator(sc,model,
   
10.         nb_epoch=nb_epoch, batch_size=batch_size, optimizer=adadelta, frequency='batch', mode='asynchronous', num_workers=2,
    
11.         verbose=0, validation_split=0.1, categorical=True, nb_classes=nb_classes)
    
12. 
    
13. fitted_model = estimator.fit(df)

复制代码

1. from __future__ import print_function
   
2. from hyperopt import Trials, STATUS_OK, tpe
   
3. 
   
4. from hyperas import optim
   
5. from hyperas.distributions import choice, uniform
   
6. 
   
7. from elephas.hyperparam import HyperParamModel
   
8. 
   
9. from pyspark import SparkContext, SparkConf
   
10. 
    
11. def data():
    
12.     '''
    
13.     Data providing function:
    
14. 
    
15.     Make sure to have every relevant import statement included here and return data as
    
16.     used in model function below. This function is separated from model() so that hyperopt
    
17.     won't reload data for each evaluation run.
    
18.     '''
    
19.     from keras.datasets import mnist
    
20.     from keras.utils import np_utils
    
21.     (X_train, y_train), (X_test, y_test) = mnist.load_data()
    
22.     X_train = X_train.reshape(60000, 784)
    
23.     X_test = X_test.reshape(10000, 784)
    
24.     X_train = X_train.astype('float32')
    
25.     X_test = X_test.astype('float32')
    
26.     X_train /= 255
    
27.     X_test /= 255
    
28.     nb_classes = 10
    
29.     Y_train = np_utils.to_categorical(y_train, nb_classes)
    
30.     Y_test = np_utils.to_categorical(y_test, nb_classes)
    
31.     return X_train, Y_train, X_test, Y_test
    
32. 
    
33. 
    
34. def model(X_train, Y_train, X_test, Y_test):
    
35.     '''
    
36.     Model providing function:
    
37. 
    
38.     Create Keras model with double curly brackets dropped-in as needed.
    
39.     Return value has to be a valid python dictionary with two customary keys:
    
40.         - loss: Specify a numeric evaluation metric to be minimized
    
41.         - status: Just use STATUS_OK and see hyperopt documentation if not feasible
    
42.     The last one is optional, though recommended, namely:
    
43.         - model: specify the model just created so that we can later use it again.
    
44.     '''
    
45.     from keras.models import Sequential
    
46.     from keras.layers.core import Dense, Dropout, Activation
    
47.     from keras.optimizers import RMSprop
    
48. 
    
49.     model = Sequential()
    
50.     model.add(Dense(512, input_shape=(784,)))
    
51.     model.add(Activation('relu'))
    
52.     model.add(Dropout({{uniform(0, 1)}}))
    
53.     model.add(Dense({{choice([256, 512, 1024])}}))
    
54.     model.add(Activation('relu'))
    
55.     model.add(Dropout({{uniform(0, 1)}}))
    
56.     model.add(Dense(10))
    
57.     model.add(Activation('softmax'))
    
58. 
    
59.     rms = RMSprop()
    
60.     model.compile(loss='categorical_crossentropy', optimizer=rms)
    
61. 
    
62.     model.fit(X_train, Y_train,
    
63.               batch_size={{choice([64, 128])}},
    
64.               nb_epoch=1,
    
65.               show_accuracy=True,
    
66.               verbose=2,
    
67.               validation_data=(X_test, Y_test))
    
68.     score, acc = model.evaluate(X_test, Y_test, show_accuracy=True, verbose=0)
    
69.     print('Test accuracy:', acc)
    
70.     return {'loss': -acc, 'status': STATUS_OK, 'model': model.to_yaml(), 'weights': pickle.dumps(model.get_weights())}
    
71. 
    
72. # Create Spark context
    
73. conf = SparkConf().setAppName('Elephas_Hyperparameter_Optimization').setMaster('local[8]')
    
74. sc = SparkContext(conf=conf)
    
75. 
    
76. # Define hyper-parameter model and run optimization.
    
77. hyperparam_model = HyperParamModel(sc)
    
78. hyperparam_model.minimize(model=model, data=data, max_evals=5)

复制代码

1. from __future__ import absolute_import
   
2. from __future__ import print_function
   
3. 
   
4. from keras.datasets import mnist
   
5. from keras.models import Sequential
   
6. from keras.layers.core import Dense, Dropout, Activation
   
7. from keras.optimizers import Adam
   
8. from keras.utils import np_utils
   
9. 
   
10. from elephas.ml_model import ElephasEstimator
    
11. from elephas.ml.adapter import to_data_frame
    
12. from elephas import optimizers as elephas_optimizers
    
13. 
    
14. from pyspark import SparkContext, SparkConf
    
15. from pyspark.mllib.evaluation import MulticlassMetrics
    
16. from pyspark.ml import Pipeline
    
17. 
    
18. 
    
19. # Define basic parameters
    
20. batch_size = 64
    
21. nb_classes = 10
    
22. nb_epoch = 1
    
23. 
    
24. # Load data
    
25. (x_train, y_train), (x_test, y_test) = mnist.load_data()
    
26. 
    
27. x_train = x_train.reshape(60000, 784)
    
28. x_test = x_test.reshape(10000, 784)
    
29. x_train = x_train.astype("float32")
    
30. x_test = x_test.astype("float32")
    
31. x_train /= 255
    
32. x_test /= 255
    
33. print(x_train.shape[0], 'train samples')
    
34. print(x_test.shape[0], 'test samples')
    
35. 
    
36. # Convert class vectors to binary class matrices
    
37. y_train = np_utils.to_categorical(y_train, nb_classes)
    
38. y_test = np_utils.to_categorical(y_test, nb_classes)
    
39. 
    
40. model = Sequential()
    
41. model.add(Dense(128, input_dim=784))
    
42. model.add(Activation('relu'))
    
43. model.add(Dropout(0.2))
    
44. model.add(Dense(128))
    
45. model.add(Activation('relu'))
    
46. model.add(Dropout(0.2))
    
47. model.add(Dense(10))
    
48. model.add(Activation('softmax'))
    
49. 
    
50. 
    
51. # Compile model
    
52. adam = Adam()
    
53. model.compile(loss='categorical_crossentropy', optimizer=adam)
    
54. 
    
55. # Create Spark context
    
56. conf = SparkConf().setAppName('Mnist_Spark_MLP').setMaster('local[8]')
    
57. sc = SparkContext(conf=conf)
    
58. 
    
59. # Build RDD from numpy features and labels
    
60. df = to_data_frame(sc, x_train, y_train, categorical=True)
    
61. test_df = to_data_frame(sc, x_test, y_test, categorical=True)
    
62. 
    
63. # Define elephas optimizer
    
64. adadelta = elephas_optimizers.Adadelta()
    
65. 
    
66. # Initialize Spark ML Estimator
    
67. estimator = ElephasEstimator()
    
68. estimator.set_keras_model_config(model.to_yaml())
    
69. estimator.set_optimizer_config(adadelta.get_config())
    
70. estimator.set_nb_epoch(nb_epoch)
    
71. estimator.set_batch_size(batch_size)
    
72. estimator.set_num_workers(1)
    
73. estimator.set_verbosity(0)
    
74. estimator.set_validation_split(0.1)
    
75. estimator.set_categorical_labels(True)
    
76. estimator.set_nb_classes(nb_classes)
    
77. 
    
78. # Fitting a model returns a Transformer
    
79. pipeline = Pipeline(stages=[estimator])
    
80. fitted_pipeline = pipeline.fit(df)
    
81. 
    
82. # Evaluate Spark model by evaluating the underlying model
    
83. prediction = fitted_pipeline.transform(test_df)
    
84. pnl = prediction.select("label", "prediction")
    
85. pnl.show(100)
    
86. 
    
87. prediction_and_label = pnl.map(lambda row: (row.label, row.prediction))
    
88. metrics = MulticlassMetrics(prediction_and_label)
    
89. print(metrics.precision())
    
90. print(metrics.recall())

复制代码