ОЦЕНКА ПОЛЕЗНОСТИ SPARK-МЕТРИК ДЛЯ ДИАГНОСТИКИ УЗКИХ МЕСТ ВЫЧИСЛЕНИЙ

​EVALUATION OF THE DIAGNOSTIC VALUE OF SPARK METRICS FOR IDENTIFYING COMPUTATIONAL BOTTLENECKS

Razumovskii Dmitrii Andreevich

PhD student, National University of Science and Technology MISIS,

Russia, Moscow

Volkov Denis Dmitrievich

PhD student, National University of Science and Technology MISIS,

Russia, Moscow

Stuchilin Vladimir Valerievich

Candidate of Sciences, Deputy Head of the Department of Information Systems, associate professor, National University of Science and Technology MISIS,

Russia, Moscow

АННОТАЦИЯ

В работе рассматривается задача диагностики узких мест в приложениях Apache Spark на основе анализа эксплуатационной телеметрии. Предложена классификация типовых сценариев деградации производительности, включающая CPU-bound, memory/GC-bound, shuffle-bound, I/O-bound и перекос распределения данных по партициям. Описан набор Spark-метрик, публикуемых через GraphiteSink и отражающих использование памяти, диска, подсистемы shuffle и внешнего ввода-вывода. Был построен ряд экспериментальный процессов на основе общедоступного датасета «New York City Taxi Trips», позволяющий оценивать применимость метрик Spark-приложений для выявления проблемных кейсы, а также отдельных типов узких мест. Для формализации оценки введена трехуровневая шкала диагностической полезности, по результатам которой построен рейтинг метрик. На основе рейтинга сформирован компактный набор метрик, достаточный для первичной диагностики производительности, пригодный для использования в системах мониторинга и автоматизированного анализа инцидентов.

ABSTRACT

This paper examines the problem of diagnosing bottlenecks in Apache Spark applications based on operational telemetry analysis. A classification of typical performance degradation scenarios is proposed, including CPU-bound, memory/GC-bound, shuffle-bound, I/O-bound, and partition-skewed performance. A set of Spark metrics published via GraphiteSink and reflecting the use of memory, disk, the shuffle subsystem, and external I/O is described. A series of experimental processes based on the publicly available «New York City Taxi Trips» dataset were constructed to evaluate the applicability of Spark application metrics for identifying problem cases and specific types of bottlenecks. To formalize the assessment, a three-level diagnostic utility scale was introduced, based on the results of which a metric rating was constructed. Based on the rating, a compact set of metrics was formed, sufficient for initial performance diagnostics, suitable for use in monitoring systems and automated incident analysis.

Ключевые слова: Apache Spark; Spark-метрики; вычислительные узкие места; мониторинг производительности; spilling памяти; операции shuffle; внешний ввод-вывод; распределенные вычисления.

Keywords: Apache Spark; Spark metrics; computational bottlenecks; performance monitoring; memory spilling; shuffle operations; external I/O; distributed computing.

Apache Spark используется как базовая платформа для распределенной обработки больших объемов данных, что сопровождается широким спектром телеметрических метрик. Большое число доступных показателей одновременно расширяет возможности анализа и создает эффект информационной перегрузки при диагностировании узких мест. В работе рассматривается подход к количественной оценке диагностической полезности отдельных Spark-метрик и формированию компактного набора показателей, пригодного для задач мониторинга и первичной диагностики производительности.

В исследовании рассматривается набор кейсов выполнения приложений Apache Spark, включающий штатные запуски и запуски с различными типами узких мест. Под диагностической полезностью метрики понимается способность ее значений различать кейсы заданного. Для количественной оценки вводится шкала диагностической полезности и проводится анализ различий распределений метрик между группами кейсов, на основе которого формируется рейтинг метрик и выделяется компактный набор показателей.

Классификация узких мест вычислений в приложениях Spark включает пять основных сценариев: CPU-bound, memory/GC-bound, shuffle-bound, I/O-bound и перекос распределения данных (skew) по партициям [1] [2]. Такие сценарии задают основу для интерпретации наблюдаемого профиля выполнения.

CPU-bound сценарий характеризуется тем, что основная доля времени выполнения определяется вычислительной сложностью операций при стабильной работе подсистем памяти, ввода-вывода и shuffle. В этом случае производительность в первую очередь зависит от числа доступных ядер и эффективности реализованных алгоритмов обработки.

