#sparksql

С версии 3.5.0Apache Spark поддерживает Datasketches – программную библиотеку стохастических потоковых алгоритмов. Разбираемся, что это такое, и при чем здесь алгоритм HyperLogLog.Что такое Apache Datasketches и зачем это нужноВ аналитике больших данных часто возникают проблемные запросы, которые не масштабируются, поскольку требуют огромных вычислительных ресурсов и времени для получения точных результатов. Например, различные счетчики, часто встречающиеся элементы, соединения, матричные вычисления и анализ графов. Причем на практике далеко не всегда нужно получить точное значение, иногда для принятия решения достаточно приближенного вычисления. В таких случаях можно применить специализированные алгоритмы, называемых потоковыми эскизами, которые дают результаты намного быстрее и с математически доказанными границами ошибок. Для анализа данных в реальном времени подобные эскизы отлично подходят. Например, эта технология помогла Yahoo сократить время обработки данных с дней до минут для нескольких внутренних платформ.Реализовать такой подход эскизов данных позволяет Apache DataSketches — высокопроизводительная библиотека стохастических потоковых алгоритмов с открытым исходным кодом. Эскизы — это небольшие программы с отслеживанием состояния, которые обрабатывают массивные данные в виде потока и могут дать приблизительные ответы с математическими гарантиями на сложные в вычислительном отношении запросы намного быстрее, чем традиционные точные методы. Библиотека DataSketches была специально разработана для производственных систем, которые должны обрабатывать большие объемы данных. Она включает адаптеры для Apache Hive, Pig и PostgreSQL, а также для совместимых двоичных представлений на разных языках (Java, C ++, Python) и платформах.Эскизы выполняются очень быстро, поскольку алгоритмы в библиотеке DataSketches обрабатывают данные за один проход и подходят как для реального времени, так и для пакетной обработки. Помимо высокой скорости обработки огромного объема данных, проектирование системы на основе эскизов позволяет упростить ее архитектуру и сократить общие вычислительные ресурсы, необходимые для решения сложных вычислительных задач.Эскизы отличаются от традиционных методов выборки тем, что исследуют все элементы потока, касаясь каждого элемента только один раз, и часто рандомизацию, которая составляет основу их стохастической природы. Это делает эскизы идеальным вариантом для потоковой обработки в реальном времени.Например, первый этап процесса создания эскизов с подсчетом различных значений представляет собой преобразование, которое придает входному потоку данных свойство белого шума или равномерное распределение значений. Обычно это достигается путем скоординированного хеширования входных уникальных ключей и последующей нормализации результата до однородного случайного числа от нуля до единицы.Второй этап эскиза представляет собой структуру данных, которая следует набору правил для сохранения небольшого набора хеш-значений, полученных на этапе преобразования. Эскизы также отличаются от простых схем выборки тем, что размер эскиза часто имеет настраиваемую фиксированную верхнюю границу, что обеспечивает простое управление памятью.Последним элементом процесса создания эскиза является набор алгоритмов оценки, которые по запросу исследуют структуру данных эскиза и возвращают значение результата. Это значение результата будет приблизительным, но будет иметь четко установленные и математически подтвержденные границы распределения ошибок.Принцип создания эскизов данных в Apache DatasketchesЭскизы данных обычно характеризуются следующими свойствами:- Небольшой размер - на несколько порядков меньше потока необработанных входных данных.Эскизы реализуют сублинейные алгоритмы, размер которых увеличивается гораздо медленнее, чем размер входного потока. Некоторые эскизы имеют конечную верхнюю границу размера, которая не зависит от размера входного потока.- Высокая скорость обработки, т.к. время обновления не зависит от размера или порядка входного потока. Для вычислений эскизу достаточно увидеть каждый элемент в потоке данных только один раз.- Высокая параллелизуемость. Структуры данных эскиза являются дополнительными, поскольку их можно объединять без потери точности.- Приблизительный характер. Например, для эскизов с уникальным подсчетом относительные границы ошибок являются функцией от заданного размера.Технология создания эскизов данных использует HyperLogLog - вероятностный алгоритм для приблизительного подсчета уникальных элементов в мультимножестве. Вычисление точной кардинальности множества требует памяти пропорциональной самой кардинальности, что не приемлемо при работе с большими объёмами данных. Вероятностные алгоритмы вычисления кардинальности, такие как HyperLogLog, потребляют намного меньше памяти, вычисляя приближенное значение кардинальности. Например, HyperLogLog способен вычислить приближенное значение кардинальности 10^9 с погрешностью в 2 %, используя 1.5 килобайт памяти.Таким образом, предложенный в 2007 году, HyperLogLog решает проблему оценки мощности в прикладной математике, вместо среднего геометрического для усреднения результата используя среднее гармоническое – величину, обратную среднему значению обратных величин.Этот алгоритм может оценить количество уникальных значений в очень большом наборе данных, используя мало памяти и времени. Алгоритм используется в key-value NoSQL-СУБД Redis, а также других нереляционных базах и хранилищах данных петабайтного масштаба, например, Amazon Redshift , Presto, BigQuery и Apache Druid. Библиотека Apache Datasketches также использует HyperLogLog. Познакомившись с Apache Datasketches, далее рассмотрим, как эта стохастическая библиотека улучшает Spark Structured Streaming.Зачем это в Spark Streaming?Использование библиотеки Datasketches для вычисления приблизительного количества отдельных объектов дает несколько преимуществ по сравнению с подсчетом целых чисел прямого результата, возвращаемым функцией approx_count_distinct, ранее доступной в Apache Spark и среде выполнения Databricks. Например, можно сохранять эскизы в хранилище и при необходимости сканировать их позже, чтобы выполнять дальнейший анализ данных или проводить вычисления без необходимости пересчитывать отдельные значения с нуля. Это экономит время и вычислительные ресурсы, поскольку приблизительное количество уникальных экземпляров доступно для последующих запросов.Еще одним преимуществом использования буферов эскизов является их гибкость при работе с различными сценариями. Эскизы можно легко комбинировать или объединять с помощью операций объединения, что позволяет пользователям объединять несколько буферов эскизов в один эскиз. Такая гибкость обеспечивает масштабируемую обработку больших наборов данных и распределенных систем, поскольку эскизы можно создавать независимо, а затем эффективно объединять. Эти буферы эскизов дают пользователям возможность выполнять над эскизами расширенные операции над наборами, такие как объединения, пересечения и различия, открывая новые возможности для сложных задач анализа данных. В общем случае, приблизительный подсчет особенно полезен в таких случаях использования инкрементального обновления, где для приложений с низкой задержкой, таких как информационные панели, допускается небольшая погрешность в угоду более быстрого выполнения запроса.Для этого в библиотеке Dataksetches есть функция hll_sketch_agg, которая подсчитывает количество уникальных значений в столбце. Используя алгоритм HyperLogLog, эта функция обеспечивает вероятностную аппроксимацию уникальности, выводя двоичное представление как буфер эскиза. Этот буфер эскизов очень эффективен для долгосрочного хранения и персистентности. Можно интегрировать hll_sketch_agg в свои запросы и с помощью полученных буферов вычислить приблизительное количество уникальных значений. Эта функция hll_sketch_estimate является мощным дополнением к hll_sketch_agg. С помощью входных данных буфера эскизов, созданного с помощью hll_sketch_agg, функция hll_sketch_estimate обеспечивает оценку количества уникальных элементов, позволяя принимать обоснованные решения на основе надежных приближений.Функция hll_union позволяет объединить два эскиза в один, когда надо агрегировать данные по разным столбцам или наборам данных. Включив hll_union в запросы, можно получить исчерпывающую информацию и вычислить приблизительное количество уникальных пользователей с помощью hll_sketch_estimate. Так можно создавать буферы эскизов и экспортировать их в управляемые таблицы, чтобы избежать повторных вычислений в дальнейшем, используя функцию hll_sketch_agg. Все эти новые функции Apache Spark SQL и Databricks Runtime основаны на библиотеке Apache Datasketches, которая поддерживается в Spark с версии 3.5.0.Освойте возможности Apache Spark для разработки приложений аналитики больших данных на специализированных курсах в нашем лицензированном учебном центре обучения и повышения квалификации для разработчиков, менеджеров, архитекторов, инженеров, администраторов, Data Scientist’ов и аналитиков Big Data в Москве:- Основы Apache Spark для разработчиков- Потоковая обработка в Apache Spark- Анализ данных с Apache Spark- Машинное обучение в Apache Spark- Графовые алгоритмы в Apache SparkИсточники1. https://www.databricks.com/blog/apache-spark-3-apache-datasketches-new-sketch-based-approximate-distinct-counting2. https://datasketches.apache.org/3. https://chengweihu.com/hyperloglog/

