Enumerators, Data Streaming и другие модные слова в Ruby

Можно ли на Ruby переварить 6GB данных, используя лишь пару мегабайт памяти? Зачем нам Streaming и потоковая обработка данных? Что такое Enumerator?

Как водится, тема для этой статьи появилась не из воздуха, а из самых что ни на есть хардкорных продуктовых потрохов. Мы рассмотрим что такое потоковая обработка (streaming), для чего она нужна и как реализуется в Ruby, а также как она ложится на HTTP-протокол.

Кто к нам с чем и зачем?

Сходу начнем с задачи, чтоб обсуждать не абстрактные модели, а вполне конкретные решения:

Есть хорошее решение, которое работает: скачиваем из БД 6GB в файл, архивируем файл в другой файл, отдаём файл клиенту через send-file или загружаем в какое-нибудь хранилище через POST. Именно такое решение будет гуглиться, и оно будет норм. Но тут есть минусы: место на файловой системе, контроль и удаление файлов, обработка ошибок и всякое такое. Минусы совсем не существенные, но в современном контейнеризированном мире 12-factor-application с readonly-контейнерами или lambda/serverless функциями очень неудобно иметь файлы на диске.

Файл хорошо, а поток лучше!

Streaming data is data that is continuously generated by different sources. Such data should be processed incrementally using stream processing techniques without having access to all of the data

А как тогда сделать удобнее, чтоб всё было воздушное, stateless, не было файлов, точек монтирования, и DevOps не подключал диски к нашим контейнерам?

Не сложно - нам нужна потоковая обработка данных. Это такой способ обработки, когда данные не требуются "целиком", а требуется только кусочек, потом еще кусок и так далее. Возвращаясь к нашей задаче, streaming будет выглядеть примерно так:

• открываем соединение к БД и делаем SQL запрос;
• получаем кусочек данных (строку или несколько строк);
• передаём этот кусочек в архиватор и получаем из него кусочек архива (несколько килобайт);
• пишем эти несколько килобайт в сокет, они отправляются клиенту или передаются в POST-запросе в какое-то хранилище (S3 или WebDav или что-то подобное);
• повторяем все эти шаги до тех пор, пока БД не закончит отдавать нам данные, и мы не передадим самый последний кусочек данных куда следует;

В чём профит? Основной профит заключается в том, что в каждый отдельный момент времени обрабатывается ограниченный и маленький кусочек данных (несколько килобайт), при этом не используется даже диск. Минусы также есть, но я ведь рекламирую данный подход, поэтому про минусы промолчу 🙃

Давайте теперь рассмотрим, как это сделать просто и со вкусом. Начнём с того, что под "потоком" будем понимать объект, у которого можно "взять следующую порцию данных". В разных языках есть разные абстракции на эту тему: интерфейсы reader/writer, IO и пр. В руби самой простой абстракцией является Enumerator - это объект, у которого есть методы next или each , с помощью которых этот объект можно "перебирать". Вот в него мы и запакуем все наши данные.

Шаг 1 - получение данных (например из БД)

В рамках нашей задачи мы будем получать данные таким образом, чтоб быть наиболее приближенными к "голому" (raw) сетевому сокету, в вашем случае источник может быть другим, или данные вообще могут генерироваться в процессе, а не извлекаться из БД.

Таким образом, от БД мы сразу начнём получать CSV-документ в виде текстовых строк. Эти строки мы загоняем в Enumerator и на выходе получаем объект потока. Тут надо обратить внимание, что пока мы не позвали метод next или each , у этого объекта никакого соединения с БД не установлено, никакой SQL-запрос не отправлен, и никакие данные не поступают - это "ленивый" 😴 поток, который начнёт шевелиться, только когда мы реально попросим из него данные.

Шаг 2 - архивируем гигабайты CSV в ZIP

Отдавать клиенту 6ГБ данных в виде огромного CSV-файла как-то некрасиво, поэтому мы обернём CSV-поток в ZIP-поток. Это делается легко с использованием гема 💎 ZipTricks:

Мы создаём ZIP-файл "на лету", и как было ранее, еще никакие соединения не установлены, никакие данные не передаются 😴.

Шаг 3 - когда же наконец пойдут данные?

Тут есть несколько вариантов, мы рассмотрим два из них: отдача клиенту в виде ответа на HTTP-запрос (как обычно это происходит в Rails-приложении) и загрузка через POST-запрос в какое-нибудь хранилище.

POST-Запрос

Многие знают, что в ответ на HTTP-запрос мы можем получить данные с заголовком Transfer-Encoding: chunked , который говорит нам о том, что финальный размер данных (файла) заранее неизвестен, и клиенту (браузеру) надо принимать данные, пока принимается. Но немногие знают, что такой же механизм есть и для исходящих данных, которые отправляются на сервер, когда клиент заранее не знает, сколько он отправит на сервер. Да, так бывает, у нас как раз такой случай.

Как ни странно, chunked HTTP-upload без проблем реализуется стандартной библиотекой Net::HTTP.

Тут надо обратить внимание на Tranfser-Encoding, который и управляет потоковой загрузкой, а также на ReadableEnumerator - Net::HTTP требует, чтоб у загружаемого объекта был определённый интерфейс (метод read). И мы легко преобразуем объект нашего потока к подходящему виду через вспомогательную абстракцию:

И наконец, именно при вызове http.request все наши "ленивые" потоки начнут работу, байтики побегут 🏃 от базы данных через несколько последовательных трансформаций в хранилище через HTTP-соединение.

HTTP-ответ

Второй вариант - это просто вернуть данные в ответе, но не пачкой как обычно, а по мере их появления, также используя Transfer-Encoding: chunked, но на этот раз уже привычным способом - в ответе. Тут я поделюсь с вами тайным знанием и секретными мантрами, как сделать так, чтоб Rails не кешировал ответ, не считал ETag, контрольные суммы, чтоб промежуточные nginx/traefik тоже не кешировали ответ, и байтики бежали от Rails-приложения к клиенту сразу по мере их появления.

Посмотрев по разным нашим проектам, я нашел, что иногда при стриминге мы используем Transfer-Encoding: binary , и я не имею ни малейшего представления почему, но оно просто работает. Все эти заголовки очень важны, без них потоковая отправка работать не будет по описанным в комментариях причинам. Искушённый в Rails читатель может сказать, что существует механизм ActionController::Live, но этот механизм никогда (никогда) не работал на наших проектах - может тому виной Phusion Passenger, который мы используем, может что-то еще, тем не менее Live скорее мёртв, чем жив.

Что мы имеем в результате?

Такой подход в обработке данных мы в RNDSOFT используем уже давно и в разных проектах:

• при генерации статистики/отчётности за большие интервалы времени;
• при генерации пакетов документов пользователей;
• при архивации/выгрузке данных по требованию клиентов;

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

Люблю гуглить за людей материалы по теме:

• История появления Enumerable::Lazy - для историофилов
• Using lazy enumerators to work with large files - повторение - мать учения
• 🏆 Ruby 2.0 Works Hard So You Can Be Lazy 🏆 - не читал, но тут много красивых картинок и что-то внятно объясняется