Memory/GC-bound сценарий отражает недостаточность ресурсов памяти для текущего профиля нагрузки. Вычислительный процесс в таких кейсах сопровождается интенсивной работой сборщика мусора и высокой чувствительностью времени выполнения к увеличению объема данных и конфигурации памяти, что приводит к ухудшению предсказуемости длительности задач и стадий.

Shuffle-bound сценарий формируется при доминирующем вкладе операций перераспределения данных между исполнителями. Он типичен для широких join-операций, переразбиения по ключам и агрегирования над крупными наборами данных, когда значимая часть времени выполнения стадий расходуется на формирование и передачу промежуточных результатов.

I/O-bound сценарий возникает при ограниченной пропускной способности или высокой латентности внешних систем хранения. В таких случаях существенная часть времени выполнения определяется операциями чтения и записи данных, а вычислительные ресурсы кластера используются менее полно по сравнению с потенциально достижимым уровнем.

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

Для оценки полезности телеметрии рассматривается набор метрик, публикуемых при работе приложения через GraphiteSink.

Базовой временной характеристикой служит метрика executorRunTime, фиксирующая длительность выполнения задач на уровне executor-процессов. Она задает опорный уровень для анализа относительного вклада вспомогательных операций и сравнения длительности задач и стадий между различными кейсами.

Состояние подсистемы управления памятью описывается показателем jvmGCTime, указывающим на время, затраченное JVM на сборку мусора, а также агрегированными по executors значениями суммарного времени выполнения задач и суммарного времени работы сборщика мусора.

Поведение операций shuffle описывается метриками блока shuffleReadMetrics и shuffleWriteMetrics. Для чтения анализируются shuffleReadMetrics.fetchWaitTime, shuffleReadMetrics.remoteBytesRead, shuffleReadMetrics.localBytesRead и shuffleReadMetrics.recordsRead. Для записи рассматриваются shuffleWriteMetrics.bytesWritten, shuffleWriteMetrics.writeTime и shuffleWriteMetrics.recordsWritten. Эти показатели характеризуют время ожидания данных shuffle, объемы сетевого и локального ввода-вывода и количество обработанных записей.

Нагрузка на подсистему памяти и диска при обработке промежуточных данных описывается метриками memoryBytesSpilled и diskBytesSpilled, фиксирующими объем данных, выгруженных из оперативной памяти в промежуточное дисковое хранилище. Их рост трактуется как индикатор нехватки оперативной памяти и усиленной нагрузки на дисковую подсистему.

Взаимодействие с внешними источниками и приемниками данных описывается метриками inputMetrics.bytesRead, inputMetrics.recordsRead, outputMetrics.bytesWritten и outputMetrics.recordsWritten. Они отражают объемы и количество записей, прочитанных и записанных задачами, и используются для сопоставления профиля ввода-вывода с наблюдаемым временем выполнения.

Дополнительную информацию об устойчивости выполнения предоставляют агрегированные счетчики числа задач, завершившихся с ошибками, в разрезе executors и приложения. Применяются при анализе сценариев перегрузки кластера и некорректной конфигурации заданий.

Предполагается, что для CPU-bound сценариев основную долю executorRunTime формируют вычислительные операции, тогда как относительный вклад jvmGCTime, memoryBytesSpilled, diskBytesSpilled и метрик блока shuffle остается умеренным. Распределения executorRunTime по задачам при этом близки к однородным.

Для memory/GC-bound сценариев ожидается рост jvmGCTime и доли времени сборки мусора в суммарном времени выполнения задач, а также увеличение memoryBytesSpilled и diskBytesSpilled [3]. В таких кейсах наблюдается усиление зависимости executorRunTime от объемов входных данных, отражаемых в inputMetrics.bytesRead и inputMetrics.recordsRead.

Для shuffle-bound сценариев ключевыми индикаторами выступают повышенные значения shuffleReadMetrics.fetchWaitTime, shuffleReadMetrics.remoteBytesRead, shuffleReadMetrics.localBytesRead, shuffleReadMetrics.recordsRead, а также shuffleWriteMetrics.bytesWritten, shuffleWriteMetrics.writeTime и shuffleWriteMetrics.recordsWritten. Значительная часть суммарного executorRunTime в этой группе кейсов связана с операциями shuffle, а при ограниченной памяти дополнительно возрастают memoryBytesSpilled и diskBytesSpilled [3].

Для I/O-bound сценариев характерны высокие значения inputMetrics.bytesRead и inputMetrics.recordsRead, а также outputMetrics.bytesWritten и outputMetrics.recordsWritten в сочетании с умеренными значениями jvmGCTime и shuffle-метрик. Такая комбинация показывает доминирование операций чтения и записи во внешние системы хранения в структуре времени выполнения.

