Пам'ять як вузьке місце в ETL: Dask, Polars, Pandas

Пам'ять як критичний ресурс

Пам'ять стала критичним ресурсом. Зростання попиту на пам'ять та інфраструктуру зберігання, спричинене бумом ШІ, досягло історичних максимумів. Компанії, такі як Micron Technology та Sandisk, привернули безпрецедентну увагу та підвищили ціни на продукцію, використовуючи сильну цінову владу. Це не є доброю новиною для компаній, які створюють інтенсивні до даних застосунки, покладаються на високоємні сховища для навчання ШІ, впроваджують великомасштабну аналітику або працюють з низькою маржею в хмарних сервісах. Для інженерів даних це не просто ринкові новини, а щоденне обмеження. Коли RAM та флеш-пам'ять стають все дорожчими, старий рефлекс «додати більше потужності» більше не працює. Бюджети не масштабуються з обсягом даних, а рахунки за хмарні послуги перебувають під пильним контролем.

Виклик: Набір даних у 6.2 мільйона рядків, що не вміщається в пам'ять

Проблема починається з нового ETL-конвеєра. Вихідні дані — це 6.2 мільйона постів із соціальної медіа-платформи. Набір даних, витягнутий з API соціальних мереж, після JSON-сплющення перетворюється на понад 200 стовпців. Найбільш проблематична частина: велика кількість полів даних має змішані типи. Наприклад, reaction_count може бути числом, рядком або null, а hashtags — списком рядків, одним рядком або null.

Оскільки PySpark вимагає послідовної схеми, а схема API соціальних мереж час від часу змінюється, для обробки стовпців змішаних типів розглядається використання Pandas. На відміну від PySpark, Pandas за замовчуванням зберігає ці стовпці як object і не вимагає, щоб кожен рядок відповідав одній і тій же схемі. Проста функція для нормалізації змішаних стовпців шляхом перетворення їх на рядки призвела до припинення виконання через перевищення використання пам'яті. Завдання завершилося невдачею.

Вузьким місцем є 6.2 мільйона рядків. Розмір набору даних становить приблизно 30 ГБ, що перевищує стандартну пам'ять хмарного воркера. Це перевищує доступну пам'ять воркера під час проміжних перетворень DataFrame.

Класичне рішення: Зменшення пікового споживання пам'яті за допомогою чанкування Pandas

Замість виконання перетворень типів даних для всього стовпця, що змушує Pandas матеріалізувати великі тимчасові об'єкти в пам'яті, техніка чанкування ділить кожен стовпець на частини. У цьому випадку підходить розмір чанку 250 000. Таким чином, Pandas потрібно обробляти лише 250 тисяч рядків за один раз, потім звільняти пам'ять і переходити до наступного чанку. Після того, як пікове споживання пам'яті значно зменшується, перетворення даних успішно завершується, і конвеєр стає стабільним. Однак час виконання стає значно довшим. Це не дивно, оскільки техніка чанкування — це компроміс: вона обмінює швидкість виконання на надійність конвеєра.

Від ручного чанкування до автоматизованого паралельного виконання: Dask

Крім ручного чанкування в Pandas, Dask автоматично розділяє DataFrame на кілька менших розділів і вирішує проблему збоїв пам'яті під час перетворення даних. Однак його внутрішня механіка виконання відрізняється від чанкування. Коли в Pandas встановлюється chunk_size, він читає один чанк, виконує код на ньому, видаляє його з RAM, а потім переходить до наступного. Він використовує одне ядро CPU за раз, тому для хмарних сервісів, які надають кілька ядер CPU, він не використовує всі можливості. Крім того, доводиться вручну писати цикл для агрегації результатів, що робить код складним і довгим.

Dask автоматично ділить набір даних на чанки. Dask створює граф завдань і планує розділи на доступних ядрах CPU, що значно скорочує час виконання. Однак не можна ігнорувати проблему змішаних типів даних у Dask. Оскільки Dask DataFrame складається з кількох розділів Pandas DataFrame, при читанні CSV або JSON файлів Dask виводить типи даних із зразка даних. Якщо стовпець містить непослідовні значення (наприклад, порожні рядки, None, цілі числа та рядки), Dask, ймовірно, видасть ValueError, TypeError або помилку виведення метаданих. Це відбувається тому, що Dask виводить тип даних стовпця з початкового зразка даних і припускає, що стовпець є цілим числом. Але якщо він потім зустрічає рядок в одному з наступних чанків, Dask видає помилку. Щоб вирішити цю проблему, необхідно явно вказати очікувані типи даних стовпців замість того, щоб покладатися на автоматичне виведення.

Dask не настільки гнучкий, як чанкування Pandas, при обробці динамічних стовпців зі змішаними типами даних, оскільки він вимагає вказати, які стовпці перетворювати. Крім того, він все ще виконує багато операцій Pandas у кожному розділі, тому робочі навантаження, де домінують стовпці об'єктів Python, можуть залишатися інтенсивними до пам'яті та нешвидкими при обробці наборів даних з мільйонами рядків.

