Ich habe versucht, Deep Learning mit Spark × Keras × Docker skalierbar zu machen

Im Februar 2017 veröffentlichte Yahoo! eine Bibliothek, die TensorFlow mit Spark vertreibt. http://yahoohadoop.tumblr.com/post/157196317141/open-sourcing-tensorflowonspark-distributed-deep

Ich persönlich spielte mit Docker, aber was ist mit Keras? Dieses Mal werde ich Dist-Keras, das Keras mit Spark vertreibt, mit Docker ausführen.

Ziel ist es, die Deep-Learning-Bibliothek skalierbar zu nutzen. Durch die Verwendung von GP GPU habe ich das Gefühl, dass sich Deep Learning auf Scale-up konzentriert.

Wenn wir jedoch Scale-up kombinieren, bei dem die Leistung von Computergeräten erforderlich ist, um die Verarbeitung zu beschleunigen, und Scale-out, bei dem die Anzahl der Computergeräte erforderlich ist, sollten wir in der Lage sein, eine schnellere Verarbeitung durchzuführen. Darüber hinaus denke ich persönlich, dass arme Leute, die keine GPU haben, ihr Bestes beim Scale-out geben werden (diese ist dringender (lacht)). Daher habe ich Keras, eine Deep-Learning-Bibliothek, im Docker-Container installiert, mehrere Container gestartet und versucht, die Last durch Scale-Out zu verteilen. Da eine Skalierung mit normalen Keras nicht möglich ist, verwenden wir dieses Mal eine Bibliothek, in der Keras mit Spark ausgeführt wird.

Keras on Spark

Es scheint, dass es zwei Bibliotheken gibt, die Keras mit Spark verteilen.

Der erste ist Elephas, der anscheinend seit etwa einem Jahr existiert. http://maxpumperla.github.io/elephas/

Das andere ist Dist-Keras, das vom CERN (European Nuclear Research Organization!) Herausgegeben wird. http://joerihermans.com/work/distributed-keras/ https://db-blog.web.cern.ch/blog/joeri-hermans/2017-01-distributed-deep-learning-apache-spark-and-keras

CERN ist die ursprüngliche Geschichte von SERN of STEINS; GATE. CERN scheint Rintaro Okabe zu kennen. http://gigazine.net/news/20140623-ama-cern/

Was ist diesmal zu tun?

Dieses Mal werde ich CERNs Dist-Keras mit Docker ausführen und MNIST ausführen. Ich habe mich für Dist-Keras entschieden, weil es kürzlich aktualisiert wurde und funktioniert hat.

Ich habe Elephas auch auf halber Strecke berührt, damit ich bald wieder anfangen kann zu spielen.

Der Grund, warum Dist-Keras unter Docker ausgeführt wird, besteht darin, dass Container einfacher einzurichten sind als virtuelle Maschinen. Es ist in Ordnung, mehrere EC2s zu haben, aber es ist ein Problem, wenn die Kosten hoch sind. Deshalb habe ich 3 Docker-Container in einer virtuellen Maschine gestartet. Einer ist ein Spark-Meister / Arbeiter (Sklave) und der Rest sind Arbeiter. Es wird so konfiguriert.

Dist-Keras CERN ist begeistert von Dist-Keras, lesen Sie also weiter.

https://db-blog.web.cern.ch/blog/joeri-hermans/2017-01-distributed-deep-learning-apache-spark-and-keras

Ich werde einige von ihnen übersetzen.

Distributed Keras Distributed Keras ist ein verteiltes Deep-Learning-Framework, das mit Apache Spark und Keras erstellt wurde. Ziel ist es, die Lernzeit des maschinellen Lernens erheblich zu verbessern und die Analyse von Datensätzen zu ermöglichen, die die Speicherkapazität überschreiten. Das Projekt startete im August 2016 in Zusammenarbeit mit CMS.

