Спарк foreachpartition улучшения связи

Question

Я написал искровой работу, которая делает ниже операций

1. Читает данные из HDFS текстовых файлов.
2. Сделайте отдельный() вызов для фильтрации дубликатов.
3. ли фазы mapToPair и генерировать pairRDD
4. Выполните reducebykey вызов
5. сделать агрегация логики для сгруппированных кортежа.

6. теперь называют Еогеасп на # 5

   здесь это

   1. сделать вызов Кассандры дб
   2. создать AWS SNS и SQS соединение клиента
   3. сделать некоторые JSON записи форматирования.
   4. опубликовать запись в SNS/SQS

когда я запускаю эту работу, он создает три искры этапы

первый этап - это занимает около 45 сек. выполняет отчетливую второго этапа - mapToPair и reducebykey = занимает 1,5 мин

третьей ступени = занимает 19 минут

что я сделал

1. Я выключил Кассандру вызов так увидеть DB ударила причину - это принимая меньше времени
2. Оскорблять часть я нашел, чтобы создать SNS/SQS соединения Foreach секционирования

его принятие более чем 60% всего рабочего времени

Я создаю соединение SNS/SQS внутри foreachPartition, чтобы улучшить меньше соединений. у нас есть еще лучший способ

Я не могу создать объект подключения на водителя, как это не сериализации

Я не использую числа исполнителя 9, executore сердечник 15, 2g памяти драйвера, Вершитель память 5G

Я использую 16 сердечник 64 гиг память размера кластера 1 мастер 9 раба все же конфигурацию ЭХ искр развертывания 1,6

Answer 1

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

Я думаю, что foreachPartition - это правильная идея здесь, но вы можете захотеть объединить ваш RDD заранее. Это приведет к срыву разделов на одном узле без перетасовки и позволит вам не запускать дополнительные соединения SNS/SQS.

Смотрите здесь: https://spark.apache.org/docs/latest/api/scala/index.html#org.apache.spark.rdd.RDD@coalesce(numPartitions:Int,shuffle:Boolean,partitionCoalescer:Option[org.apache.spark.rdd.PartitionCoalescer])(implicitord:Ordering[T]):org.apache.spark.rdd.RDD[T]