Как избежать одного окна Spark Streaming, блокирующего другое окно с запуском некоторого собственного кода Python

Я запускаю Spark Streaming с двумя разными окнами (в окне для обучения модели с SKLearn, а другой для прогнозирования значений на основе этой модели), и мне интересно, как я могу избежать одного окна ( "медленное" обучение окно) для обучения модели без "блокировки" окна "быстрого" прогноза.
Мой упрощенный код выглядит следующим образом:

conf = SparkConf()
conf.setMaster("local[4]")
sc = SparkContext(conf=conf)
ssc = StreamingContext(sc, 1)

stream = ssc.socketTextStream("localhost", 7000)


import Custom_ModelContainer

### Window 1 ###
### predict data based on model computed in window 2 ###

def predict(time, rdd):
    try:
       # ... rdd conversion to df, feature extraction etc...

       # regular python code 
       X = np.array(df.map(lambda lp: lp.features.toArray()).collect())
       pred = Custom_ModelContainer.getmodel().predict(X)

       # send prediction to GUI

    except Exception, e: print e

predictionStream = stream.window(60,60)
predictionStream.foreachRDD(predict)


### Window 2 ###
### fit new model ###

def trainModel(time, rdd):
try:
    # ... rdd conversion to df, feature extraction etc...

    X = np.array(df.map(lambda lp: lp.features.toArray()).collect())
    y = np.array(df.map(lambda lp: lp.label).collect())

    # train test split etc...

    model = SVR().fit(X_train, y_train)
    Custom_ModelContainer.setModel(model)

except Exception, e: print e

modelTrainingStream = stream.window(600,600)
modelTrainingStream.foreachRDD(trainModel)

(Примечание: Custom_ModelContainer - это класс, который я написал для сохранения и получения обучаемой модели)

Моя настройка работает нормально, за исключением того, что каждый раз, когда новая модель обучается во втором окне (что занимает около минуты), первые окна не вычисляют прогнозы до завершения обучения модели. Фактически, я предполагаю, что это имеет смысл, поскольку подгонка моделей и предсказания вычисляются на главном node (в нераспределенной настройке - из-за SKLearn).

Итак, мой вопрос заключается в следующем: можно ли подготовить модель для одного рабочего node (вместо мастера node)? Если да, то как я мог бы достичь последнего и действительно ли разрешил бы мою проблему?

Если нет, любое другое предложение о том, как я мог бы выполнить такую настройку, без задержки вычислений в окне 1?

Любая помощь очень ценится.

EDIT: Я думаю, что более общий вопрос: Как я могу запустить две разные задачи для двух разных рабочих параллельно?

Ответ 1

Отказ от ответственности: это всего лишь набор идей. Ни одно из них не было проверено на практике.

Несколько вещей, которые вы можете попробовать:

Не collect до predict. scikit-learn модели обычно сериализуемы, поэтому процесс прогнозирования можно легко обрабатывать в кластере:
```
def predict(time, rdd):
    ... 

    model = Custom_ModelContainer.getmodel()
    pred = (df.rdd.map(lambda lp: lp.features.toArray())
        .mapPartitions(lambda iter: model.predict(np.array(list(iter)))))
    ...
```
Он должен не только распараллеливать прогнозы, но также, если необработанные данные не передаются в GUI, уменьшите количество данных, которые необходимо собрать.

Попробуйте collect и отправьте данные асинхронно. PySpark не предоставляет метод collectAsync, но вы можете попытаться достичь чего-то подобного с concurrent.futures:

from pyspark.rdd import RDD
from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor(max_workers=4)

def submit_to_gui(*args): ...

def submit_if_success(f):
    if not f.exception():
        executor.submit(submit_to_gui, f.result())

продолжить с 1.

def predict(time, rdd):
    ...
    f = executor.submit(RDD.collect, pred)
    f.add_done_callback(submit_if_success)
    ...

Если вы действительно хотите использовать локальную модель scikit-learn, попробуйте collect и fit использовать фьючерсы, как указано выше. Вы также можете попытаться собрать только один раз, особенно если данные не кэшируются:
```
def collect_and_train(df):
    y, X = zip(*((p.label, p.features.toArray()) for p in df.collect()))
    ...
    return SVR().fit(X_train, y_train)

def set_if_success(f):
    if not f.exception():
        Custom_ModelContainer.setModel(f.result())  

def trainModel(time, rdd): 
   ...
    f = excutor.submit(collect_and_train, df)
    f.add_done_callback(set_if_success) 
   ...
```
Перенесите процесс обучения в кластер либо с помощью уже существующих решений, например spark-sklearn, либо настраиваемого подхода:
- наивное решение - подготовьте свои данные, coalesce(1) и подготовьте одну модель, используя mapPartitions.
- распределенное решение - создайте и подтвердите отдельную модель для каждого раздела с помощью mapPartitions, собирайте модели и используйте в качестве ансамбля, например, принимая среднее или медианное предсказание.
Отбросьте scikit-learn и используйте модель, которую можно обучить и поддерживать в распределенной потоковой среде (например StreamingLinearRegressionWithSGD).

Ваш текущий подход делает Spark устаревшим. Если вы можете обучать модель локально, есть хороший шанс, что вы можете выполнять все другие задачи намного быстрее на локальной машине. В противном случае ваша программа просто выйдет из строя на collect.

Ответ 2

Я думаю, что вы ищете свойство: "spark.streaming.concurrentJobs", которое по умолчанию равно 1. Увеличение этого должно позволить вам параллельно запускать несколько функций foreachRDD.

В JobScheduler.scala:

private val numConcurrentJobs = ssc.conf.getInt("spark.streaming.concurrentJobs", 1)

Напоминаем, что вы также должны знать о безопасности потоков в своем контейнере пользовательской модели, если вы собираетесь мутировать и читать параллельно.:)