Spark кластер на EC2 использует только один узел

Question

Я использую flintrock для запуска кластера Spark с 8 + 1 узлами на Amazon EC2.

> flintrock --config config.yaml launch cluster-8nodes

Затем я использую flintrock для входа в кластер:

> flintrock --config config.yaml login cluster-8nodes

Работа, которую я выполняю, - это, по сути, простой код подсчета биграмм для большого текстового файла:

@contextmanager
def use_spark_context(appName):
    conf = SparkConf().setAppName(appName) 
    spark_context = SparkContext(conf=conf)

    try:
        print("starting ", appName)
        yield spark_context
    finally:
        spark_context.stop()
        print("stopping ", appName)

with use_spark_context("AppName") as spark:
    text_file = spark.textFile(text_path)
    bigrams = text_file.flatMap(lambda line: line.split(".")) \
                       .map(lambda line: line.strip().split(" ")) \
                       .flatMap(lambda xs: (tuple(x) for x in zip(xs, xs[1:])))
    counts = bigrams.map(lambda bigram: (bigram, 1)) \
            .reduceByKey(lambda x, y: x + y) \
            .filter(lambda bigram: bigram in name_bigrams) \
            .collect()

Он сохраняется в файле .py и передается следующим образом после входа в систему через flintrock:

> PYSPARK_PYTHON=python3 spark-submit --num-executors 8 my_job.py --input data/bigtext.txt

Программа работает нормально и выдает следующий вывод.Однако все узлы, кроме одного, не работают.Разве эта установка не должна распределять задание по 8 узлам кластера?

18/06/08 09:50:48 INFO Executor: Finished task 10.0 in stage 0.0 (TID 10). 1998 bytes result sent to driver
18/06/08 09:50:48 INFO TaskSetManager: Starting task 12.0 in stage 0.0 (TID 12, localhost, executor driver, partition 12, PROCESS_LOCAL, 4851 bytes)
18/06/08 09:50:48 INFO Executor: Running task 12.0 in stage 0.0 (TID 12)
18/06/08 09:50:48 INFO TaskSetManager: Finished task 10.0 in stage 0.0 (TID 10) in 30285 ms on localhost (executor driver) (11/382)
18/06/08 09:50:48 INFO HadoopRDD: Input split: file:/home/ec2-user/data/enwiki-extract.txt:402653184+33554432
18/06/08 09:50:53 INFO PythonRunner: Times: total = 32160, boot = -586, init = 588, finish = 32158
18/06/08 09:50:54 INFO Executor: Finished task 11.0 in stage 0.0 (TID 11). 1998 bytes result sent to driver
18/06/08 09:50:54 INFO TaskSetManager: Starting task 13.0 in stage 0.0 (TID 13, localhost, executor driver, partition 13, PROCESS_LOCAL, 4851 bytes)
18/06/08 09:50:54 INFO TaskSetManager: Finished task 11.0 in stage 0.0 (TID 11) in 32785 ms on localhost (executor driver) (12/382)
18/06/08 09:50:54 INFO Executor: Running task 13.0 in stage 0.0 (TID 13)
18/06/08 09:50:54 INFO HadoopRDD: Input split: file:/home/ec2-user/data/enwiki-extract.txt:436207616+33554432
18/06/08 09:51:19 INFO PythonRunner: Times: total = 30232, boot = -571, init = 578, finish = 30225
18/06/08 09:51:19 INFO Executor: Finished task 12.0 in stage 0.0 (TID 12). 1998 bytes result sent to driver
18/06/08 09:51:19 INFO TaskSetManager: Starting task 14.0 in stage 0.0 (TID 14, localhost, executor driver, partition 14, PROCESS_LOCAL, 4851 bytes)
18/06/08 09:51:19 INFO Executor: Running task 14.0 in stage 0.0 (TID 14)
18/06/08 09:51:19 INFO TaskSetManager: Finished task 12.0 in stage 0.0 (TID 12) in 30794 ms on localhost (executor driver) (13/382)
18/06/08 09:51:19 INFO HadoopRDD: Input split: file:/home/ec2-user/data/enwiki-extract.txt:469762048+33554432
18/06/08 09:51:25 INFO PythonRunner: Times: total = 31385, boot = -608, init = 611, finish = 31382
18/06/08 09:51:26 INFO Executor: Finished task 13.0 in stage 0.0 (TID 13). 1998 bytes result sent to driver
18/06/08 09:51:26 INFO TaskSetManager: Starting task 15.0 in stage 0.0 (TID 15, localhost, executor driver, partition 15, PROCESS_LOCAL, 4851 bytes)
18/06/08 09:51:26 INFO TaskSetManager: Finished task 13.0 in stage 0.0 (TID 13) in 32061 ms on localhost (executor driver) (14/382)
18/06/08 09:51:26 INFO Executor: Running task 15.0 in stage 0.0 (TID 15)
18/06/08 09:51:26 INFO HadoopRDD: Input split: file:/home/ec2-user/data/enwiki-extract.txt:503316480+33554432

РЕДАКТИРОВАТЬ: если я задаю основной URL-адрес как вывод от flintrock launch до spark-submit --master, задание запускается, но не выполняется, потому чтовходной файл, который хранится локально на узле входа в систему, не найден:

py4j.protocol.Py4JJavaError: An error occurred while calling z:org.apache.spark.api.python.PythonRDD.collectAndServe.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 3 in stage 0.0 failed 4 times, most recent failure: Lost task 3.3 in stage 0.0 (TID 30, 172.31.28.28, executor 5): java.io.FileNo$
FoundException: File file:/home/ec2-user/data/enwiki-extract.txt does not exist

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

Answer 1

По умолчанию spark-submit запускает Spark в локальном режиме.То, что работало, указывало мастер через --master spark://<masterURL>:7077 и устанавливало --num-executors как минимум на количество рабочих узлов, в зависимости от конфигурации кластера.

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