Как читать и сохранять ограничение на уведомление для базы данных для каждого элемента Java RDD

Question

Я новичок в Apache Spark. Я пишу программу, в которой я создаю JavaRDD, получая данные из базы данных Apache Cassandra. Затем я обрабатываю каждый элемент в RDD, вызывая некоторые вспомогательные методы, которые не являются частью моей программы драйверов.

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

Так как я могу получить эти данные, а также обновить это в базе данных, когда я не в программе драйвера, и у меня нет метода sparkContext в моем вспомогательном методе?

Просьба предложить. Я использую Apache Спарк 2.0.0 и Спарк разъем Cassandra Java

Спасибо заранее

Answer 1

Это не возможно на исполнителях только, потому что они не synchroniza состояния друг с другом. Но вы можете попробовать это решение:

1. Перепишите свой вспомогательный метод для возврата dataframe со схемой (user, notification) для каждого уведомления каждого пользователя, которого вы хотите отправить
2. Выполните collect_list агрегацию на этом dataframe (схема изменится на (user, [notif1, notif2, ...])
3. Загрузка данных уже было отправлено уведомления новым dataframe, например (user, remaining_notifications_number)
4. Регистрация наборов данных из (2) и (3) и для каждого результата отправленного количество уведомлений, не превышающих предел.

Надеюсь, что это поможет!