Что такое Spark Job?

Ответ 1

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

Тем не менее, у Spark есть свое определение "работа", непосредственно из глоссария:

Работа. Параллельное вычисление, состоящее из нескольких задач, которые получают порождается в ответ на действие Spark (например, save, collect); вот увидишь этот термин используется в журналах драйверов.

Итак, в этом контексте, скажем, вам нужно сделать следующее:

• Загрузите файл с именами и адресами людей в RDD1
• Загрузите файл с именами людей и телефонами в RDD2
• Присоедините RDD1 и RDD2 по имени, чтобы получить RDD3
• Карта на RDD3, чтобы получить хорошую презентационную карточку HTML для каждого человека как RDD4
• Сохранить RDD4 в файл.
• Карта RDD1 для извлечения zipcodes из адресов для получения RDD5
• Совокупность на RDD5 для подсчета количества людей, живущих на каждом zipcode, как RDD6
• Соберите RDD6 и распечатайте эти данные в стандартный вывод.

Итак,

• Программа драйвера - это весь фрагмент кода, выполняющий все 8 шагов.
• Создание всей HTML-карты, установленной на шаге 5, - это задание (ясно, потому что мы используем действие save, а не преобразование). То же самое со сборкой на шаге 8
• Другие шаги будут организованы в этапы, причем каждое задание будет результатом последовательности этапов. Для простых вещей задание может иметь один этап, но необходимость перераспределения данных (например, соединение на шаге 3) или что-либо, что нарушает локальность данных, обычно приводит к появлению большего количества этапов. Вы можете рассматривать этапы как вычисления, которые производят промежуточные результаты, которые на самом деле могут сохраняться. Например, мы можем сохранить RDD1, так как мы будем использовать его более одного раза, избегая перерасчета.
• Все 3 выше в основном говорят о том, как будет нарушена логика данного алгоритма. Напротив, задача представляет собой конкретный фрагмент данных, которые пройдут через данный этап, на данном исполнителе.

Надеюсь, что это станет проще, -)

Ответ 2

Эй, вот что я сделал раньше, надеюсь, что это сработает для вас:

#!/bin/bash
# Hadoop and Server Variables
HADOOP="hadoop fs"
HDFS_HOME="hdfs://ha-edge-group/user/max"
LOCAL_HOME="/home/max"

# Cluster Variables
DRIVER_MEM="10G"
EXECUTOR_MEM="10G"
CORES="5"
EXECUTORS="15"

# Script Arguments
SCRIPT="availability_report.py" # Arg[0]
APPNAME="Availability Report" # arg[1]

DAY=`date -d yesterday +%Y%m%d`

for HOUR in 00 01 02 03 04 05 06 07 08 09 10 11 12 13 14 15 16 17 18 19 20 21 22 23
do
        #local directory to getmerge to
        LOCAL_OUTFILE="$LOCAL_HOME/availability_report/data/$DAY/$HOUR.txt"

        # Script arguments
        HDFS_SOURCE="webhdfs://1.2.3.4:0000/data/lbs_ndc/raw_$DAY'_'$HOUR" # arg[2]
        HDFS_CELLS="webhdfs://1.2.3.4:0000/data/cells/CELLID_$DAY.txt" # arg[3]
        HDFS_OUT_DIR="$HDFS_HOME/availability/$DAY/$HOUR" # arg[4]

        spark-submit \
        --master yarn-cluster \
        --driver-memory $DRIVER_MEM \
        --executor-memory $EXECUTOR_MEM \
        --executor-cores $CORES \
        --num-executors $EXECUTORS \
        --conf spark.scheduler.mode=FAIR \
        $SCRIPT $APPNAME $HDFS_SOURCE $HDFS_CELLS $HDFS_OUT_DIR

        $HADOOP -getmerge $HDFS_OUT_DIR $LOCAL_OUTFILE
done