Использование Spark на многоузловом кластере с pyspark

Question

Я настроил Spark для многоузлового кластера, как описано в этой статье: https://medium.com/ymedialabs-innovation/apache-spark-on-a-multi-node-cluster-b75967c8cb2b

Я проверил, что все сотрудники находятся в состоянии "ALIVE", используя веб-интерфейс: http://MASTER -IP: 8080 /

Я также проверил, что могу запустить оболочку spark, выполнить команды Scala и посмотреть результаты работы в http://MASTER -HOSTNAME: 4041 .

В Python я могу создать контекст pyspark без указания мастера ИЛИ, установив для мастера значение local[*].Ниже приведен основной скрипт Python, который я создал для тестирования.

# spark-basic.py
from pyspark import SparkConf
from pyspark import SparkContext

conf = SparkConf()
#conf.setMaster('spark://<MASTER-HOSTNAME>:7077')
conf.setMaster('local[*]')
conf.setAppName('spark-basic')
sc = SparkContext(conf=conf)

def mod(x):
    import numpy as np
    return (x, np.mod(x, 2))

rdd = sc.parallelize(range(1000)).map(mod).take(10)
print(rdd)

Почему я не могу установить мастер на 'spark://<MASTER-HOSTNAME>:7077', как описано во многих онлайн-учебниках, таких как this one?Ошибка, которую я получаю, когда я пытаюсь сделать это, довольно длинная, но вот ее часть (надеюсь, этой части сообщения об ошибке достаточно, чтобы кто-нибудь мне помог, но я всегда могу вставить сюда все это):

19/05/01 14:29:17 WARN StandaloneAppClient$ClientEndpoint: Failed to 
connect to master MASTER-HOSTNAME:7077
org.apache.spark.SparkException: Exception thrown in awaitResult
    at org.apache.spark.rpc.RpcTimeout$$anonfun$1.applyOrElse(RpcTimeout.scala:77)
    at org.apache.spark.rpc.RpcTimeout$$anonfun$1.applyOrElse(RpcTimeout.scala:75)
    at scala.runtime.AbstractPartialFunction.apply(AbstractPartialFunction.scala:36)
    at org.apache.spark.rpc.RpcTimeout$$anonfun$addMessageIfTimeout$1.applyOrElse(RpcTimeout.scala:59)
    at org.apache.spark.rpc.RpcTimeout$$anonfun$addMessageIfTimeout$1.applyOrElse(RpcTimeout.scala:59)
    at scala.PartialFunction$OrElse.apply(PartialFunction.scala:167)
    at org.apache.spark.rpc.RpcTimeout.awaitResult(RpcTimeout.scala:83)
    at org.apache.spark.rpc.RpcEnv.setupEndpointRefByURI(RpcEnv.scala:100)
    at org.apache.spark.rpc.RpcEnv.setupEndpointRef(RpcEnv.scala:108)
    at org.apache.spark.deploy.client.StandaloneAppClient$ClientEndpoint$$anonfun$tryRegisterAllMasters$1$$anon$1.run(StandaloneAppClient.scala:106)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.RuntimeException: java.io.EOFException
    at java.io.DataInputStream.readFully(DataInputStream.java:197)
    at java.io.DataInputStream.readUTF(DataInputStream.java:609)
    at java.io.DataInputStream.readUTF(DataInputStream.java:564)
    at org.apache.spark.rpc.netty.RequestMessage$.readRpcAddress(NettyRpcEnv.scala:593)

Тогда последующий вопрос - имеет ли это значение?Если я просто установлю мастер на local[*], получу ли я полную функциональность кластера Spark?Как мне узнать, что я использую кластер Spark для более быстрой работы?(Извините, я новичок в этом. Я пытался искать в интернете как сумасшедший, но не нашел ничего, что помогло бы мне. Если кто-то может указать мне некоторые полезные ресурсы, чтобы помочь мне, это было бы здорово!цель состоит в том, чтобы использовать кластер Spark для быстрой обработки очень больших наборов данных в Python.)

Последний дополнительный вопрос - я продолжаю получать предупреждение 19/04/30 13:19:24 WARN TaskSetManager: Stage 0 contains a task of very large size (5786 KB). The maximum recommended task size is 100 KB. На основании этого вопроса похожеМне нужно использовать метод parallelize во всех моих манипуляциях с данными.Этот метод возвращает RDD, и в настоящее время весь мой код написан для работы с объектами pyspark DataFrame.Просто проверяю - нужно ли переписывать весь мой код для манипулирования RDD, чтобы использовать кластер Spark?

Answer 1

зависит от того, откуда вы запускаете скрипт.Вы пытаетесь запустить его с вашего ноутбука или главного узла.

• запустить с главного узла: вам нужно будет установить переменную SPARK_HOME env, чтобы она указала на вашу установку spark lib.

• если вы работаете с вашего ноутбука:

• убедитесь, что SPARK_HOME указывает на правильную библиотеку.

• скопируйте все конфигииз: ~ / SPARK_HOME в ваш локальный SPARK_HOME

продолжение Q1: это сложно, потому что вы можете запускать бот-подчиненных и мастеров на одном узле.и настройте спарк для работы в конфигурации главный-подчиненный.

настройка спарка для оптимизированной производительности: хотя конфигурация спарка из коробки довольно хорошая.но чтобы оптимизировать его производительность, вам нужно специально настроить несколько свойств: 1. spark.executor.cores 2. spark.executor.memory 3. spark.driver.cores 4. spark.driver.memory

дополнительно выхотите, чтобы ваш кластер был как можно ближе друг к другу, то есть в одних и тех же группах размещения.хотя не обязательно использовать hdfs при работе с большими наборами данных и сохранять промежуточные результаты в hdfs.Вы также можете использовать менеджер ресурсов (опять же не требуется, спарк поставляется с одним), как пряжа.(это также идет с дополнительной настройкой конфигурации).

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

для последующей проверки Q2: ну, вы, как правило, не отправляете все данные от master к slave-устройствам, но salves собирают данные самостоятельно,для этого вам нужно поместить данные в x, а не в небольшие файлы, и позволить рабам читать их через read apis.

Нужно ли переписывать код для использования RDD?это зависит от версии искры, которую вы используете.датафреймы были введены в 2.0.