Architecture Grundsätzlich wird der Lernprozess von der Lambda-Funktion an den Spark-Worker übergeben. Wenn Sie jedoch mehrere Parameter übergeben, z. B. die Portnummer des Parameterservers, haben wir sie alle in Objekte eingeschlossen und sie in Lernfunktionen umgewandelt, die die für Spark erforderlichen Parameter verwenden können. Um das Ganze aus der Vogelperspektive zu betrachten, erklären wir dies anhand der folgenden Abbildung. Das Lernobjekt startet zuerst den Parameterserver mit dem Spark-Treiber. Starten Sie dann den Arbeitsprozess. Es enthält alle Parameter und Prozesse, die zum Trainieren des Keras-Modells erforderlich sind. Um eine notwendige Anzahl von Arbeitern für die Parallelverarbeitung vorzubereiten, wird der Datensatz durch eine vorbestimmte Kapazität geteilt. Bei der Verarbeitung von Big Data ist es jedoch wünschenswert, die Elemente der Parallelverarbeitung zu erhöhen. Auf diese Weise können Sie eine Situation vermeiden, in der ein Arbeiter untätig ist, während ein anderer weniger fähiger Arbeiter den Stapel nicht fertig gedreht hat (dies wird als Kampfproblem bezeichnet). In diesem Fall gehen Sie zu Spark-Dokumentation. Daher wird empfohlen, das Verteilungselement auf 3 zu setzen.

Es kann jedoch vorkommen, dass Sie die Verwendung größerer verteilter Elemente in Betracht ziehen müssen. Grundsätzlich ist die verteilte Verarbeitung proportional zur Anzahl der Partitionen (Anzahl der Divisionen). Angenommen, Sie weisen einer Aufgabe 20 Mitarbeiter zu und setzen das Verteilungselement auf 3. Spark unterteilt den Datensatz in 60 Shards. Bevor Mitarbeiter mit der Verarbeitung von Partitionen beginnen können, müssen sie zuerst alle Python-Bibliotheken laden, die zur Verarbeitung der Aufgabe erforderlich sind, und dann das Keras-Modell deserialisieren und kompilieren. Dieser Prozess ist ein großer Aufwand. Daher ist diese Technik nur nützlich, wenn große Datenmengen auf nicht heterogenen Systemen verarbeitet werden, die einen langen Aufwärmaufwand erfordern (dh wenn jeder Mitarbeiter eine andere Hardwarelast oder einen anderen Wert hat).

Dist-Keras Instrallation Dist-Keras wird auf Github veröffentlicht. https://github.com/cerndb/dist-keras

Spark und Keras müssen vor der Installation von Dist-Keras installiert werden. Die Installationsmethode hierfür ist wie folgt. https://spark.apache.org/docs/latest/index.html https://keras.io/ja/

Verwenden Sie den folgenden Befehl, um Dist-Keras zu installieren.

## if you need git, run "yum -y install git" beforehand
git clone https://github.com/JoeriHermans/dist-keras
cd dist-keras
pip install -e .

Ich habe es auf der virtuellen Maschine CentOS 7.3 und dem Docker-Container (CentOS 7.3) versucht und konnte es installieren, ohne hängen zu bleiben.

Was diesmal wieder zu tun

Nun, es ist lange her, aber diesmal führen wir Dist-Keras in einem Docker-Container aus und verteilen das Keras-Modelllernen auf die Container. Wir werden die folgende Konfiguration erneut bereitstellen.

Dist-Keras on Docker Dist-Keras hat weder eine Docker-Datei noch ein Docker-Image, also habe ich es selbst gemacht. https://github.com/shibuiwilliam/distkeras-docker

Der Git-Klon lädt die Docker-Datei und die Datei spark_run.sh herunter.

git clone https://github.com/shibuiwilliam/distkeras-docker.git

Wechseln Sie in das Verzeichnis distkeras-docker und erstellen Sie einen Docker.

cd distkeras-docker
docker build -t distkeras .

Es wird einige Zeit dauern, aber warten wir langsam. Die wichtigsten Dinge, die Docker Build macht, sind:

--Installation der notwendigen Werkzeuge --Installation und Konfiguration von Python und Jupyter Notebook --Installation und Konfiguration von Standalone Spark --Installation von Dist keras

SparkMaster & Jupyter Notebook-Startskripte spark_master.sh und spark_slave.sh hinzugefügt

Der Basiscontainer ist CentOS. Spark ist 2.1.0 und Keras ist 2.0.2, beide verwenden die neueste Version.

Verschieben wir es, sobald der Docker-Build abgeschlossen ist. Dieses Mal werden wir 3 Docker-Container starten, um sie zu skalieren.

