Ich habe den Code geschrieben, um die Ähnlichkeit des Inhalts in pyspark zu messen. Es gab nur wenige japanische Dokumente, daher werde ich dies notieren. Ich habe Docker verwendet, weil es schmerzhaft war, Funken von Grund auf neu einzuführen.
Docker vorstellen http://qiita.com/hshimo/items/e24b1fbfbf775ec7c941 Ich bezog mich auf.
Grundsätzlich Get started with Docker for Mac Laden Sie einfach weitere dmg-Dateien herunter und installieren Sie sie.
Klonen Sie das Docker-Image von spark + jupyter. Ich habe es auf https://github.com/busbud/jupyter-docker-stacks/tree/master/all-spark-notebook geklont. Wie in der README angegeben
$ docker run -d -p 8888:8888 jupyter/all-spark-notebook -e GRANT_SUDO=yes
Geben Sie den obigen Befehl am Terminal ein. Daraufhin wird ein Jupyter-Notizbuch angezeigt, das pyspark unter localhost: 8888 verwenden kann. (-E GRANT_SUDO = yes ist eine Option, mit der Sie Jupyter ohne Benutzerrechte verwenden können.)
Wenn Sie den Docker-Befehl ps drücken
$ docker ps
CONTAINER ID IMAGE COMMAND CREATED STATUS PORTS NAMES
91fc42290759 jupyter/all-spark-notebook "tini -- start-not..." 3 hours ago Up 3 hours 0.0.0.0:8888->8888/tcp nifty_bartik
Sie können sehen, dass das angegebene Docker ausgeführt wird.
Der erstellte Code ist https://github.com/kenchin110100/machine_learning/blob/master/samplePysparkSimilarity.ipynb Es ist in.
Importieren Sie zunächst die erforderlichen Bibliotheken und initialisieren Sie pyspark.
# coding: utf-8
"""
Beispiel für Ähnlichkeit mit pyspark
"""
import numpy as np
from pyspark import SQLContext, sql
import pyspark
from pyspark.sql import functions, Row
from pyspark.mllib.linalg import DenseVector
sc = pyspark.SparkContext('local[*]')
sqlContext = sql.SQLContext(sc)
sc ist eine Instanz, die erforderlich ist, um den RDD-Typ von pyspark zu verwenden. sqlContext ist eine Instanz, die zur Verwendung des DataFrame-Typs erforderlich ist.
Als Nächstes habe ich Beispieldaten erstellt, um die Ähnlichkeit zu messen.
#Beispieldaten erstellen
samples = [['aaa', 'a', 30, 1,2,3,4,5] + np.random.randn(5).tolist(),
['aaa', 'b', 30,2,1,3,4,1] + np.random.randn(5).tolist(),
['bbb', 'a', 30,4,5,3,2,4] + np.random.randn(5).tolist(),
['bbb', 'b', 30,1,2,4,3,1] + np.random.randn(5).tolist(),
['ccc', 'a', 30,4,5,2,1,2] + np.random.randn(5).tolist(),
['ccc', 'b', 30,1,2,5,4,1] + np.random.randn(5).tolist(),]
#Spaltennamen erstellen
colnames = [
'mc', 'mtc', 'area_cd',
'label1', 'label2', 'label3', 'label4', 'label5',
'label6', 'label7', 'label8', 'label9', 'label10'
]
colnames1 = [col + '_1' for col in colnames]
colnames2 = [col + '_2' for col in colnames]
#Konvertieren Sie die erstellten Beispieldaten in den Typ pyspark DataFrame
df1 = sqlContext.createDataFrame(sc.parallelize(samples), colnames1)
df2 = sqlContext.createDataFrame(sc.parallelize(samples), colnames2)
Wir betrachten mc und mtc als eindeutige Schlüssel und label1 bis label10 als Identitäten.
Die gleichen Beispieldaten werden in zwei Datenrahmen gespeichert Hiermit wird eine Kombination für Ähnlichkeit mit Join erstellt. Nach dem Konvertieren von Samples in den RDD-Typ mit sc.parallelize (Samples) wird es mit createDataFrame in den DataFrame-Typ konvertiert.
Verwenden Sie dann einen Join vom Typ DataFrame, um die Kombinationen aufzulisten, die die Ähnlichkeit messen.
joined_df = df1.join(df2, df1.area_cd_1 == df2.area_cd_2).filter(functions.concat(df1.mc_1, df1.mtc_1) != functions.concat(df2.mc_2, df2.mtc_2))
Join vom Typ DataFrame
df1.join(df2, <Bedingungen>, 'left' or 'inner' or ...)
Sie können es mit erstellen. In dem diesmal erstellten Code wird durch Ausführen eines Filters nach dem Join Folgendes ausgeführt: Ich versuche die Ähnlichkeit mit mir selbst nicht zu messen.
functions.concat(df1.mc_1, df1.mtc_1)
Durch Kombinieren der beiden Schlüssel von mc und mtc, die Sie eindeutig machen möchten, werden sie dann als ein eindeutiger Schlüssel behandelt.
Wir werden die Ähnlichkeit mit dem bisher erstellten DataFrame berechnen.
Definieren Sie zunächst die Funktion.
def match_sim(row1 ,row2):
keys = row1.asDict().keys()
total = len(keys)
count = 0
for key in keys:
if row1[key] == row2[key]:
count += 1
return float(count)/total
def cosine_sim(vec1 ,vec2):
dot = abs(vec1.dot(vec2))
n1 = vec1.norm(None)
n2 = vec1.norm(None)
return float(dot/(n1*n2))
match_sim ist eine Funktion zum Messen der Ähnlichkeit kategorialer Variablen. Übergeben Sie den Row-Typ des Pysparks. Gibt 1 zurück, wenn sie übereinstimmen, 0, wenn sie nicht übereinstimmen, und gibt den Wert geteilt durch die verglichenen Primzahlen zurück.
cosine_sim ist eine Funktion zur Berechnung der Cosinusähnlichkeit Übergeben Sie den DenseVector-Typ von pyspark.mllib.
Verwenden Sie diese Funktion, um die Ähnlichkeit für jede Zeile zu berechnen.
joined_rdd = joined_df.rdd.map(lambda x: (
Row(mc_1=x.mc_1, mtc_1=x.mtc_1, mc_2=x.mc_2, mtc_2=x.mtc_2),
Row(label1=x.label1_1, label2=x.label2_1, label3=x.label3_1, label4=x.label4_1, label5=x.label5_1),
DenseVector([x.label6_1,x.label7_1,x.label8_1,x.label9_1,x.label10_1]),
Row(label1=x.label1_2, label2=x.label2_2, label3=x.label3_2, label4=x.label4_2, label5=x.label5_2),
DenseVector([x.label6_2,x.label7_2,x.label8_2,x.label9_2,x.label10_2])
)) \
.map(lambda x: (x[0], match_sim(x[1], x[3]), cosine_sim(x[2], x[4]))) \
.map(lambda x: (x[0].mc_1, x[0].mtc_1, x[0].mc_2, x[0].mtc_2, x[1], x[2]))
Konvertieren Sie zunächst den zuvor erstellten DataFrame-Typ join_df in den rdd-Typ und ordnen Sie ihn zu (1. Zeile). Speichern Sie 5 Datentypen für jede Kombination. Zeile (mc_1 = x.mc_1 ...) ist eine Zeile (2. Zeile) zum Speichern eines eindeutigen Schlüssels für Ähnlichkeit. Zeile (label1 = x.label1_1 ...) ist die Zeile zum Speichern kategorialer Variablen (3. Zeile) DenseVector (x.label6_1, ...) ist ein Vektor zum Speichern kontinuierlicher Variablen (4. Zeile) Die 5. und 6. Zeile speichern die kategorialen Variablen und kontinuierlichen Variablen der anderen Zeile auf Ähnlichkeit.
Ordnen Sie das RDD weiter zu, das die 5 auf diese Weise erstellten Datentypen speichert (Zeile 8). Mit x [0] wird die Übereinstimmungsähnlichkeit für x [1] und x [3] berechnet, und die Kosinusähnlichkeit wird für x [2] und x [4] berechnet. Formatieren Sie es schließlich so, dass es wieder an den DataFrame-Typ übergeben werden kann (Zeile 9).
Die auf diese Weise erstellte Ähnlichkeitstabelle lautet wie folgt.
sqlContext.createDataFrame(joined_rdd, ['tar_mc', 'tar_mtc', 'res_mc', 'res_mtc', 'match_sim', 'cosine_sim']).show()
+------+-------+------+-------+---------+--------------------+
|tar_mc|tar_mtc|res_mc|res_mtc|match_sim| cosine_sim|
+------+-------+------+-------+---------+--------------------+
| aaa| a| aaa| b| 0.4| 0.2979433262317515|
| aaa| a| bbb| a| 0.2| 0.2161103600613806|
| aaa| a| bbb| b| 0.4| 0.6933162039799152|
| aaa| a| ccc| a| 0.0| 0.34941331375143353|
| aaa| a| ccc| b| 0.6| 0.5354750033557132|
| aaa| b| aaa| a| 0.4| 0.19428899651078324|
| aaa| b| bbb| a| 0.2| 0.10702152405150611|
| aaa| b| bbb| b| 0.2| 0.4033681950723296|
| aaa| b| ccc| a| 0.0| 0.20097172584128625|
| aaa| b| ccc| b| 0.4| 0.6861144738544892|
| bbb| a| aaa| a| 0.2| 0.3590385377694502|
| bbb| a| aaa| b| 0.2| 0.27266040008605663|
| bbb| a| bbb| b| 0.0| 1.1313716028957246|
| bbb| a| ccc| a| 0.4|0.009321106239696326|
| bbb| a| ccc| b| 0.0| 1.0017633803368193|
| bbb| b| aaa| a| 0.4| 0.2176828683879606|
| bbb| b| aaa| b| 0.2| 0.194213765887726|
| bbb| b| bbb| a| 0.0| 0.21381230488831227|
| bbb| b| ccc| a| 0.0| 0.21074015342537053|
| bbb| b| ccc| b| 0.6| 0.34536679942567616|
+------+-------+------+-------+---------+--------------------+
only showing top 20 rows
Für jeden Schlüssel werden die Ähnlichkeit und die Kosinusähnlichkeit gemäß dem Übereinstimmungsgrad der Kategorie berechnet.
Dieses Mal habe ich Dockers Jupiter-Funken-Bild verwendet, um ein Beispiel zu erstellen, um die Ähnlichkeit zwischen Daten zu messen. Es sollte eine präzisere Möglichkeit geben, es zu schreiben (z. B. mit ColumnSimilarity von MLlib), aber aus verschiedenen Gründen habe ich dieses Mal diesen Kreisverkehrscode geschrieben. Ich bin ein Anfänger von Docker und Spark, daher möchte ich von nun an ein bisschen mehr üben.
Recommended Posts