Отправка скрипта Python с помощью API-интерфейса Apache Spark Hidden REST

Question

Мне нужно отправить файл py с API Apache Spark Hidden REST. Как я следовал учебнику arturmkrtchyan , я не смог найти ни одного примера или документа относительно того, как отправить файл py.

У кого-нибудь есть идеи?Можно ли заменить файл py вместо jar:

curl -X POST http://spark-cluster-ip:6066/v1/submissions/create --header "Content-Type:application/json;charset=UTF-8" --data '{
    "action" : "CreateSubmissionRequest",
      "appArgs" : [ "myAppArgument1" ],
      "appResource" : "file:/path/to/py/file/file.py",
      "clientSparkVersion" : "1.5.0",
      "environmentVariables" : {
        "SPARK_ENV_LOADED" : "1"
      },
      "mainClass" : "com.mycompany.MyJob",
      "sparkProperties" : {
        "spark.submit.pyFiles": "/path/to/py/file/file.py",
        "spark.driver.supervise" : "false",
        "spark.app.name" : "MyJob",
        "spark.eventLog.enabled": "true",
        "spark.submit.deployMode" : "cluster",
        "spark.master" : "spark://spark-cluster-ip:6066"
      }
    }'

Или есть другой способ сделать это?

Answer 1

Подход на самом деле похож на тот, который описан в ссылке, которой вы поделились.

Вот пример:

Давайте сначала определим скрипт Python, который нам нужно запустить.Я взял пример spark pi, то есть spark_pi.py:

from __future__ import print_function

import sys
from random import random
from operator import add

from pyspark.sql import SparkSession


if __name__ == "__main__":
    """
        Usage: pi [partitions]
    """
    spark = SparkSession\
        .builder\
        .appName("PythonPi")\
        .getOrCreate()

    partitions = int(sys.argv[1]) if len(sys.argv) > 1 else 2
    n = 100000 * partitions

    def f(_):
        x = random() * 2 - 1
        y = random() * 2 - 1
        return 1 if x ** 2 + y ** 2 <= 1 else 0

    count = spark.sparkContext.parallelize(range(1, n + 1), partitions).map(f).reduce(add)
    print("Pi is roughly %f" % (4.0 * count / n))

    spark.stop()

Вам нужно убедиться, что /tmp/spark-events уже есть, прежде чем запускать задание.

Теперь вы можетеотправьте следующим образом:

curl -X POST http://[spark-cluster-ip]:6066/v1/submissions/create --header "Content-Type:application/json;charset=UTF-8" --data '{
   "action":"CreateSubmissionRequest",
   "appArgs":[
      "/home/eliasah/Desktop/spark_pi.py"
   ],
   "appResource":"file:/home/eliasah/Desktop/spark_pi.py",
   "clientSparkVersion":"2.2.1",
   "environmentVariables":{
      "SPARK_ENV_LOADED":"1"
   },
   "mainClass":"org.apache.spark.deploy.SparkSubmit",
   "sparkProperties":{
      "spark.driver.supervise":"false",
      "spark.app.name":"Simple App",
      "spark.eventLog.enabled":"true",
      "spark.submit.deployMode":"cluster",
      "spark.master":"spark://[spark-master]:6066"
   }
}' 

Как вы заметили, мы указали путь к файлу нашего скрипта в качестве ресурсов приложения, а также в качестве аргументов приложения.

PS: Замените [spark-cluster-ip] и [spark-master] на соответствующие значения, соответствующие вашему искровому кластеру.

Это приведет к следующему:

{
  "action" : "CreateSubmissionResponse",
  "message" : "Driver successfully submitted as driver-20180522165321-0001",
  "serverSparkVersion" : "2.2.1",
  "submissionId" : "driver-20180522165321-0001",
  "success" : true
}

Вы также можете проверить Spark UI , чтобы контролировать свою работу.

Чтобы использовать аргументы со сценарием ввода, вы можете добавить их в свойство appArgs:

"appArgs": [ "/home/eliasah/Desktop/spark_pi.py", "arg1" ]