Я запускаю приложение spark grep на своем 1-узловом кластере искр. (Сам хозяин и раб). У меня есть текстовый файл 913 МБ на удаленном источнике на сервере hdfs. Текстовый файл разделен на куски размером 128 МБ с репликацией, установленной на ноль. Мое приложение spark читает текстовый файл и применяет преобразование фильтра для поиска ключевого слова, а затем при выводе в hdfs выводит только строки с этим ключевым словом. Чтобы получить представление о работе spark, я зарегистрировал больше деталей в исходном коде spark и собрал его для запуска модифицированной версии. Я зарегистрировал информацию о считанных байтах, используя reader.getpos () в getNext (K, V) в hadoopRDD.scala до 10 записей. Я также регистрирую некоторые сообщения в операции фильтра, чтобы получить, когда вычисления фактически начинаются. Поскольку spark использует ленивую загрузку, фильтр входа в систему показывает, прежде чем будут прочитаны какие-либо байты данных. Поэтому я попытался прочитать полный итератор перед операцией фильтрации, используя:
while(iter.hasNext){
logInfo(s"value of the iterator: ${iter.next}")
}
Приведенный выше код регистрирует следующую строку в текстовом файле каждый раз, когда читается запись (запись здесь будет строкой в текстовом файле в качестве операции записи по умолчанию для чтения хадупа). Но после этого не выполняет никакой фильтрации, и я получаю пустой вывод. Если я просто распечатываю журнал после операции фильтрации, то в результате он появляется только после того, как все записи прочитаны, и программа чтения хопов закрыта.
Я пытаюсь разделить вычисления и сетевой ввод / вывод, по крайней мере, для первого набора байтов, чтобы показать узкое место в производительности, но я не уверен, действительно ли спарк выполняет вычисления для каждой записи или один раз для всех записей читать для задачи? и как мне вычислить это отдельно?