# docker dist-keras for spark master and slave
docker run -it -p 18080:8080 -p 17077:7077 -p 18888:8888 -p 18081:8081 -p 14040:4040 -p 17001:7001 -p 17002:7002 \
 -p 17003:7003 -p 17004:7004 -p 17005:7005 -p 17006:7006 --name spmaster -h spmaster distkeras /bin/bash

# docker dist-keras for spark slave1
docker run -it --link spmaster:master -p 28080:8080 -p 27077:7077 -p 28888:8888 -p 28081:8081 -p 24040:4040 -p 27001:7001 \
-p 27002:7002 -p 27003:7003 -p 27004:7004 -p 27005:7005 -p 27006:7006 --name spslave1 -h spslave1 distkeras /bin/bash

# docker dist-keras for spark slave2
docker run -it --link spmaster:master -p 38080:8080 -p 37077:7077 -p 38888:8888 -p 38081:8081 -p 34040:4040 -p 37001:7001 \
-p 37002:7002 -p 37003:7003 -p 37004:7004 -p 37005:7005 -p 37006:7006 --name spslave2 -h spslave2 distkeras /bin/bash

Es besteht aus Spark Master / Worker (spmaster), Worker 1 (spslave1) und Worker 2 (spslave2). Führen Sie den Spark-Master und den Worker mit spmaster aus und verbinden Sie spslave1 und spslave2 als Worker (Slave) mit dem Spark-Cluster.

Die Spark-Master- und Worker-Startskripte werden als /opt/spark_master.sh bzw. /opt/spark_slave.sh bereitgestellt. Wenn jeder Container gestartet wird, wird er in das Verzeichnis / opt / verschoben, sodass Sie den Master und den Worker starten können, indem Sie dort Befehle ausführen.

#Auf Spark Master ausführen
#Master und Worker starten, Worker tritt dem Cluster bei
sh spark_master.sh

# Spark spslave1,Führen Sie in 2
#Worker startet und tritt dem Master-Cluster bei
sh spark_slave.sh

Sie haben jetzt einen Master und drei Worker in Ihrem Spark-Cluster. Sie können dies auf der Spark Master-Konsole überprüfen. http://＜spark master＞:18080

In diesem Bereich möchte ich mit Docker Compose oder Swarm mehrere Einheiten gleichzeitig starten, aber dies ist auch meine Hausaufgabe zu einem späteren Zeitpunkt.

MNIST bei Keras on Spark

Versuchen wir MNIST mit Dist-Keras. Der MNIST-Beispielcode wird von Dist-Keras bereitgestellt. Das Verzeichnis lautet / opt / dist-keras / examples und enthält die folgenden Beispieldaten und -programme.

[root@spm examples]# tree
.
|-- cifar-10-preprocessing.ipynb
|-- data
|   |-- atlas_higgs.csv
|   |-- mnist.csv
|   |-- mnist.zip
|   |-- mnist_test.csv
|   `-- mnist_train.csv
|-- example_0_data_preprocessing.ipynb
|-- example_1_analysis.ipynb
|-- kafka_producer.py
|-- kafka_spark_high_throughput_ml_pipeline.ipynb
|-- mnist.ipynb
|-- mnist.py
|-- mnist_analysis.ipynb
|-- mnist_preprocessing.ipynb
|-- spark-warehouse
`-- workflow.ipynb

Dieses Mal wird mnist.py ausgeführt, aber ein Teil des Codes muss je nach Umgebung geändert werden.

Kopieren Sie die Originaldatei, sichern Sie sie und übernehmen Sie die folgenden Änderungen.

cp mnist.py mnist.py.bk

Ändern Sie 1 Import der Spark-Sitzung

Fügen Sie am Anfang Folgendes hinzu.

from pyspark.sql import SparkSession

2 Parametereinstellung ändern

Ändern Sie die Spark-Parameter für diese Umgebung. Die Absicht der Änderung ist wie folgt.

Mit Spark2
Verwendung der lokalen Umgebung
Definieren Sie die Master-URL in der lokalen Umgebung
Die Anzahl der Arbeitnehmer wurde von 1 auf 3 geändert

