Изменить PYSPARK_ PYTHON для работников Spark - PullRequest
Новая
       23

Изменить PYSPARK_ PYTHON для работников Spark

0 голосов
/ 24 марта 2020

Мы распространяем наше приложение Python, которое использует Spark, вместе с Python 3.7 интерпретатором (python.exe со всеми необходимыми библиотеками лежит около MyApp.exe).

Для установки PYSPARK_PYTHON мы имеем есть функция, которая определяет путь к нашему python.exe: 

os.environ['PYSPARK_PYTHON'] = get_python()

в Windows PYSPARK_PYTHON станет C:/MyApp/python.exe
в Ubuntu PYSPARK_PYTHON станет /opt/MyApp/python.exe

Запускаем узел master / driver и создаем SparkSession на Windows. Затем мы запускаем рабочий узел в Ubuntu, но рабочий не работает с: 

Job aborted due to stage failure: Task 1 in stage 11.0 failed 4 times, most recent failure: Lost task 1.3 in stage 11.0 (TID 1614, 10.0.2.15, executor 1): java.io.IOException: Cannot run program "C:/MyApp/python.exe": error=2, No such file or directory

Конечно, в Ubuntu C:/MyApp/python.exe нет.

Если я правильно понимаю эту ошибку, PYSPARK_PYTHON из драйвера отправляется всем работникам.

Также пытался установить PYSPARK_PYTHON в spark-env.sh и spark-defaults.conf. Как я могу изменить PYSPARK_PYTHON для работников Ubuntu, чтобы они стали /opt/MyApp/python.exe?

1 Ответ

0 голосов
/ 24 марта 2020

Просматривая исходный код, похоже, что код драйвера Python помещает значение исполняемого пути Python из контекста Spark при создании рабочих элементов для запуска функций Python в spark/rdd.py: 

def _wrap_function(sc, func, deserializer, serializer, profiler=None):
    assert deserializer, "deserializer should not be empty"
    assert serializer, "serializer should not be empty"
    command = (func, profiler, deserializer, serializer)
    pickled_command, broadcast_vars, env, includes = _prepare_for_python_RDD(sc, command)
    return sc._jvm.PythonFunction(bytearray(pickled_command), env, includes, sc.pythonExec,
                                                                             ^^^^^^^^^^^^^
                                  sc.pythonVer, broadcast_vars, sc._javaAccumulator)

Затем бегун Python PythonRunner.scala использует путь, сохраненный в первом полученном рабочем элементе, для запуска новых экземпляров интерпретатора: 

private[spark] abstract class BasePythonRunner[IN, OUT](
    funcs: Seq[ChainedPythonFunctions],
    evalType: Int,
    argOffsets: Array[Array[Int]])
  extends Logging {
  ...
  protected val pythonExec: String = funcs.head.funcs.head.pythonExec
                                     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  ...
  def compute(
      inputIterator: Iterator[IN],
      partitionIndex: Int,
      context: TaskContext): Iterator[OUT] = {
    ...
    val worker: Socket = env.createPythonWorker(pythonExec, envVars.asScala.toMap)
    ...
  }
  ...
}

Исходя из этого, я боюсь, что в настоящее время кажется невозможным иметь отдельные конфигурации для исполняемого файла Python в мастере и в рабочих. Также см. Третий комментарий к выпуску SPARK-26404 . Возможно, вам следует подать RFE в проект Apache Spark.

Я не гуру Spark, и, возможно, есть способ сделать это, возможно, установив PYSPARK_PYTHON просто "python" и затем убедитесь, что система PATH настроена соответствующим образом, чтобы ваш исполняемый файл Python был первым.

...