В кейсах с перекосом распределения данных по партициям анализируются распределения inputMetrics.bytesRead и inputMetrics.recordsRead по задачам в пределах одной стадии. Существенный разрыв между типичными значениями и верхними квантилями этих распределений и наличие небольшого числа задач с аномально высокими значениями интерпретируются как признаки skew-сценариев.

Сформулированные связи используются далее для количественной оценки диагностической полезности рассматриваемых Spark-метрик на выборке реальных запусков приложений.

Методика оценки диагностической полезности Spark-метрик опирается на анализ набора кейсов выполнения приложений. В качестве основы для формирования кейсов использовался открытый датасет «New York City Taxi Trips», содержащий записи о поездках такси с временными метками, географическими координатами посадки и высадки, показателями стоимости и вспомогательными атрибутами, загруженный в распределенное хранилище. Обработка данных выполнялась на локальном кластере Apache Spark, развернутом в окружении Docker. Пример SQL плана пакетного ETL представлен на рис. 1. Выборка включает в себя штатные запуски без выраженной деградации производительности и запуски, отнесенные к одному из типов узких мест: CPU-bound, memory/GC-bound, shuffle-bound, I/O-bound или перекос распределения данных по партициям. Выборки с наличием узких мест получаются из штатной выборки путем добавления определенной нагрузки, свойственной данному кейсу. Для shuffle-bound кейса добавляется операция join, для skew по партициям – аггрегация по ключам, для I/O-bound - обогащение из внешнего хранилища, для memory/GC-bound – фильтрация и сортировка крупных наборов данных, для CPU-bound – вычислительно интенсивный расчет.

Рисунок 1. Пример SQL плана пакетного ETL

Для каждого кейса фиксируется тип нагрузки и результат экспертного анализа, позволяющий отнести запуск к определенному типу узкого места или к штатному режиму. При разметке учитываются наблюдавшиеся симптомы деградации, характеристики конфигурации и профиля данных. Структура выборки иллюстрируется примером, приведенным в таблице 1.

Таблица 1.

Описание набора кейсов

Сбор метрик организуется в виде конвейера, в рамках которого значения, публикуемые Spark через GraphiteSink, соотносятся с идентификаторами приложений и кейсов и агрегируются до табличного представления формата «кейс - метрика - агрегированное значение». Для каждой пары «кейс – метрика» вычисляются агрегированные показатели, например средние или максимальные значения по задачам или executors, в зависимости от характера метрики. Схема процесса представлена на диаграмме рис. 2.

Рисунок 2. Схема сбора метрик для кейсов

Полученная таблица «кейс x метрика» используется как входные данные для оценки диагностической полезности метрик относительно типов узких мест, заданных разметкой кейсов.

Оценка диагностической полезности метрик направлена на количественное измерение способности каждой метрики различать классы кейсов. В качестве целевых классов рассматриваются штатные и проблемные кейсы с наличием узких мест, описанные ранее. Для каждой метрики формируется вектор агрегированных значений по всем кейсам, после чего анализируются различия между группами.

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

Для перехода к унифицированной оценке вводится шкала диагностической полезности. Качественная интерпретация шкалы приведена в таблице 2.

Таблица 2.

Шкала оценивания диагностической полезности метрик

Для каждой метрики и каждого типа узкого места на основе распределений значений в соответствующей группе кейсов и в остальных группах присваивается балл по указанной шкале. Сумма баллов по всем типам узких мест рассматривается как интегральная оценка общей диагностической ценности метрики для задач эксплуатации и мониторинга.

На основе полученных интегральных оценок формируется рейтинг метрик. Этот рейтинг используется при отборе компактного набора наиболее информативных показателей и при формировании практических рекомендаций по диагностике узких мест в приложениях Apache Spark.

Для дальнейшего анализа вводится оценка метрик m для отдельных типов кейсов, рассчитываемая по формуле:

,                                                                                       (1)

где

 – среднее значение метрики m по всем кейсам типа c;

 – среднее значение той же метрики по группе штатных кейсов.

Для каждого типа кейсов использовалось не менее 10 запусков, при этом для каждой метрики вычислялось среднее значение по всем задачам в рамках данного типа. Значения нормировались на соответствующее среднее значение для группы штатных кейсов, что позволило представить результаты в относительных единицах с базовым уровнем 1,0.

Усредненные значения метрик m, рассчитанные по формуле (1), округленные до первого знака после запятой, приведенные в таблице 3, получены на основе серии запусков приложений Apache Spark, обрабатывающих подвыборку датасета «New York City Taxi Trips» на локальном кластере, развернутом в окружении Docker.