# Modify these variables according to your needs.
application_name = "Distributed Keras MNIST"
using_spark_2 = True  # False→True
local = True  # False→True
path_train = "data/mnist_train.csv"
path_test = "data/mnist_test.csv"
if local:
    # Tell master to use local resources.
#     master = "local[*]"   comment out
    master = "spark://spm:7077"  # add
    num_processes = 1
    num_executors = 3  # 1→3
else:
    # Tell master to use YARN.
    master = "yarn-client"
    num_executors = 20
    num_processes = 1

Ändern Sie 3 Arbeitsspeicher

Ändern Sie den Arbeitsspeicher von 4G auf 2G. Dies ist keine obligatorische Änderung, da sie einfach auf Ihre Umgebung zugeschnitten ist.

conf = SparkConf()
conf.set("spark.app.name", application_name)
conf.set("spark.master", master)
conf.set("spark.executor.cores", `num_processes`)
conf.set("spark.executor.instances", `num_executors`)
conf.set("spark.executor.memory", "2g") # 4G→2G
conf.set("spark.locality.wait", "0")
conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer");

Sie sind jetzt bereit. Bewegen wir MNIST.

python mnist.py

Sie können den Ausführungsstatus auf der Spark Master-Konsole überprüfen. http://＜spark master＞:18080

Natürlich können Sie auch Details wie Jobs und Executer öffnen.

Jobbildschirm

Ausführungsbildschirm

Die Daten werden in 3 Mitarbeiter geladen und MNIST wird geschult. Das Modell von MNIST ist übrigens wie folgt.

_________________________________________________________________
Layer (type)                 Output Shape              Param #
=================================================================
conv2d_1 (Conv2D)            (None, 26, 26, 32)        320
_________________________________________________________________
activation_1 (Activation)    (None, 26, 26, 32)        0
_________________________________________________________________
conv2d_2 (Conv2D)            (None, 24, 24, 32)        9248
_________________________________________________________________
activation_2 (Activation)    (None, 24, 24, 32)        0
_________________________________________________________________
max_pooling2d_1 (MaxPooling2 (None, 12, 12, 32)        0
_________________________________________________________________
flatten_1 (Flatten)          (None, 4608)              0
_________________________________________________________________
dense_1 (Dense)              (None, 225)               1037025
_________________________________________________________________
activation_3 (Activation)    (None, 225)               0
_________________________________________________________________
dense_2 (Dense)              (None, 10)                2260
_________________________________________________________________
activation_4 (Activation)    (None, 10)                0
=================================================================
Total params: 1,048,853.0
Trainable params: 1,048,853.0
Non-trainable params: 0.0
_________________________________________________________________

Das Programm wird in ca. 30 Minuten abgeschlossen sein.

MNIST-Ergebnisse

Das Ergebnis des Versuchs mit 3 Arbeitern ist wie folgt.

Training time: 1497.86584091
Accuracy: 0.9897
Number of parameter server updates: 3751

Bei einer Lernzeit von weniger als 25 Minuten betrug die Genauigkeit 0,9897. soso.

Das Ergebnis des Versuchs mit einem Arbeiter ist übrigens wie folgt.

Training time: 1572.04011703
Accuracy: 0.9878
Number of parameter server updates: 3751

~~ ・・・ Es ist nur ungefähr 1 Minute für 3 Arbeiter oder 1 Arbeiter (-_-;) ~~

~~ Ist dies die Wahl von Steins Gate ... ~~

Zukunftsaussichten

Dieses Mal habe ich den Docker-Container auf demselben Server gestartet, aber ich denke, dass die ursprüngliche Lastverteilung auf mehrere Server realisiert wird. In Zukunft möchte ich den Dist-Keras-Container auf mehreren Servern und ECS starten, einen Spark-Cluster konfigurieren und die Last verteilen.

[16.04.2017 Nachtrag] Wir haben überprüft, wie die Leistung verbessert werden kann. http://qiita.com/cvusk/items/f54ce15f8c76a396aeb1

[26.05.2017 Nachtrag] Mit Kubernetes geclustert. http://qiita.com/cvusk/items/42a5ffd4e3228963234d

[PYTHON] Ich habe versucht, Deep Learning mit Spark × Keras × Docker skalierbar zu machen