[PYTHON] Erste Schritte mit Spark

Was ist Spark?

Blitzschnelles Cluster-Computing. Eine Bibliothek, die die Stapelverarbeitung in großem Maßstab verteilt. Die verteilte Verarbeitung ist gut. Sie können SQL verwenden. Streaming-Daten können verwendet werden. Kann für maschinelles Lernen verwendet werden. Die Graphentheorie kann verwendet werden. Deep Learning kann platziert werden. Diese nutzen den Speicher voll aus und verteilen den Cluster mit hoher Geschwindigkeit.

Versuchte Umgebung

• mac
• python2.7.12
• spark-1.6.2-bin-hadoop2.6

Spark-Installation

JDK-Installation

Ubuntu

sudo apt-get install -y openjdk-8-jdk

Mac

brew cask install java

Maven Installation

Ubuntu

sudo apt install maven

mac

brew install maven

Spark-Installation

Sei / usr / local / spark SPARK_HOME. Wählen Sie eine beliebige Version. http://ftp.riken.jp/net/apache/spark/

Ubuntu

wget http://ftp.riken.jp/net/apache/spark/spark-1.6.2/spark-1.6.2-bin-hadoop2.6.tgz
$ tar zxvf spark-1.6.2-bin-hadoop2.6.tgz
$ sudo mv spark-1.6.2-bin-hadoop2.6 /usr/local/
$ sudo ln -s /usr/local/spark-1.6.2-bin-hadoop2.6 /usr/local/spark

Fügen Sie Folgendes zu .bashrc hinzu

Ubuntu

export SPARK_HOME=/usr/local/spark
export PATH=$PATH:$SPARK_HOME/bin

Mac

brew install apache-spark

Spark-Shell-Ausführung

python

$ spark-shell --master local[*]
(Unterlassung)
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_\   version 1.6.1
      /_/

Using Scala version 2.10.5 (OpenJDK 64-Bit Server VM, Java 1.8.0_91)
Type in expressions to have them evaluated.
Type :help for more information.
(Unterlassung)

scala> val textFile = sc.textFile("/usr/local/spark/README.md")
textFile: org.apache.spark.rdd.RDD[String] = /usr/local/spark/README.md MapPartitionsRDD[1] at textFile at <console>:27

scala> val wordCounts = textFile.flatMap(line => line.split(" ")).map(word => (word, 1)).reduceByKey((a, b) => a + b)
wordCounts: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[4] at reduceByKey at <console>:29

