Zum ausführen mehrerer jobs in einer Sparkcontext von separaten threads in PySpark?

Es versteht sich von Spark Dokumentation über Scheduling Innerhalb einer Anwendung:

Innerhalb einer bestimmten Funke Anwendung (SparkContext-Instanz), die mehrere parallele jobs können gleichzeitig ausgeführt werden, wenn Sie eingereicht wurden aus separaten threads. Durch "job" in diesem Abschnitt, meinen wir, ein Funke-Aktion (z.B. speichern, sammeln) und alle Aufgaben, die ausgeführt werden müssen, zu bewerten, die Aktion. Spark ' s scheduler ist vollständig thread-safe und die diesen Anwendungsfall unterstützt, damit die Anwendungen, die dazu dienen, mehrere Anfragen (z.B. Abfragen für mehrere Benutzer)."

Konnte ich einige Beispiel-code in Scala und Java.
Kann jemand ein Beispiel geben, wie diese umgesetzt werden können, mit PySpark?

Jemals eine Antwort hier? Ich versuche das gleiche zu tun, und denken, es ist eigentlich unmöglich, bis bessere Verriegelung Hinzugefügt SparkContexts.
was meinst du ? Dieses Stück doc reden auch nicht über einen spezifischen Spark-API, es nur scheint zu funktionieren für alle. Der eigentliche code, der ausgeführt wird, wenn Sie mithilfe der APIs ist der Scala-code, und einige interface-code für Java und Python.
Kannst du den link, wo diese Aussage herkommt?

InformationsquelleAutor Meethu Mathew | 2015-05-13

Ich lief in das gleiche Problem, so habe ich eine kleine Selbstversorger-Beispiel. Ich erstelle mehrere threads mit python threading Modul und legt mehrere spark-jobs gleichzeitig.

Beachten Sie, dass standardmäßig, Funken, führen die jobs, die in First-In-First-Out (FIFO): http://spark.apache.org/docs/latest/job-scheduling.html#scheduling-within-an-application. Im Beispiel unten, ändere ich es um GERECHTE Planung

# Prereqs:
# set 
# spark.dynamicAllocation.enabled         true
# spark.shuffle.service.enabled           true
  spark.scheduler.mode                    FAIR
# in spark-defaults.conf

import threading
from pyspark import SparkContext, SparkConf

def task(sc, i):
  print sc.parallelize(range(i*10000)).count()

def run_multiple_jobs():
  conf = SparkConf().setMaster('local[*]').setAppName('appname')
  # Set scheduler to FAIR: http://spark.apache.org/docs/latest/job-scheduling.html#scheduling-within-an-application
  conf.set('spark.scheduler.mode', 'FAIR')
  sc = SparkContext(conf=conf)
  for i in range(4):
    t = threading.Thread(target=task, args=(sc, i))
    t.start()
    print 'spark task', i, 'has started'


run_multiple_jobs()

Ausgabe:

spark task 0 has started
spark task 1 has started
spark task 2 has started
spark task 3 has started
30000
0 
10000
20000

eine Idee ist dies ist der beste Weg, es zu tun? Vor allem, wenn Sie auf einem cluster. Da der sc bleibt auf master und der master verteilt diese auf worker-Knoten, wurde ich denken, wenn dies ist der beste Weg, es zu tun.
Es gibt einige schöne Noten hier auf mit der threading-Bibliothek, vor allem für Dinge wie die Rückgabe der Ergebnisse der threaded-Berechnungen.

InformationsquelleAutor sparknoob

3

Heute, ich Frage mich das gleiche. Das multiprocessing-Modul bietet eine ThreadPool, die laichen ein paar threads für dich und läuft somit die jobs parallel. Zuerst instanziieren Sie die Funktionen ein, erstellen Sie dann den Pool, und dann map es über den Bereich, den Sie Durchlaufen.

In meinem Fall war ich der Berechnung dieser WSSSE zahlen für die unterschiedliche Anzahl der Zentren (hyperparameter tuning), um eine "gute" k-means-clustering ... so wie es beschrieben ist in der MLSpark Dokumentation. Ohne weitere Erklärungen, hier sind einige Zellen aus meinem IPython Arbeitsblatt:
```
from pyspark.mllib.clustering import KMeans
import numpy as np
```
c_points sind 12dim arrays:
```
>>> c_points.cache()
>>> c_points.take(3)
[array([ 1, -1,  0,  1,  0,  0,  0,  0,  0,  0,  0,  0]),
array([-2,  0,  0,  1,  0,  0,  0,  0,  0,  0,  0,  0]),
array([ 7, -1,  1,  0,  0,  0,  0,  0,  0,  0,  0,  0])]
```
In der folgenden, für jede i ich bin computing dieses WSSSE Wert und er es als ein Tupel:
```
def error(point, clusters):
    center = clusters.centers[clusters.predict(point)]
    return np.linalg.norm(point - center)

def calc_wssse(i):
    clusters = KMeans.train(c_points, i, maxIterations=20,
        runs=20, initializationMode="random")
    WSSSE = c_points\
        .map(lambda point: error(point, clusters))\
        .reduce(lambda x, y: x + y)
    return (i, WSSSE)
```
Hier beginnt der interessante Teil:
```
from multiprocessing.pool import ThreadPool
tpool = ThreadPool(processes=4)
```
Ausführen:
```
wssse_points = tpool.map(calc_wssse, range(1, 30))
wssse_points
```
gibt:
```
[(1, 195318509740785.66),
 (2, 77539612257334.33),
 (3, 78254073754531.1),
 ...
]
```
- ... und aus Neugier, habe ich die benchmark, die es über %timeit. die serielle Ausführung (mit einem "pulsierenden" Verhalten) nahm 53.2 Sekunden, während die parallelisierte Ansatz mit 4 threads fertig in 16.2 Sek. Also, es ist wirklich ein Unterschied. Mehr aktive Phasen parallel und immer einige in der Warteschlange.
- Nicht diese erhöhen die Möglichkeit von Race Conditions?
- Es kommt wohl auf die Daten, die Sie manipulieren. Da Sie verwalten das threading, die Sie haben, um sicherzustellen, dass Sie nicht erhöhen Race-Bedingungen selbst.
InformationsquelleAutor Harald Schilly

Schreibe einen Kommentar

Du musst angemeldet sein, um einen Kommentar abzugeben.