Почему операция на аккумуляторе не работает без collect ()?

Question

Я изучаю Spark и следую инструкциям.В упражнении я пытаюсь провести некоторый анализ набора данных.Этот набор данных содержит данные в каждой строке, например:

userid | age | gender | ...

У меня есть следующий фрагмент кода:

....

under_age = sc.accumulator(0)
over_age = sc.accumulator(0)


def count_outliers(data):
    global under_age, over_age
    if data[1] == '0-10':
        under_age += 1
    if data[1] == '80+':
        over_age += 1
    return data


data_set.map(count_outliers).collect()
print('Kids: {}, Seniors: {}'.format(under_age, over_age))

Я обнаружил, что должен использовать метод ".collect ()"чтобы этот код работал.То есть без вызова этого метода код не будет считать два аккумулятора.Но в моем понимании «.collect ()» используется, чтобы получить весь набор данных в память.Зачем здесь это нужно?Это связано с ленивой оценкой?Пожалуйста, сообщите.

Answer 1

Да, это из-за ленивых вычислений.

Spark ничего не вычисляет, пока вы не выполните действие , такое как collect, и аккумуляторы не будут обновлены только как сторона-эффект этого вычисления.

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

Это описано в документации :

Аккумуляторы не изменяют ленивую оценочную модель Spark.Если они обновляются в операции над RDD, их значение обновляется только после того, как RDD вычисляется как часть действия.Следовательно, не гарантируется, что обновления аккумулятора будут выполняться при выполнении ленивого преобразования, например map ().

Также важно отметить, что:

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

, поэтому ваши аккумуляторы не обязательно будут давать правильные ответы;они могут завышать итоги.