Установить путь NFS для распределенной спарк-сессии - PullRequest
Новая
       23

Установить путь NFS для распределенной спарк-сессии

0 голосов
/ 11 мая 2019

Я пытаюсь продемонстрировать выполнение сценария Pyspark Mllib в распределенном кластере Spark. Он работает в локальном сеансе Spark (Windows 10), но не работает, когда мастер настроен на удаленную машину с двумя подчиненными (все Ubuntu 18.04). Я думаю, это потому, что локальный файл обучающих данных должен быть доступен со всех узлов. Для этого я настроил NFS на главном сервере для совместного использования файла.

В качестве демонстрации я хотел бы избежать любых тяжелых решений, таких как установка Hadoop или использование корзины S3. Документы Spark говорят, что я могу использовать NFS, поэтому я установил NFS-сервер на главном узле Spark, создал каталог «share» в корне, поделился им с подчиненными узлами, смонтировал его с подчиненных узлов как «share» и проверил, что он работает из командной строки master и slave.

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

```
"""
Random Forest Regressor Example - modified for distributed cluster.
"""
from __future__ import print_function

from pyspark.ml import Pipeline
from pyspark.ml.regression import RandomForestRegressor
from pyspark.ml.feature import VectorIndexer
from pyspark.ml.evaluation import RegressionEvaluator

from pyspark.sql import SparkSession

if __name__ == "__main__":    
    spark = SparkSession\
        .builder\
        .appName("RandomForestRegressorExample")\
        .master("spark://10.20.0.163:7077")\
        .getOrCreate()

    data = spark.read.format("libsvm").load\
    ("file:////share//sample_libsvm_data.txt")

```

Когда я запускаю скрипт, я получаю сообщение об ошибке «AnalysisException:« Путь не существует: file: /share/sample_libsvm_data.txt; »*

Полный ответ об ошибке: 

---------------------------------------------------------------------------
Py4JJavaError                             Traceback (most recent call last)
C:\ProgramData\Anaconda3\lib\site-packages\pyspark\sql\utils.py in deco(*a, **kw)
     62         try:
---> 63             return f(*a, **kw)
     64         except py4j.protocol.Py4JJavaError as e:

C:\ProgramData\Anaconda3\lib\site-packages\py4j\protocol.py in get_return_value(answer, gateway_client, target_id, name)
    327                     "An error occurred while calling {0}{1}{2}.\n".
--> 328                     format(target_id, ".", name), value)
    329             else:

Py4JJavaError: An error occurred while calling o108.load.
: org.apache.spark.sql.AnalysisException: Path does not exist: file:/share/sample_libsvm_data.txt;
    at org.apache.spark.sql.execution.datasources.DataSource$$anonfun$org$apache$spark$sql$execution$datasources$DataSource$$checkAndGlobPathIfNecessary$1.apply(DataSource.scala:558)
    at org.apache.spark.sql.execution.datasources.DataSource$$anonfun$org$apache$spark$sql$execution$datasources$DataSource$$checkAndGlobPathIfNecessary$1.apply(DataSource.scala:545)
    at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:241)
    at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:241)
    at scala.collection.immutable.List.foreach(List.scala:392)
    at scala.collection.TraversableLike$class.flatMap(TraversableLike.scala:241)
    at scala.collection.immutable.List.flatMap(List.scala:355)
    at org.apache.spark.sql.execution.datasources.DataSource.org$apache$spark$sql$execution$datasources$DataSource$$checkAndGlobPathIfNecessary(DataSource.scala:545)
    at org.apache.spark.sql.execution.datasources.DataSource.resolveRelation(DataSource.scala:359)
    at org.apache.spark.sql.DataFrameReader.loadV1Source(DataFrameReader.scala:223)
    at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:211)
    at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:178)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
    at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
    at py4j.Gateway.invoke(Gateway.java:282)
    at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
    at py4j.commands.CallCommand.execute(CallCommand.java:79)
    at py4j.GatewayConnection.run(GatewayConnection.java:238)
    at java.lang.Thread.run(Thread.java:748)


During handling of the above exception, another exception occurred:

AnalysisException                         Traceback (most recent call last)
<ipython-input-22-fcd877efb9c4> in <module>
     37         .getOrCreate()
     38 
---> 39     data = spark.read.format("libsvm").load("file:////share//sample_libsvm_data.txt")
     40 

C:\ProgramData\Anaconda3\lib\site-packages\pyspark\sql\readwriter.py in load(self, path, format, schema, **options)
    164         self.options(**options)
    165         if isinstance(path, basestring):
--> 166             return self._df(self._jreader.load(path))
    167         elif path is not None:
    168             if type(path) != list:

C:\ProgramData\Anaconda3\lib\site-packages\py4j\java_gateway.py in __call__(self, *args)
   1255         answer = self.gateway_client.send_command(command)
   1256         return_value = get_return_value(
-> 1257             answer, self.gateway_client, self.target_id, self.name)
   1258 
   1259         for temp_arg in temp_args:

C:\ProgramData\Anaconda3\lib\site-packages\pyspark\sql\utils.py in deco(*a, **kw)
     67                                              e.java_exception.getStackTrace()))
     68             if s.startswith('org.apache.spark.sql.AnalysisException: '):
---> 69                 raise AnalysisException(s.split(': ', 1)[1], stackTrace)
     70             if s.startswith('org.apache.spark.sql.catalyst.analysis'):
     71                 raise AnalysisException(s.split(': ', 1)[1], stackTrace)

AnalysisException: 'Path does not exist: file:/share/sample_libsvm_data.txt;'

У меня есть несколько вопросов:

1) Есть ли лучший способ получить данные для распределенного кластера из одного файла, который можно быстро настроить? Я попытался загрузить в локальный фрейм данных сеанса, а затем скопировать его в распределенный фрейм данных, но также не смог этого достичь.

2) В противном случае, как мне настроить общий ресурс NSF, чтобы он работал?

3) Или я делаю какую-то другую ошибку, которая мешает ее работе?

...