Из-за чего приложения Flink работают быстрее Spark: разница в моделях обработки данных, управлении памятью, методах оптимизации, дизайне API и личный опыт использования.Apache Flink vs Spark: сходства и отличияApache Spark и Flink считаются наиболее популярными фреймворками разработки распределенных приложений в области Big Data. Они достаточно похожи, что мы ранее рассматривали здесь: оба имеют высокоуровневый API для потоковой и пакетной обработки, позволяя разработчикам писать сложные вычислительные конвейеры обработки. Также эти фреймворки похожи по набору встроенных модулей и библиотек, например, поддержка ANSI SQL и машинного обучения. Однако, в части производительности, Spark в большинстве случаев проигрывает Flink. Это происходит из-за отличий технологий по следующим ключевым критериям:- модели обработки данных и структуры их представления;- управление памятью;- методы оптимизации;- дизайн API;- пользовательский опыт.Сравним обе технологии по этим критериям более подробно.Различия в моделях обработки данныхОдна из основных причин, почему производительность Spark ниже, чем у Flink, связана с их разными моделями обработки данных. Хотя оба фреймворка используют модель распределенной обработки данных, они по-разному обрабатывают распределение данных и параллелизм. Spark считается платформой обработки данных третьего поколения и изначально поддерживает пакетную и потоковую обработку. Spark использует микропакетную обработку, которая делит неограниченный поток событий на небольшие фрагменты (пакеты) и запускает вычисления. Spark повысил производительность классической модели MapReduce из Apache Hadoop, выполняя обработку в памяти вместо того, чтобы записывать каждый шаг обратно на диск.Реализация вычислительной модели MapReduce в Hadoop и в SparkSpark использует отказоустойчивую распределенную коллекцию данных (RDD) в качестве фундаментальной структуры. RDD представляет собой неизменяемую версию данных, доступную только для чтения, хранящуюся в распределенном кластере. Будучи отказоустойчивой структурой данных, RDD может восстанавливаться после сбоев без потери информации. Однако, такая конструкция не оптимизирована для параллелизма и не гарантирует отсутствия конфликтов рассогласования, когда несколько операций выполняются над одним и тем же RDD одновременно.Разумеется, кроме RDD в Spark есть и другие структуры данных, которые предоставляют гибкий высокоуровневый API: Dataframe и Dataset, о чем мы писали здесь. Однако, они все равно реализуют модель микропакетных, а не потоковых вычислений, в отличие от истинно потоковой природы Flink, который уже несколько лет назад объединил API пакетной и потоковый обработки. Подробнее об универсальном API Apache Flink читайте в этом материале.Apache.Flink является распределенным фреймворком обработки данных четвертого поколения, изначально поддерживая пакетную и потоковую обработку. Он обеспечивает непрерывную потоковую передачу, при которой вычисления событий запускаются сразу после их получения. Flink использует более гибкую распределенную модель, основанную на потоковой обработке с отслеживанием состояния. При потоковой обработке с отслеживанием состояния данные обрабатываются в непрерывном потоке и могут храниться в памяти для более быстрого доступа. Такая конструкция позволяет Flink оптимизировать параллелизм и обрабатывать большие наборы данных более эффективно, чем Spark.Вычислительная модель Apache FlinkFlink предлагает встроенную потоковую передачу, а Spark использует микропакеты для эмуляции потоковой передачи. Это означает, что Flink обрабатывает каждое событие в режиме реального времени и обеспечивает очень низкую задержку. Spark, используя микропакетную обработку, может обеспечить обработку только в режиме, близком к реальному времени. Для многих случаев использования Spark обеспечивает приемлемый уровень производительности. Однако, благодаря низкой задержке Flink превосходит производительность Spark даже при более высокой пропускной способности. Spark может обеспечить низкую задержку при более низкой пропускной способности, но увеличение пропускной способности также приведет к увеличению задержки. Такой компромисс означает, что пользователям Spark необходимо настроить конфигурацию для достижения приемлемой производительности, что повышает сложность разработки.Управление памятьюЕще одна причина, по которой производительность Spark ниже, чем у Flink, связана с управлением памятью. Как упоминалось ранее, RDD являются неизменяемыми и должны храниться в памяти во время работы, что увеличивает интенсивное потребление этого ресурса и может привести к конфликтам. Кроме того, модель отложенных вычислений Spark также может стать причиной перетасовке данных с их передачей по сети, что также снижает производительность. Напротив, Flink использует более эффективную модель управления памятью, основанную на колоночном хранении и партиционировании данных. Колоночное хранилище организует данные в столбцы фиксированного размера, которые можно обрабатывать параллельно без необходимости перемешивания, т.е. shuffle-операций. Такой подход снижает потребление памяти и повышает производительность, позволяя обрабатывать данные непосредственно с диска, а не загружать их в память сразу.Методы оптимизацииОба фреймворка предлагают широкий набор методов оптимизации для повышения производительности, но они существенно различаются. Spark в большей степени полагается на преобразования и действия для оптимизации производительности, тогда как Flink использует более декларативную модель программирования, которая работает быстрее. В Spark разработчики могут использовать преобразования и действия для оптимизации производительности, применяя определенные оптимизации к своему коду. Например, использование соединений на этапе сопоставления (Map) или фильтрации перед группировкой может повысить производительность за счет уменьшения объема данных, которые необходимо передать по сети. Однако, эти оптимизации довольно сложны в реализации.Декларативная модель программирования Apache Flink позволяет разработчикам выражать логику обработки данных проще и поддерживает оптимизацию производительности за счет автоматического применения лучших практик, таких как отложенные вычисления, партиционирование и кэширование. Кроме того, Flink предоставляет встроенную поддержку широковещательных переменных и аккумуляторов, что также повышает производительность.Задания Spark должны быть оптимизированы разработчиками вручную с помощью расширяемого оптимизатора Catalyst, основанном на конструкции функционального программирования Scala. Spark упрощает создание новых оптимизаций и позволяет разработчикам расширять оптимизатор Catalyst. Spark также включает в себя механизм выполнения Tungsten, который оптимизирует физическое выполнение операций для повышения производительности.Flink имеет оптимизатор пакетной обработки на основе затрат, который анализирует поток данных и оптимизирует задания перед выполнением на движке потоковой передачи, выбирая наиболее эффективный план выполнения на основе доступных ресурсов и характеристик данных. Оптимизатор Flink не зависит от интерфейса программирования и работает аналогично оптимизаторам реляционных баз данных, прозрачно применяя оптимизацию к потокам данных. Потоковая обработка Flink выигрывает от выполнения на основе конвейера и планирования с малой задержкой.Дизайн APIХотя оба фреймворка предлагают мощные высокоуровневые API, разобраться в них не так-то просто. Apache Flink имеет многоуровневые API, которые предлагают разные уровни выразительности и контроля и предназначены для разных типов использования. Тремя уровнями API являются функции процесса (также известные как API обработки потоков с отслеживанием состояния), поток данных, а также таблицы и SQL. API потоковой обработки с отслеживанием состояния требует написания подробного кода, но обеспечивает максимальный контроль над временем и состоянием, которые являются основными концепциями потоковой обработки с отслеживанием состояния. API DataStream поддерживает Java, Scala и Python и предлагает примитивы для многих распространенных операций потоковой обработки, а также баланс между подробностью кода или выразительностью и контролем. API таблиц и SQL — это реляционные API, которые обеспечивают поддержку Java, Scala, Python и SQL. Они предлагают высочайшую абстракцию и интуитивно понятный, подобный SQL, декларативный контроль над потоками данных. Flink также обеспечивает плавный переход и переключение между этими API.Структурированная потоковая передача Apache Spark предлагает API-интерфейсы Dataset и DataFrame, которые предоставляют высокоуровневые API-интерфейсы декларативной потоковой передачи для представления статических, ограниченных данных, а также потоковых, неограниченных данных. Операции поддерживаются в Scala, Java, Python и R. Spark имеет богатый набор функций и синтаксис с простыми конструкциями для выбора, агрегирования, управления окнами, соединениями и пр. Можно использовать API потоковых таблиц для чтения таблиц как потоковых кадров данных в качестве расширения API DataFrame.В целом можно сказать, что API-интерфейсы Spark Structured Streaming эквивалентны API-интерфейсам Flink Table и SQL. Spark предоставляет функции управления окнами для обработки потоковых данных в фиксированных или скользящих временных окнах. Однако, оконное управление Spark менее гибкое и эффективное по сравнению с Flink из-за микропакетной обработки. Flink имеет расширенную поддержку окон, включая окна на основе времени события и времени обработки, окна сеанса и гибкие настраиваемые оконные функции. Управление окнами Flink более эффективно и точно для потоковой обработки, поскольку оно разработано специально для непрерывных потоков данных.Пользовательский опыт и поддержка сообществаС точки зрения пользовательского опыта, мне лично Apache Flink показался более сложным по сравнению со Spark. В частности, из-за того, что Flink основан на потоковой модели обработки данных, чтобы использовать возможности этого фреймворка разработчик должен хорошо понимать особенности этой вычислительной парадигмы и управлять состоянием данных в реальном времени. А это на порядок сложнее пакетной и микропакетной обработки в Apache Spark.Дополнительную сложность вносит большая зрелость и распространенность Spark по сравнению с Flink. В свою очередь, Flink отличный выбор для сложных случаев обработки событий или встроенной потоковой передачи, поскольку обеспечивает лучшую производительность, задержку и масштабируемость. Кроме того, он имеет лучшую поддержку оконных функций и управления состоянием. Наконец, он позволяет делать сложные вычисления с помощью примитивных операций, что потребует разработки специальной логики в Spark. Однако, порог входа в Apache Flink выше.Таким образом, хотя разница в моделях обработки данных, управлении памятью, методах оптимизации и дизайне API обусловливают более высокую производительность Apache Flink По сравнению со Spark, решение о выборе технологии следует принимать с учетом компетенций команды реализации. Если разработчики лучше владеют Spark, и в ТЗ нет жестких требований к обработке данных строго в реальном времени, этот фреймворк отлично позволит реализовать распределенное отказоустойчивое приложение. Однако, если задержка должна измеряться в микросекундах, имеет смысл рассмотреть Flink с его широким набором инструментов.Освойте возможности Apache Flink и Spark для потоковой обработки событий в распределенных приложениях аналитики больших данных и машинного обучения на специализированных курсах в нашем лицензированном учебном центре обучения и повышения квалификации для разработчиков, менеджеров, архитекторов, инженеров, администраторов, Data Scientist’ов и аналитиков Big Data в Москве:- Потоковая обработка в Apache Spark- Потоковая обработка данных с помощью Apache FlinkИсточники1. https://medium.com/@acmilanellosw/why-spark-performance-is-less-than-flink-a6a02a8b622. https://aws.amazon.com/ru/blogs/big-data/a-side-by-side-comparison-of-apache-spark-and-apache-flink-for-common-streaming-use-cases3. https://www.macrometa.com/event-stream-processing/spark-vs-flink4. https://www.dataversity.net/spark-vs-flink-key-differences-and-how-to-choose/

