Распределенная система для потоковой обработки спутниковой информации

Центр мониторинга социально-экономических процессов и природной среды ИВТ СО РАН осуществляет оперативную обработку спутниковых снимков с 2009 года. Для этой цели нами был создан специализированный программно-аппаратный комплекс, обеспечивающий весь цикл обработки от получения данных из приемного центра до предоставления их конечным пользователям. В данной работе описывается реализация программной части, обеспечивающей генерацию стандартных продуктов MODIS.
Получение стандартных продуктов MODIS заключается в запуске цепочек программ, предоставляемых NASA. Обычным способом для автоматизации этого процесса является использование традиционных скриптовых языков, таких как Bash или Perl. В частности, именно так реализован международный пакет для обработки снимков IMAPP. Этот подход позволяет максимально быстро развернуть обработку. Все используемые средства являются стандартными, что упрощает работу администратору системы. В то же время, с развитием нашего комплекса мы выявили ряд проблем.
Во-первых, последовательный запуск обработчиков приводит к неэффективному использованию современных многопроцессорных серверов. Распараллеливание самих алгоритмов является очень трудоемким в силу большого размера кодовой базы, а параллельный запуск различных модулей плохо выражается в скриптовых языках. Во-вторых, со временем от системы потребовались дополнительные возможности, такие как: распределенное выполнение, обработка по заказу, управление расположением продуктов. Система, основанная на скриптах, оказалась недостаточно гибкой для добавления этих возможностей. Кроме того, система была монолитной, её было трудно поддерживать.
Для преодоления этих трудностей мы решили перевести систему управления запуском обработчиков на средства более высокого уровня абстракции. Сценарий обработки описывается в виде графа потока данных. В результате возможна его интерпретация без потери параллелизма между модулями. Это позволяет добиться эффективного использования многопроцессорных систем.
В качестве базы для нового комплекса управления была выбрана система Taverna. Этот выбор обусловлен с одной стороны моделью вычислений, хорошо подходящей для описания процесса обработки спутниковых снимков, а с другой стороны богатыми возможностями для расширения её функционала через подключаемые модули.
Алгоритм в Taverna представляется в виде набора «процессоров» - черных ящиков с некоторым количеством входных и выходных портов. Порты процессоров связаны друг с другом и образуют ациклический граф. Этот граф описывает передачу параметров между процессорами и определяет ограничения на порядок запуска процессоров. Любой порт может быть определен как одиночный или списковый. При этом нет ограничений на связи портов друг с другом. Передача списка на одиночный порт приводит к выполнению процессора для каждого элемента списка, причем отдельные итерации могут быть произведены параллельно. И наоборот, при связи одиночного порта со списковым будет произведена синхронизация всех итераций порождающего процессора, значения на выходных портах будут объединены в список и переданы дочернему процессору. В этих терминах удобно выражать такие операции как нарезка исходного снимка на гранулы и сбор всех снимков за день для получения агрегатного продукта.
Для того, чтобы воспользоваться этими возможностями для Taverna был написан плагин, позволяющий использовать в качестве её процессоров Bash-скрипты (https://github.com/kikht/ict-taverna-modules/wiki/Bash-activity-plugin). Каждый из них отвечает за подготовку параметров и запуск одной программы-обработчика. Значения входных портов подставляются в качестве переменных окружения, а значения выходных портов также считываются из состояния переменных скрипта на момент завершения. Плагин обеспечивает абстракцию от того где будет исполнен модуль. В настоящее время поддерживаются выполнение на локальной машине, удаленной с доступом по SSH, а также системы управления очередями кластера Torque и Slurm. Единственным требованием является наличие общей файловой системы для всех узлов, участвующих в обработке.
Существенным аспектом системы обработки является то, что сами снимки имеют довольно большой размер, вплоть до нескольких гигабайт. Внутри Taverna мы оперируем лишь ссылками на них, и вся нагрузка, связанная с передачей данных от одного модуля на другой, ложится на файловую систему. В то же время общий объем данных центра не позволяет хранить продукты только на быстрых локальных дисках, неизбежно приходится использовать сетевые системы хранения данных.
Испытания новой версии комплекса показали, что на 16-ядерном сервере при использовании локальной дисковой файловой системы он обеспечивает ускорение дневной обработки до 5 раз по сравнению с полностью последовательной версией. При использовании же сетевой СХД время обработки увеличилось в 1.5 раза и наблюдалась стагнация распараллеливания: увеличение числа активных потоков более 12 не давало прироста производительности. Работа в распределенном режиме на кластере из нескольких машин давала еще худшие результаты.
Проведенный анализ производительности показал, что виной такому поведению является характер доступа к данным снимка при обработке. Обращения к файлам происходят в основном случайным образом. Это приводит к деградации производительности на сетевых файловых системах. Локальные же файловые системы используют агрессивный кеш в оперативной памяти, что сглаживает случайный характер доступа к данным.
Для преодоления неэффективности сетевых ФС мы действуем следующим образом. Модули обработки запускаются на локальном диске, а затем результаты асинхронно копируются на сетевое хранилище. При запуске каждого модуля, мы стараемся выбрать узел для него так, чтобы максимальное количество данных лежало у него на локальном диске. Таким образом достигаются сразу несколько положительных эффектов. Во-первых, продукты копируются на СХД последовательно и целиком. Это наилучший режим работы для практически любой системы хранения. Во-вторых, по возможности, чтение данных производится с быстрого локального диска. И в-третьих, на локальном диске нужно хранить только данные участвующие в текущем процессе обработки, а не весь архив данных. Все вместе это позволяет нам обрабатывать объемы данных возможные только для СХД с производительностью локальных файловых систем.
С учетом этой оптимизации производительность распределенной системы на небольших наборах данных практически сравнялась с производительностью локальной версии. А при увеличении объемов обработки (например, пересчет всех продуктов за предыдущий месяц), производительность системы ограничивается скоростью последовательной записи на внешнюю систему хранения.

Работа выполняется при поддержке проекта IV.31.2.1. Программы фундаментальных исследований СО РАН на 2010 - 2012 гг., Российского фонда фундаментальных исследований (гранты 11-07-12048-офи-м-2011, 12-07-00545-а); Программы интеграционных фундаментальных исследований Президиума СО РАН (междисциплинарные проекты No. 131, Программы поддержки ведущих научных школ (грант НШ-931.2008.9).
Список литературы


Ю.И. Шокин, Н.Н. Добрецов, В.В. Смирнов, А.А. Лагутин, В.Н. Антонов, A.В. Калашников Система информационной поддержки задач оперативного мониторинга на основе данных дистанционного зондирования // Тезисы докладов Восьмой открытой Всероссийской конференции "Современные проблемы дистанционного зондирования земли из космоса" (Москва, 15 - 19 ноября 2010 г.). М.: ИКИ РАН, 2010. – С. 40-41.
Duncan Hull и др. Taverna: a tool for building and running workflows of services. // Nucleic Acids Research, vol. 34, 2006.