Потужніша альтернатива: Polars

Polars — це бібліотека DataFrame, реалізована на двигуні Rust. Порівняно з Python, Rust виробляє високооптимізований нативний машинний код і пропонує відмінне управління пам'яттю. Він мінімізує виділення пам'яті та усуває накладні витрати на збирання сміття. Polars був запущений у 2020 році та розроблений для обробки масивних наборів даних значно швидше, ніж нативний Python Pandas. Він зберігає переваги двигуна Rust, але дозволяє користувачам Python імпортувати його як бібліотеку Python.

Polars використовує формат стовпчастих даних Apache Arrow in-memory, який розроблений для мінімізації копіювання пам'яті та максимізації ефективності кешу CPU. Він виконує операцію .cast(pl.String) безпосередньо в коді Rust. Ці функції дозволяють Polars працювати в кілька разів швидше, ніж Python, і використовувати лише частину пам'яті. Polars створює лінивий план запитів, і оптимізатор запитів визначає найефективніший план виконання перед читанням даних. Ці механізми зменшують непотрібне використання пам'яті. Тому, при обробці наборів даних, що перевищують доступну пам'ять, Polars може обробляти дані в потоковому режимі та запобігає завантаженню всього набору даних в RAM одночасно.

Polars перевершує попередні два рішення в управлінні пам'яттю, оскільки він розроблений з нуля для більш ефективного використання пам'яті, тоді як Pandas Chunking та Dask зосереджені на зменшенні навантаження на пам'ять під час виконання. Однак Polars не є універсальним рішенням. Він має свої недоліки. По-перше, Polars вводить власний API DataFrame, хоча і використовує Python. Загальні операції Pandas, такі як apply(), індексування та синтаксис groupby, часто потребують переписування. По-друге, багато сторонніх бібліотек все ще побудовані навколо Pandas, тому інтеграція з Polars все ще вимагає перетворення між форматами DataFrame. Подібно до Dask, необхідно вирішувати проблеми, спричинені змішаними типами даних, та забезпечувати послідовність схеми.

Висновок: Вибір правильного інструменту

Чи застаріло чанкування Pandas? Не обов'язково. Кожен із трьох підходів вирішує різну проблему. Найкращий вибір залежить від обмежень, а не від новітніх технологій. Якщо у вас дійсно обмежені обчислювальні ресурси та ви обробляєте динамічні схеми, чанкування Pandas все ще є відмінним рішенням. Воно різко зменшує пікове використання пам'яті. Компроміс — це час виконання. Але в багатьох виробничих середовищах повільніший, але стабільніший конвеєр набагато цінніший, ніж швидший, який постійно виходить з ладу.

Якщо ваше робоче навантаження вже переросло одну машину і ви хочете використовувати кілька ядер CPU, Dask є кращим варіантом. Він автоматизує розділення та паралельне виконання. Однак слід звернути увагу на послідовність схеми та типи даних, особливо при роботі з напівструктурованими даними.

Polars обирається, коли робоче навантаження є критичним до продуктивності, і ви освоїли новий API DataFrame. Polars зазвичай вважається найсильнішим варіантом завдяки своєму двигуну Rust, формату пам'яті Apache Arrow та оптимізатору запитів. Всі ці функції дозволяють Polars обробляти великі набори даних зі значно меншим споживанням пам'яті та набагато вищою продуктивністю.

На завершення, оптимізація пам'яті — це не пошук єдиного найкращого рішення. Натомість, це розуміння обмежень вашого проєкту та вибір правильного інструменту. В епоху ШІ здатність оптимізувати конвеєри даних в умовах обмежень пам'яті стає цінною навичкою для інженерів даних. Надійний ETL-конвеєр вимагає більше, ніж ефективність пам'яті. Він також залежить від тестування, підтримки та надійності розгортання.

Що це означає для розробників

Для інженерів даних це означає необхідність переосмислення підходів до обробки даних, оскільки просте додавання пам'яті більше не є ефективним. Вони повинні освоїти техніки оптимізації, такі як чанкування, паралельна обробка та використання ефективних бібліотек, щоб забезпечити стабільність та продуктивність ETL-конвеєрів в умовах обмежених ресурсів.

Ключові факти

Пам'ять стала критичним ресурсом через бум ШІ, що призвело до зростання цін на продукти Micron Technology та Sandisk.
Проблема: набір даних із 6.2 мільйона постів у соціальних мережах (близько 30 ГБ) з полями змішаних типів не вміщається в стандартну пам'ять хмарного воркера.
Pandas чанкування зменшує пікове споживання пам'яті, обробляючи дані частинами, але збільшує час виконання.
Dask автоматично розділяє DataFrame та планує завдання на кількох ядрах CPU, скорочуючи час виконання, але вимагає явного визначення типів даних для уникнення помилок зі змішаними типами.
Polars, побудований на Rust та Apache Arrow, забезпечує значно вищу швидкість та менше споживання пам'яті завдяки оптимізованому коду та лінивому виконанню запитів.