13 сентября 2023 года вышел Apache Spark 3.5. Знакомимся с самыми важными новинками свежего релиза: расширения Spark Connect и SQL, поддержка DeepSpeed, улучшения потоковой передачи и свежие UDF-функции Python.ТОП-5 новинок Apache Spark 3.5.0В Apache Spark 3.5. добавлено много исправлений и улучшений, а также реализованы новые функции. Наиболее интересными из них с точки зрения дата-инженера можно назвать следующие:- Spark Connect поддерживает больше сценариев с общедоступностью клиента Scala, поддержкой распределенного обучения и вывода, паритетом API Pandas в SPARK и улучшенной совместимостью для структурированной потоковой передачи.- функции SQL для управления массивами, расширенная поддержка функций SQL для API Scala, Python и R, поддержка именованных аргументов для вызовов функций SQL;- агрегаты, а также оптимизированные для Arrow пользовательские функции Python, табличные UDF-функции Python, API тестирования PySpark и расширенные классы ошибок в PySpark;- распределенное обучение с помощью DeepSpeed ​​в кластерах Spark и английский SDK для Apache Spark, о котором мы писали здесь;- улучшения потоковой передачи, включая новый API и изменения в провайдере key-value базы данных RocksDB, которые уменьшают компромиссы по сравнению с другими резидентными хранилищами состояний.Рассмотрим некоторые из этих новинок более подробно.Улучшения Spark ConnectНапомним, Spark Connect представляет собой клиент-серверную архитектуру, которая позволяет разделить клиентский и серверный компоненты, обеспечивая удаленное подключение к кластерам Spark через API DataFrame и неразрешенные логические планы в качестве протокола. Благодаря такому разделению между клиентом и сервером Spark, его можно использовать откуда угодно, включая различные IDE, интерактивные среды выполнения типа Google Colab и Jupyter Notebook, а также различные языки программирования. Кроме того, в эти среды можно встроить Spark Connect, чтобы обеспечить бесшовную интеграцию со Spark. Впервые полноценная поддержка Spark Connect была выпущена в версии 3.4.0, опубликованной в апреле 2023 года, о чем мы писали здесь.С момента выпуска релиза 3.4.0 было выполнено более 600 улучшений Spark Connect. Ключевым результатом для Spark 3.5 и компонента Spark Connect является общедоступность клиента Scala для Spark Connect, реализованного с помощью рефакторинга подмодуля SQL. Теперь он разделен на клиентский (SQL-API) и серверный модули, чтобы уменьшить набор зависимостей, необходимых клиенту для изоляции пути к классам. До выпуска Spark 3.5 было невозможно использовать библиотеку машинного обучения MLlib Apache Spark напрямую с Spark Connect, поскольку он использует шлюз Py4J, требующий совместного размещения клиентского приложения. В Spark 3.5 можно выполнять распределенное обучение и логический вывод с помощью Spark Connect с использованием новой платформы распределенного выполнения на основе PyTorch. В настоящее время этот модуль поддерживает классификаторы логистической регрессии, преобразователи базовых функций, оценщики базовых моделей, конвейеры машинного обучения и кросс-валидацию. Эта платформа легко интегрируется с векторизованной платформой Python UDF в Spark, расширяя ее возможностью выполнения UDF с использованием барьерного режима выполнения.Новые возможности PySparkВ выпуске Apache Spark 3.5.0 представлены значительные улучшения PySpark, включая оптимизированные для Arrow пользовательские функции Python, в т.ч. табличные (UDTF), улучшенные сообщения об ошибках и новый API тестирования, который значительно повышает удобство использования, производительность и тестируемость PySpark-приложений. Теперь UDF- функции Python будут использовать колоночный формат Apache Arrow для повышения производительности, когда для конфигурации spark.sql.execution.pythonUDF.arrow.enabled установлено значение True, или когда для параметра useArrow установлено значение True с помощью декоратора UDF. Благодаря этой оптимизации пользовательские функции Python могут работать до 2 раз быстрее на современных архитектурах ЦП, благодаря векторизованному вводу-выводу. Рассмотрим, как включить возможность использования Apache Arrow через установку конфигурации:spark.conf.set("spark.sql.execution.pythonUDF.arrow.enabled", True)или в коде PySpark-приложения через декоратор:@udf("integer", useArrow=True)def my_len_udf(s: str) -> int:return len(s)Пользовательская табличная функция (UDTF, User Defined Table Function) — это тип определяемой пользователем функции, которая возвращает всю выходную таблицу вместо одного скалярного значения результата. Пользователи PySpark теперь могут писать свои собственные UDTF, интегрируя свою логику Python, и использовать их в PySpark и SQL.Также Apache Spark 3.5 представляет новые служебные функции API для проверки равенства DataFrame, включая подробные сообщения об ошибках теста с цветовой кодировкой, которые четко указывают на различия между схемами датафрейма и данными внутри него. Эти функции (pyspark.testing.assertDataFrameEqual, pyspark.testing.assertPandasOnSparkEqual, pyspark.testing.assertSchemaEqual) позволяют разработчикам легко добавлять тесты на равенство, которые дают практические результаты для их приложений и повышают производительность.Наконец, добавлены расширенные сообщения об ошибках в PySpark. Ранее набор исключений, создаваемых драйвером Python Spark, не использовал классы ошибок, представленные в Apache Spark 3.3. Теперь все ошибки из DataFrame и SQL перенесены в соответствующий пакет и содержат нужные классы и коды ошибок.Новинки Spark SQLApache Spark 3.5 представляет множество новых функций и улучшений SQL, упрощая создание запросов с помощью API-интерфейсов SQL и DataFrame. В частности, новые встроенные функции SQL для управления массивами помогают пользователям легко манипулировать их значениями вместо создания собственных UDF. А предложение IDENTIFIER позволяет гибко и безопасно создавать новые шаблоны SQL-запросов без риска атак через SQL-инъекции. Например, предложение IDENTIFIER для строковых литералов с указанием имен таблиц/столбцов/функций можно использовать вместе с функцией параметров запроса, добавленной в предыдущем выпуске Spark:spark.sql("CREATE TABLE IDENTIFIER(:tbl)(col INT) USING json",args = {"tbl": "my_schema.my_tbl"})spark.sql("SELECT IDENTIFIER(:col) FROM IDENTIFIER(:tbl)",args = {"col": "col","tbl": "my_schema.my_tbl"}).show()До версии 3.5 в Spark было множество функций SQL, недоступных в API-интерфейсах Scala, Python или R DataFrame. В выпуске 3.5 представлено более 150 функций SQL в API DataFrame, включая поддержку именованных аргументов для вызовов функций SQL. Теперь в SQL-запросах, подобно Python-скрипту, можно вызывать функции с именами параметров, предшествующими их значениям. Это соответствует спецификации стандарта SQL и приводит к более четкому и надежному языку запросов, когда функция имеет много параметров и/или некоторые параметры имеют значения по умолчанию, например:SELECT mask('AbCD123-@$#',lowerChar => 'q',upperChar => 'Q',digitChar => 'd')Наконец, Apache Spark 3.5 включает новые функции SQL для точного и эффективного подсчета уникальных значений внутри групп, включая сохранение результатов промежуточных вычислений в буферах эскизов, которые можно сохранить в хранилище и позже загрузить обратно. В этих реализациях используется библиотека Apache Datasketches для простой интеграции с другими инструментами.Новости для специалистов по Data ScienceВ релизе 3.5.0 в Python-интерфейсе PySpark добавлен модуль DeepspeedTorchDistributor, который упрощает распределенное обучение с помощью библиотеки DeepSpeed ​​в кластерах Spark. Эта библиотека оптимизации глубокого обучения с открытым исходным кодом для PyTorch снижает вычислительную мощность и потребление памяти. Она предназначена для обучения больших распределенных моделей с лучшим параллелизмом. Модуль DeepspeedTorchDistributor является расширением модуля TorchDistributor, выпущенного в Apache Spark 3.4. Под капотом DeepspeedTorchDistributor инициализирует среду и каналы связи, необходимые для DeepSpeed. Модуль поддерживает распределение обучающих заданий как в одноузловых кластерах с несколькими графическими процессорами, так и в многоузловых кластерах графических процессоров.Также в релиз 3.5.0 добавлен англоязычный набор инструментов разработчика (English SDK for Apache Spark), который компилирует инструкции на английском языке в датафреймы Python. English SDK основан на PySpark-AI, Python-оболочке, которая использует модели генеративного языка для упрощения генерации кода PySpark. Принимая инструкции на английском языке, он объединяет возможности фреймворка с такими моделями, как GPT-4 и GPT-3.5. PySpark-AI принимает на вход англоязычные инструкции и выполняет их, позволяя пользователю сфокусироваться на обработке данных, а не на кодировании. Подробнее об этом мы писали здесь.Потоковая передачаПредыдущая версия фреймворка позволяла выполнять операции с отслеживанием состояния (агрегацию, дедупликацию, соединения потоков и т. д.) несколько раз в одном и том же запросе, включая агрегирование временных окон по цепочке. Однако, соединение временных интервалов потока с последующим оператором с отслеживанием состояния не поддерживалось. Это ограничение снято в релизе 3.5, чтобы обеспечить более сложные рабочие нагрузки, например соединение потоков рекламы и кликов, а также агрегирование по временному интервалу.Кроме того, в релизе 3.5.0 представлен новый механизм контрольных точек для провайдера хранилища состояний RocksDB под названием Changelog Checkpointing, который сохраняет журнал изменений состояния. Это значительно уменьшает задержку фиксации и снижает сквозную задержку. Чтобы включить эту функцию, надо установить для свойства конфигурации spark.sql.streaming.stateStore.rocksdb.changelogCheckpointing.enabled значение true. Примечательно, что включить новую функцию можно даже с существующими контрольными точками.Также в Apache Spark реализовано более детальное управление памятью, которое позволяет пользователям ограничивать общее использование памяти экземплярами RocksDB в одном процессе-исполнителе, позволяя анализировать и настраивать использование памяти для каждого процесса-исполнителя. Это повышает производительность и надежность использования key-value СУБД в качестве хранилища состояний для stateful-приложений.Для дедупликации событий в потоковых запросах представлен новый API dropDuplicatesWithinWatermark(), который позволяет обрабатывать ситуацию, с дублирующимися событиями, даже если их фактические отметки времени немного отличаются. Такое часто случается, когда данные потребляются из топика Kafka, будучи отправлены туда неидемпотентным продюсером, а в качестве времени события используется автоматическая отметка времени записи.Освойте возможности Apache Spark для разработки приложений аналитики больших данных на специализированных курсах в нашем лицензированном учебном центре обучения и повышения квалификации для разработчиков, менеджеров, архитекторов, инженеров, администраторов, Data Scientist’ов и аналитиков Big Data в Москве:- Основы Apache Spark для разработчиков- Потоковая обработка в Apache Spark- Анализ данных с Apache Spark- Машинное обучение в Apache Spark- Графовые алгоритмы в Apache SparkИсточники1. - https://spark.apache.org/releases/spark-release-3-5-0.html2. - https://www.databricks.com/blog/introducing-apache-sparktm-35