scala> wordCounts.collect()
res0: Array[(String, Int)] = Array((package,1), (For,2), (Programs,1), (processing.,1), ...(Unterlassung)...,  (>>>,1), (programming,1), (T...
scala>

Überprüfen Sie, ob es auf der Konsole funktioniert. Bei Verwendung mit Python

python

./bin/pyspark

Wenn Sie pyspark mit jupyter ausführen möchten

Fügen Sie Folgendes zu .bashrc hinzu

python

#spark                                                                                                                                                        
export SPARK_HOME=/usr/local/spark/spark-1.6.2-bin-hadoop2.6
export PATH=$PATH:$SPARK_HOME/bin
#jupyter spark
export PYSPARK_PYTHON=$PYENV_ROOT/shims/python #Passen Sie den Pfad an die Umgebung an
export PYSPARK_DRIVER_PYTHON=$PYENV_ROOT/shims/jupyter
export PYSPARK_DRIVER_PYTHON_OPTS="notebook"

python

source .bashrc
pyspark

Durch Ausführen des Befehls pyspark wird jupyter gestartet. Wenn Sie einen Fehler erhalten, der die RDD von spark nicht erfasst, können Sie ihn beheben, indem Sie den Kernel neu starten.

Verteilter Datensatz (RDD)

Parallelisierte Sammlung

Parallele Ausführung wird möglich.

Scala

val data = Array(1, 2, 3, 4, 5)
val distData = sc.parallelize(data)

Java

List<Integer> data = Arrays.asList(1, 2, 3, 4, 5);
JavaRDD<Integer> distData = sc.parallelize(data);

Python

data = [1, 2, 3, 4, 5]
distData = sc.parallelize(data)

Externer Datensatz

Scala

val distFile = sc.textFile("data.txt")
distFile: org.apache.spark.rdd.RDD[String] = data.txt

Java

JavaRDD<String> distFile = sc.textFile("data.txt");

Python

distFile = sc.textFile("data.txt")

RDD-Betrieb

Basic

Holen Sie sich die Daten mit textFile und legen Sie sie auf rdd Mit Karte konvertieren Mit reduzieren aggregieren

Scala

val lines = sc.textFile("data.txt")
val lineLengths = lines.map(s => s.length)
val totalLength = lineLengths.reduce((a, b) => a + b)

Java

JavaRDD<String> lines = sc.textFile("data.txt");
JavaRDD<Integer> lineLengths = lines.map(s -> s.length());
int totalLength = lineLengths.reduce((a, b) -> a + b);

Python

lines = sc.textFile("data.txt")
lineLengths = lines.map(lambda s: len(s))
totalLength = lineLengths.reduce(lambda a, b: a + b)

Übergabe einer Funkenfunktion

Scala

object MyFunctions {
  def func1(s: String): String = { ... }
}

myRdd.map(MyFunctions.func1)

Java

JavaRDD<String> lines = sc.textFile("data.txt");
JavaRDD<Integer> lineLengths = lines.map(new Function<String, Integer>() {
  public Integer call(String s) { return s.length(); }
});
int totalLength = lineLengths.reduce(new Function2<Integer, Integer, Integer>() {
  public Integer call(Integer a, Integer b) { return a + b; }
});

Python

"""MyScript.py"""
if __name__ == "__main__":
    def myFunc(s):
        words = s.split(" ")
        return len(words)

    sc = SparkContext(...)
    sc.textFile("file.txt").map(myFunc)

Verschluss verstehen

Beispiel

Scala

var counter = 0
var rdd = sc.parallelize(data)

// Wrong: Don't do this!!
rdd.foreach(x => counter += x)

println("Counter value: " + counter)

Java

int counter = 0;
JavaRDD<Integer> rdd = sc.parallelize(data);

// Wrong: Don't do this!!
rdd.foreach(x -> counter += x);

println("Counter value: " + counter);

Python

counter = 0
rdd = sc.parallelize(data)

# Wrong: Don't do this!!
def increment_counter(x):
    global counter
    counter += x
rdd.foreach(increment_counter)

print("Counter value: ", counter)

Manipulieren von Schlüssel-Wert-Paaren

Scala

val lines = sc.textFile("data.txt")
val pairs = lines.map(s => (s, 1))
val counts = pairs.reduceByKey((a, b) => a + b)

Java

JavaRDD<String> lines = sc.textFile("data.txt");
JavaPairRDD<String, Integer> pairs = lines.mapToPair(s -> new Tuple2(s, 1));
JavaPairRDD<String, Integer> counts = pairs.reduceByKey((a, b) -> a + b);

Python

lines = sc.textFile("data.txt")
pairs = lines.map(lambda s: (s, 1))
counts = pairs.reduceByKey(lambda a, b: a + b)

Umwandlung

Aktion

Verschieben wir es zuerst (Python Edition)

Ich gabelte mich, weil es sehr leicht zu verstehen war, was bei Wettbewerben in Übersee verwendet wurde. Da es sich um einen Jupyter handelt, führen Sie ihn nur in der Reihenfolge von oben aus.

Klicken Sie hier für die Quelle https://github.com/miyamotok0105/spark-py-notebooks

Inhaltsverzeichnis

RDD erstellen

Informationen zum Lesen und Parallelisieren von Dateien

RDD-Grundlagen

Über Karte, Filter, Sammeln

RDD-Abtastung

Erläutert die RDD-Stichprobenmethode.

RDD-Set-Betrieb

Eine kurze Einführung in einige RDD-Pseudo-Set-Operationen.

Datenaggregation auf RDD

Über RDD-Aktionen reduzieren, falten, aggregieren.

Bedienung des Schlüssel / Wert-Paares RDD

Umgang mit Schlüssel / Wert-Paaren zum Aggregieren und Erkunden von Daten.

MLlib: Grundlegende Statistiken und explorative Datenanalyse

Ein Notizbuch, das die grundlegenden Statistiken von MLlib für lokale Vektortypen, explorative Datenanalyse und Modellauswahl enthält.

MLlib: Logistische Rendite

Klassifizierung markierter Punkte und logistischer Regressionen für Netzwerkangriffe in MLlib. Anwendung der Modellauswahlmethode unter Verwendung der Korrelationsmatrix und des Hypothesentests.

MLlib: Entscheidungsbaum

Eine Methode, die die Verwendung baumbasierter Methoden und die Auswahl von Modellen und Features erklärt.

Spark SQL: Strukturierte Verarbeitung zur Datenanalyse

Dieses Notizbuch leitet das Schema für einen Datensatz von Netzwerkinteraktionen ab. Auf dieser Grundlage verwenden wir die SQL DataFrame-Abstraktion von Spark, um eine strukturiertere explorative Datenanalyse durchzuführen.

Clustering mit MLlib (K Means)

Clustering-Prozess von Irisdaten.

Hauptinhalt

RDD erstellen

data_file = "./kddcup.data_10_percent.gz"
#Allgemeine Schöpfung
raw_data = sc.textFile(data_file) 
#Parallel erstellen
raw_data = sc.parallelize(data_file) 

RDD-Grundlagen

#Filterkonvertierung
normal_raw_data = raw_data.filter(lambda x: 'normal.' in x)
#Kartenkonvertierung
csv_data = raw_data.map(lambda x: x.split(","))

RDD-Abtastung

Gibt ein Array zurück, das Zufallsstichproben der num-Elemente des Datasets enthält, die mit einem Zufallsgenerator-Startwert vorgegeben sind.

raw_data_sample = raw_data.takeSample(False, 400000, 1234)

Stellen Sie den Betrieb von RDD ein

normal_raw_data = raw_data.filter(lambda x: "normal." in x)
#Subtrahieren
attack_raw_data = raw_data.subtract(normal_raw_data)

#Kartesisches Produkt (direktes Produktset)
product = protocols.cartesian(services).collect()
print "There are {} combinations of protocol X service".format(len(product))

Empfehlung durch Mlib-Co-Filterung

Was ist Co-Filterung?

Es handelt sich um eine Produktempfehlung für Kunden, die eine Matrix aus Benutzern und Artikeln verwendet. Aus dieser Matrix kann gesagt werden, dass es sich um einen Mechanismus handelt, um die Korrelation von Benutzern zu analysieren und Empfehlungen abzugeben, die auf der Annahme beruhen, dass ähnliche Benutzer die Produkte kaufen, die sie kaufen. Referenz

Inhaltsbasiert und Co-Filterung

Kooperative Filterung

• Empfehlung basierend auf Benutzerverhalten

Inhaltsbasierte (inhaltsbasierte) Filterung

• Ähnlichkeit sortiert nach Artikelmerkmalsvektor und empfohlen

Details

Vor- und Nachteile von inhaltsbasiertem und Co-Filtern

Laden des Empfehlungsmoduls

from pyspark.mllib.recommendation import ALS, MatrixFactorizationModel, Rating

Empfehlungen aussprechen

model = ALS.train(ratings, rank, numIterations)

Prognose

predictions_all = model.predictAll(sc.parallelize(f_XY)).map(lambda r: ((r[0], r[1]), limitter(r[2]) ))

Praktische Datenanalyse mit dem Spark-Machine Learning Casebook für umfangreiche Daten

https://www.oreilly.co.jp/books/9784873117508/

Quelle herunterladen. Ganz Scala. Und dieses Buch hat eine ziemlich starke Scala-Farbe. Ich schreibe unter der Annahme, dass ich Spark kenne.

https://github.com/sryza/aas.git
git checkout 1st-edition

Kapitel 2 Musikempfehlungen und Audioscrobbler-Datensatz

Daten bekommen

wget http://www.iro.umontreal.ca/~lisa/datasets/profiledata_06-May-2005.tar.gz
tar xzvf profiledata_06-May-2005.tar.gz 
Ergebnis
profiledata_06-May-2005/
profiledata_06-May-2005/artist_data.txt
profiledata_06-May-2005/README.txt
profiledata_06-May-2005/user_artist_data.txt
profiledata_06-May-2005/artist_alias.txt

Quelle https://github.com/sryza/aas/blob/1st-edition/ch03-recommender/src/main/scala/com/cloudera/datascience/recommender/RunRecommender.scala

Plus α

Ein Beispiel für ein Deep-Learning-Framework, das auf Spark basiert

BigDL(torch base) https://github.com/intel-analytics/BigDL TensorFlow https://github.com/yahoo/TensorFlowOnSpark keras https://github.com/maxpumperla/elephas

Wenn Sie Jupyter aus der Ferne anrufen möchten

http://qiita.com/joemphilips/items/de5d12723b9b88b5b090

Sicher funktioniert das. Ich war in Schwierigkeiten, weil ich einen Fehler mit permmision bekommen habe, aber ich habe das Gefühl, dass ich einen Fehler bekommen habe, weil ich überhaupt nicht genug Ordner und Dateien hatte. Ich erinnere mich, dass ich etwas hinzugefügt habe, indem ich mir das Fehlerprotokoll von Spark oder so angesehen habe.

Spark-Programmhandbuch

http://spark.apache.org/docs/latest/programming-guide.html Mastering Apache Spark 2 https://jaceklaskowski.gitbooks.io/mastering-apache-spark/content/spark-mllib/spark-mllib-transformers.html

Scala-Schlüsselwort

http://yuroyoro.hatenablog.com/entry/20100317/1268819400

Liste ist wichtig in Scala Verwenden Sie die Case-Klasse Unveränderliches Programm

Kooperative Filterung

http://en.wikipedia.org/wiki/Collaborative_filtering http://d.hatena.ne.jp/EulerDijkstra/20130407/1365349866 https://www.slideshare.net/hoxo_m/ss-53305070

Setup beim Clustering mit aws

https://www.youtube.com/watch?v=qIs4nNFgi0s