Как сравнить два набора данных?

Я запускаю искровое приложение, которое считывает данные из нескольких таблиц hive (IP-адресов) и сравнивает каждый элемент (IP-адрес) в наборе данных со всеми другими элементами (IP-адресами) из других наборов данных. Конечным результатом будет что-то вроде:

+---------------+--------+---------------+---------------+---------+----------+--------+----------+
|     ip_address|dataset1|dataset2       |dataset3       |dataset4 |dataset5  |dataset6|      date|
+---------------+--------+---------------+---------------+---------+----------+--------+----------+
| xx.xx.xx.xx.xx|     1  |              1|              0|        0|         0|      0 |2017-11-06|
| xx.xx.xx.xx.xx|     0  |              0|              1|        0|         0|      1 |2017-11-06|
| xx.xx.xx.xx.xx|     1  |              0|              0|        0|         0|      1 |2017-11-06|
| xx.xx.xx.xx.xx|     0  |              0|              1|        0|         0|      1 |2017-11-06|
| xx.xx.xx.xx.xx|     1  |              1|              0|        1|         0|      0 |2017-11-06|
---------------------------------------------------------------------------------------------------

Для сравнения, я преобразовываю dataframes полученный в hiveContext.sql("query") в объекты Fastutil. Как это:

val df= hiveContext.sql("query")
val dfBuffer = new it.unimi.dsi.fastutil.objects.ObjectArrayList[String](df.map(r => r(0).toString).collect())

Затем я использую iterator для итерации по каждой коллекции и записи строк в файл с помощью FileWriter.

val dfIterator = dfBuffer.iterator()
while (dfIterator.hasNext){
     val p = dfIterator.next().toString
     //logic
}

Я запускаю приложение с помощью --num-executors 20 --executor-memory 16g --executor-cores 5 --driver-memory 20g

Этот процесс длится около 18-19 часов в общей сложности около 4-5 миллионов записей с сопоставлениями один к одному на ежедневной основе.

Однако, когда я проверил Application Master UI, я заметил, что никакой активности не происходит после первоначального преобразования dataframes в fastutil collection objects делаются (это занимает всего несколько минут после того, как работа будет запущена). Я вижу count и collect заявления, используемые в коде, создавая новые рабочие места, пока не будет сделано преобразование. После этого никакие новые задания не запускаются, когда сравнение выполняется.

  • Что это значит? Означает ли это, что распределенная обработка вообще не происходит?

  • Я понимаю, что объекты коллекции не рассматриваются как RDD, могут
    это может быть причиной этого?

  • Как искра выполняет мою программу без использования назначенных ресурсов?

Любая помощь будет оценена, спасибо!

Ответы

Ответ 1

После строки:

val dfBuffer = new it.unimi.dsi.fastutil.objects.ObjectArrayList[String](df.map(r => r(0).toString).collect())

особенно эта часть вышеприведенной строки:

df.map(r => r(0).toString).collect()

которые collect - это самое главное, чтобы заметить, никакие задания Spark никогда не выполняются на dfBuffer (который является обычной локальной структурой данных JVM).

Означает ли это, что распределенная обработка вообще не происходит?

Правильный. collect все данные на одном JVM, где работает драйвер (и именно поэтому вы не должны этого делать, если... вы не знаете, что делаете и какие проблемы могут возникнуть).

Я думаю, что выше сказанное отвечает на все остальные вопросы.

Возможным решением вашей проблемы сравнения двух наборов данных (в Spark и распределенной моде) было бы join набора данных с эталонным набором данных и count чтобы сравнить, не изменилось ли количество записей.