Spark: Могу ли я использовать один и тот же аккумулятор для разных преобразований? [Дубликат]

Question

Spark: Могу ли я использовать один и тот же аккумулятор для разных преобразований? [Дубликат]

Давайте посмотрим на лес сначала, прежде чем смотреть на деревья.

Здесь есть много информативных ответов с большими подробностями, я не буду повторять ни одного из них. Ключ к программированию в JavaScript имеет сначала правильную ментальную модель общего исполнения.

Ваша точка входа (ов) выполняется в результате события. Например, в браузер загружается тег сценария с кодом. (Соответственно, поэтому вам, возможно, придется заботиться о готовности страницы запускать ваш код, если он требует, чтобы элементы dom были сконструированы первыми и т. Д.)
Ваш код выполняется до завершения, однако многие асинхронные вызовы, которые он делает, без выполнения каких-либо ваших обратных вызовов, включая запросы XHR, установку тайм-аутов, обработчиков событий dom и т. д. Каждый из этих обратных вызовов, ожидающих выполнения, будет находиться в очереди, ожидая, что их очередь будет запущена после других событий
Каждый отдельный обратный вызов XHR-запроса, установленного таймаута или dom события после вызова будет завершен.

Хорошие новости заключается в том, что, если вы хорошо понимаете этот момент, вам никогда не придется беспокоиться о гоночных условиях. Прежде всего вы должны понимать, как вы хотите упорядочить свой код как по существу ответ на разные дискретные события, и как вы хотите объединить их в логическую последовательность. Вы можете использовать обещания или новые асинхронные / ожидающие более высокие уровни в качестве инструментов для этой цели, или вы можете откатывать свои собственные.

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

40

java apache-spark accumulator

задан Patsy Issa 15 April 2015 в 15:42

3 ответа

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

Я предпочитаю полагаться на reduce и aggregate вместо

Действие запускает задачи.

Если действие зависит от более раннего этапа, а результаты

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

g5] Тем не менее, существует множество простых случаев, когда аккумуляторы могут полностью доверять.

val acc = sc.accumulator(0)
val rdd = sc.parallelize(1 to 10, 2)
val accumulating = rdd.map { x => acc += 1; x }
accumulating.count
assert(acc == 10)

Будет ли это гарантировано правильным (не имеют дубликатов)?

Да, если спекулятивное выполнение отключено. map и count будут на одном этапе, так что, как вы говорите, нет возможности успешно выполнить задачу более одного раза.

Но аккумулятор обновляется как боковое, эффект. Поэтому вы должны быть очень осторожны, думая о том, как будет выполняться код. Рассмотрим это вместо accumulating.count:
// Same setup as before.
accumulating.mapPartitions(p => Iterator(p.next)).collect
assert(acc == 2)
Это также создаст одну задачу для каждого раздела, и каждая задача будет гарантирована выполнить ровно один раз. Но код в map не будет выполнен для всех элементов, только первый в каждом разделе.

Аккумулятор похож на глобальную переменную. Если вы разделяете ссылку на RDD, которая может наращивать аккумулятор, тогда другой код (другие потоки) может также привести к его увеличению.
// Same setup as before.
val x = new X(accumulating) // We don't know what X does.
                            // It may trigger the calculation
                            // any number of times.
accumulating.count
assert(acc >= 10)

16

ответ дан Daniel Darabos 27 August 2018 в 08:46

1

– Daniel Langdon 17 April 2015 в 12:53
2

– Daniel Darabos 17 April 2015 в 13:58
3

– Daniel Langdon 17 April 2015 в 15:03
4

– Justin Pihony 18 April 2015 в 20:22
5

– Daniel Darabos 27 April 2015 в 20:08

Я думаю, что Матей ответил на это в указанной документации:

Как обсуждалось в https://github.com/apache/spark/pull/2524 , это довольно сложно обеспечить хорошую семантику в общем случае (обновления аккумулятора внутри этапов без результата) по следующим причинам:

RDD может быть вычислен как часть нескольких этапов. Например, если вы обновите аккумулятор внутри MappedRDD и затем перетасовываете его, это может быть один этап. Но если вы затем снова назовете map () на MappedRDD и перетасовываете результат, вы получаете второй этап, на котором эта карта является конвейером. Вы хотите пересчитать это обновление аккумулятора дважды или нет?

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

Итак, я собираюсь отметить это как «не будет исправлять» на данный момент, за исключением части для этапов результата, выполненных в SPARK-3628.

1

ответ дан Justin Pihony 27 August 2018 в 08:46

1

– Daniel Langdon 7 April 2015 в 16:42

Другие вопросы по тегам:

java apache-spark accumulator

Похожие вопросы:

score 17 · Accepted Answer

Чтобы ответить на вопрос «Когда аккумуляторы действительно надежны?»

Ответ: Когда они присутствуют в действии.

В соответствии с документацией в Action Task, даже если все перезапущенные задачи присутствуют, он будет обновлять Accumulator только один раз.

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

И действие разрешает запуск пользовательского кода .

Для примера.
val accNotEmpty = sc.accumulator(0)
ip.foreach(x=>{
  if(x!=""){
    accNotEmpty += 1
  }
})
Но почему Map + Action, а именно. Операции задания результата не являются надежными для операции Аккумулятора?

Задача завершилась неудачно из-за некоторого исключения в коде. Spark попробует 4 раза (количество попыток по умолчанию). Если задача терпит неудачу каждый раз, когда она даст исключение. Если случайно это удастся, то Spark продолжит работу и просто обновит значение аккумулятора для успешного состояния, а состояния накопившихся состояний игнорируются. Вердикт : Правильно обработано

Неисправность этапа: если узел исполнителя вылетает из строя, нет ошибки пользователя, но аппаратный сбой - И если узел опускается в режиме тасования. Поскольку выход в случайном порядке хранится локально, если узел опускается , что выход в случайном порядке отсутствует. Так что Spark возвращается на сцену, которая генерирует вывод в случайном порядке, ищет, какие задачи нужно перезапустить, и выполняет их на одном из узлов, которые все еще живы. После того как мы восстановим отсутствующий вывод в случайном порядке, этап, который сгенерировал вывод карты, выполнял некоторые из его задач несколько раз.Spark подсчитывает обновления аккумулятора от всех них. Вердикт: не обрабатывается в Задаче результата. Аккумулятор выдаст неправильный вывод.

Если задание выполнено медленная работа, Spark может запустить specul аверсивная копия этой задачи на другом узле. Вердикт: не обрабатывается.Accumulator выдаст неверный результат.

RDD, который кэшируется, огромен и не может находиться в памяти. Поэтому всякий раз, когда используется RDD, запустите операцию «Карта», чтобы получить RDD, и снова аккумулятор будет обновлен им. Вердикт: не обрабатывается. Счетчик выдаст неправильный вывод.

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

Так что лучше использовать Accumulator in Action в Spark.

Чтобы узнать больше об Аккумуляторе и его проблемах относятся к этой статье в блоге - Имраном Рашидом.