Сегодня рассмотрим, какие улучшения Apache Spark опубликованы в 2023 году и как подать свое предложение по улучшению самого популярного вычислительного движка в стеке Big Data.Что такое SPIP и как подать свое предложение по улучшению фреймворкаВ любом продукте помимо ошибок есть также предложения по улучшению. В Apache Spark они называются SPIP - Spark Improvements Proposals. SPIP используется для значительных изменений, ориентированных на пользователя, или для сквозных изменений, а не для небольших постепенных улучшений. По своей сути SPIP аналогичен требованиям к продукту, однако причиной его регистрации является не острая потребность, а гипотеза, что реализация этой фичи позволит улучшить пользовательский опыт.Поскольку Apache Spark является проектом с открытым исходным кодом, активно развиваемым сообществом, предложить SPIP может любой пользователь, от разработчика до аналитика данных, главное – обосновать актуальность и выполнимость предлагаемого изменения, ответив на следующие вопросы:- Что вы предлагаете? Здесь следует кратко, но точно и понятно сформулировать цели изменения в общепринятых терминах без жаргонизмов- Какую проблему НЕ решает предложение? Ответ на этот вопрос позволит четче очертить границы и содержание SPIP.- Как рассматриваемая проблема решается сегодня и каковы ограничения нынешней практики?- Что нового в предложении и почему оно будет успешным?- Каков эффект от реализации предложения?- Каковы риски?- Как много времени это займет?- Каковы промежуточные и итоговые критерии проверки успешности реализации предложения?Также автор SPIP может дополнить свой документ несколькими приложениями, описав свое видение изменений API с учетом обратной и прямой совместимости. Дополнительно можно предоставить эскиз проекта для оценки технической осуществимости и ранее рассмотренные альтернативы, которые были отклонены.Разумеется, отправляя предложения по улучшению проекта Apache Spark, автор соглашается с необходимостью принять участие в его обсуждении и, желательно, в решении затронутого вопроса. После создания SPIP автор должен отправить электронное письмо на [email protected], чтобы уведомить сообщество о своем предложении, после чего начинается обсуждение задачи, созданной в JIRA. Если SPIP слишком мал или, наоборот, представляет собой полноценный инкремент, который должен быть выполнен в рамках обычного процесса разработки, метка SPIP удаляется.Все обсуждения SPIP происходят на публичном форуме в рамках обсуждения задачи в пространстве JIRA. Любые обсуждения на внешних ресурсах должны быть доступны и запротоколированы. По окончании дискуссии проводится открытое голосование в течение 3-х суток. Решение о принятии SPIP в бэклог продукта принимается по итогам голосования на основе консенсуса и с учетом того, что как минимум один разработчик-член сообщества обязуется реализовать предлагаемые изменения в течение месяца. Иначе SPIP отклоняется с указанием причины отказа. Реализация предложенных улучшений происходит посредством стандартного процесса внесения изменений в код с последующим выпуском релиза и документации.Чтобы посмотреть список открытых или принятых в работу предложений по улучшению, в поле фильтра JIRA-пространства проекта Spark надо ввести следующий запрос:project = SPARKAND status in (Open, Reopened, "In Progress")AND (labels = SPIP OR summary ~ "SPIP")ORDER BY createdDate DESCЭтот запрос выведет перечень актуальных предложений по улучшению, отсортированных по дате создания. Рассмотрим некоторые из них более внимательно.4 недавних предложения улучшить Spark SQLНапример, 30 августа 2023 года предложено добавить в PySpark возможность работать с хранимыми процедурами, которые расширяют стандартный ANSI SQL, позволяя многократно выполнять сложную логику обработки данных. Это предложение направлено на расширение Spark SQL путем введения поддержки хранимых процедур, начиная с Python в качестве процедурного языка. Также пользователи смогут сохранять эти процедуры в каталогах, например, Hive Metastore, для повторного использования в будущем.Еще одним полезным предложением в августе 2023 года стала идея повысить производительность запросов широковещательного хэш-соединения (BroadcastHashJoin) с помощью ключа соединения на стороне потока для столбцов без разделов. Поскольку ключи BroadcastHashJoin уже доступны до фактической оценки итератора потока, их можно передать в источник данных как SortedSet. Для столбцов без разделов источники данных, такие как Iceberg, уже имеют максимальную/минимальную статистику для столбца, доступную на уровне манифеста, а для таких форматов, как Parquet, эта статистика есть различных уровнях хранения. Переданный SortedSet можно использовать для сокращения использования диапазонов как на уровне драйвера (файлы манифестов), так и на уровне исполнителя при фактическом прохождении групп строк и пр. Если данные хранятся в формате Columnar Batch, отфильтровать отдельные строки на уровне DataSource невозможно, даже при наличии ключей. Однако, на уровне сканирования (ColumnToRowExec) отфильтровать как можно больше строк все же возможно, если запрос включает вложенные соединения. Таким образом, можно сократить количество строк для соединения на более высоких уровнях, что ускорит выполнение многоуровневых вложенных запросов с оператором BroadcastHashJoin.С PySpark связано еще одно предложение по улучшению, поданное в июне 2023 года. Автор рекомендует создать Python API для источников данных, чтобы разработчики, использующие этот популярный язык программирования, могли создавать собственные источники данных. Также в июне этого же года подано предложение по реализации платформы тестирования PySpark-приложений, чтобы оптимизировать и упростить этот процесс с помощью базового PySpark-класса Test Base и специальных тестовых функций. Это позволит тестам совместно использовать сеансы Spark и автоматизировать проверку кода.Освойте возможности Apache Spark для разработки приложений аналитики больших данных на специализированных курсах в нашем лицензированном учебном центре обучения и повышения квалификации для разработчиков, менеджеров, архитекторов, инженеров, администраторов, Data Scientist’ов и аналитиков Big Data в Москве:- Основы Apache Spark для разработчиков- Потоковая обработка в Apache Spark- Анализ данных с Apache Spark- Машинное обучение в Apache Spark- Графовые алгоритмы в Apache SparkИсточники1. https://spark.apache.org/improvement-proposals.html2. https://issues.apache.org/jira/browse/SPARK-45023