Таблица 3.

Усредненные значения метрик по типам кейсов (отн. единицы)

Для количественной оценки диагностической полезности метрик использовалась шкала из раздела 3. Для каждой метрики и каждого типа узкого места присваивались баллы от 0 до 2. Временные показатели в оценку не включались.

Баллы присваивались по следующим правилам:

• 0 – если для всех типов кейсов значение , находилось в интервале [0,8; 1,2] и выраженного смещения относительно штатного профиля не наблюдалось;
• 1 – если для рассматриваемого типа кейсов  находилось в интервале [1,2; 2,0] либо [0,5; 0,8], то есть фиксировалось умеренное, но устойчивое смещение;
• 2 – если для рассматриваемого типа кейсов  или , то есть смещение по метрике как минимум двукратное по сравнению со штатным профилем.

Таблица 4 содержит оценки для всех рассматриваемых объемных и ресурсных метрик.

Таблица 4.

Диагностическая полезность объемных и ресурсных метрик

Метрики memoryBytesSpilled и diskBytesSpilled имеют высокие оценки для memory/GC-bound и Shuffle-bound кейсов и умеренные для skew-сценариев. Они чувствительны к перегрузке подсистемы памяти и интенсивным операциям shuffle и служат индикаторами сочетания дефицита памяти и тяжелых перераспределений данных.

Группа метрик shuffleRead.remoteBytesRead, shuffleRead.localBytesRead, shuffleRead.recordsRead, shuffleWrite.bytesWritten и shuffleWrite.recordsWritten обладает высокой диагностической полезностью для Shuffle-bound кейсов и заметной полезностью для skew-сценариев. Метрика shuffleRead.recordsRead дополнительно отражает концентрацию объема shuffle-данных на части задач и поэтому получает повышенную оценку для кейсов с перекосом по партициям.

Метрики input.bytesRead и input.recordsRead одновременно выделяют I/O-bound и skew-сценарии. В I/O-bound кейсах они сочетаются с ростом output.bytesWritten и output.recordsWritten при умеренном spilling, в skew-сценариях повышенные значения наблюдаются для подмножества задач и согласуются с увеличением объемов shuffle.

Метрики output.bytesWritten и output.recordsWritten наиболее информативны для I/O-bound кейсов с высокой нагрузкой на запись. Метрика failedTasks демонстрирует высокую суммарную полезность за счет ростов в memory/GC-bound кейсах и во всех остальных проблемных сценариях и отражает переход от деградации производительности к отказам задач. Связи между типами узких мест и метриками иллюстрируются на рис. 3.

Рисунок 3. Связи между типами узких мест и метриками

На основе результатов, представленных в таблице 4, с учетом суммарных баллов формируется рекомендуемый минимальный набор метрик для эксплуатационного мониторинга: memoryBytesSpilled, diskBytesSpilled, shuffleRead.recordsRead, shuffleRead.remoteBytesRead, shuffleWrite.bytesWritten, input.bytesRead, input.recordsRead, output.bytesWritten, failedTasks. Этот набор обеспечивает различимость основных типов узких мест при ограниченном объеме наблюдаемых показателей. Дальнейшее развитие подхода может включать автоматизацию отнесения запусков к типам узких мест с использованием моделей машинного обучения и уточнение критериев диагностической полезности на расширенной выборке реальных кейсов.

Список литературы:

1. Ren X. HybridTune: Spatio-temporal Data and Model Driven Performance Diagnosis for Big Data Systems / X. Ren, Z. Cheng, H. Xu. // ArXiv – 2017. – abs/1711.07639.
2. Perez T. PETS: Bottleneck-Aware Spark Tuning with Parameter Ensembles / T. Perez, W. Chen, R. Ji, L. Liu, X. Zhou. // 27th International Conference on Computer Communication and Networks (ICCCN). – 2018.
3. Adinew D. Spark Performance Optimization Analysis In Memory Management with Deploy Mode In Standalone Cluster Computing / D. M. Adinew, Z. Shijie, Y. Liao // IEEE 36th International Conference on Data Engineering (ICDE) – Dallas, TX, USA, 2020. – pp. 2049-2053.
4. Karthikeya Sahith С. Apache Spark Big data Analysis, Performance Tuning, and Spark Application Optimization / C. S. Karthikeya Sahith, S. Muppidi and S. Merugula // International Conference on Evolutionary Algorithms and Soft Computing Techniques (EASCT) – Bengaluru, India, 2023. – pp. 1-8.