Руководство по Apache Spark не для начинающих: оптимизация

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

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

[Оглавление]

Зачем оптимизировать Spark?

№1. Базовые концепции оптимизации

— Схема данных
— Кеширование и персистенция
— Управление партициями

№2. Оптимизация запросов

Выбор правильных операций
— Трансформации
— Действия
— Принцип ленивых вычислений
— Как выбрать правильные операции
— Использование map и flatMap
— Использование reduceByKey и groupByKey

Порядок выполнения операций
— Важность выполнения фильтрации и агрегации до соединений
— Стратегии построения запросов от меньших таблиц к большим
— Различные стратегии для уменьшения данных до соединения

Зачем оптимизировать Spark?

Для этого есть несколько веских причин.

Сокращение времени выполнения задач.

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

Повышение эффективности использования ресурсов.

Оптимизация помогает более эффективно использовать вычислительные ресурсы, такие как процессорное время и оперативная память. Это позволяет обрабатывать больше данных с меньшими затратами и предотвращает перегрузку системы.

Улучшение производительности приложений.

Оптимизированные Spark-приложения выполняются быстрее и с меньшими задержками. Это особенно важно для приложений реального времени, где скорость обработки данных имеет критическое значение.

Снижение затрат.

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

Стабильность и надёжность.

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

Масштабируемость.

Оптимизированные решения легче масштабировать, так как они используют ресурсы более эффективно и могут обрабатывать увеличивающиеся объёмы данных без существенного увеличения времени выполнения.

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

Перейдём к концепциям.

№1. Базовые концепции оптимизации

В эту главу входят: схема данных, кэширование и персистенция, управление партициями. Начнём со схемы данных.

Схема данных

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

Использование схемы вместо автоматического определения типов

По умолчанию, Spark может автоматически определять типы данных при чтении файлов, таких как CSV, JSON и Parquet. Это удобно для быстрой разработки. Однако, автоматическое определение типов данных (schema inference) может негативно сказаться на производительности. 

Вот почему:

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

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

Явное задание схемы данных имеет несколько значительных преимуществ

• Ускорение чтения данных. Когда схема данных задана явно, Spark не тратит время на анализ структуры данных. Это сокращает время чтения данных, особенно при работе с большими файлами.

• Оптимизация плана выполнения. Явно заданная схема позволяет Spark лучше оптимизировать план выполнения, так как известны точные типы данных и структура таблиц. Это улучшает распределение задач и использование ресурсов.

• Предотвращение ошибок. С явно заданной схемой уменьшается риск ошибок, связанных с неправильным определением типов данных, что делает обработку данных более надёжной.

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

Рассмотрим пример чтения CSV-файла с явным заданием схемы данных.

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, IntegerType

# Создание SparkSession
spark = SparkSession.builder.appName("SchemaExample").getOrCreate()

# Определение схемы данных
schema = StructType([
    StructField("name", StringType(), True),
    StructField("age", IntegerType(), True),
    StructField("city", StringType(), True)
])

# Чтение CSV-файла с явно заданной схемой
df = spark.read.schema(schema).csv("path/to/file.csv")

# Показ первых 5 строк
df.show(5)

В этом примере:

• мы создаём SparkSession для работы с данными;

• определяем схему данных с помощью StructType и StructField;

• читаем CSV-файл с использованием явно заданной схемы;

• отображаем первые 5 строк для проверки.

Кэширование и персистенция

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

Когда использовать cache(), а когда persist(), и как?

Использование cache():

• Когда требуется повторное использование одного и того же набора данных в нескольких последующих действиях (actions).

• Подходит для часто используемых данных, так как кэширование хранит данные в памяти (RAM), что ускоряет доступ.

Использование persist():

• Когда необходимо контролировать уровень хранения данных (например, в памяти, на диске или комбинации).

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

Различия между cache() и persist():

• cache(): эквивалентно вызову persist(StorageLevel.MEMORY_ONLY), хранит данные только в памяти.

• persist(): позволяет выбирать различные уровни хранения (Storage Levels) в зависимости от потребностей.

Уровни хранения:

• MEMORY_ONLY. Хранит данные только в памяти. Быстро, но требует много памяти.

• MEMORY_AND_DISK. Хранит данные в памяти, а если памяти недостаточно, сохраняет на диск.

• DISK_ONLY. Хранит данные только на диске. Медленнее, но экономит память.

• MEMORY_ONLY_SER. Хранит данные в памяти в сериализованном виде. Экономит память, но увеличивает затраты на сериализацию/десериализацию.

• MEMORY_AND_DISK_SER. Комбинация MEMORY_ONLY_SER и DISK_ONLY.

Пример с использованием cache().

from pyspark.sql import SparkSession

# Создание SparkSession
spark = SparkSession.builder.appName("CacheExample").getOrCreate()

# Чтение данных из CSV-файла
df = spark.read.csv("path/to/file.csv", header=True, inferSchema=True)

# Кэширование DataFrame
df.cache()

# Первое действие: подсчет строк
count = df.count()
print(f"Total count: {count}")

# Второе действие: фильтрация данных
filtered_df = df.filter(df['age'] > 30)
filtered_count = filtered_df.count()
print(f"Filtered count: {filtered_count}")

# Отключение кэширования
df.unpersist()

# Остановка SparkSession
spark.stop()

Пример с использованием persist().

from pyspark.sql import SparkSession
from pyspark.storagelevel import StorageLevel

# Создание SparkSession
spark = SparkSession.builder.appName("PersistExample").getOrCreate()

# Чтение данных из CSV-файла
df = spark.read.csv("path/to/file.csv", header=True, inferSchema=True)

# Персистенция DataFrame с использованием уровня MEMORY_AND_DISK
df.persist(StorageLevel.MEMORY_AND_DISK)

# Первое действие: подсчет строк
count = df.count()
print(f"Total count: {count}")

# Второе действие: фильтрация данных
filtered_df = df.filter(df['age'] > 30)
filtered_count = filtered_df.count()
print(f"Filtered count: {filtered_count}")

# Отключение персистенции
df.unpersist()

# Остановка SparkSession
spark.stop()

Примеры, показывающие разницу в производительности.

from pyspark.sql import SparkSession

# Создание SparkSession
spark = SparkSession.builder.appName("NoCacheExample").getOrCreate()

# Чтение данных из CSV-файла
df = spark.read.csv("path/to/file.csv", header=True, inferSchema=True)

# Первое действие: подсчет строк (данные читаются заново)
count = df.count()
print(f"Total count: {count}")

# Второе действие: фильтрация данных и подсчет строк (данные читаются заново)
filtered_df = df.filter(df['age'] > 30)
filtered_count = filtered_df.count()
print(f"Filtered count: {filtered_count}")

# Остановка SparkSession
spark.stop()

С кэшированием.

from pyspark.sql import SparkSession

# Создание SparkSession
spark = SparkSession.builder.appName("CacheExample").getOrCreate()

# Чтение данных из CSV-файла
df = spark.read.csv("path/to/file.csv", header=True, inferSchema=True)

# Кэширование DataFrame
df.cache()

# Первое действие: подсчет строк (данные читаются и кэшируются)
count = df.count()
print(f"Total count: {count}")

# Второе действие: фильтрация данных и подсчет строк (данные берутся из кэша)
filtered_df = df.filter(df['age'] > 30)
filtered_count = filtered_df.count()
print(f"Filtered count: {filtered_count}")

# Отключение кэширования
df.unpersist()

# Остановка SparkSession
spark.stop()

Разница в производительности.

• Без кэширования/персистенции. Для каждого действия (например, count или show) Spark заново читает данные из исходного файла, что приводит к значительным временным затратам.

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

Управление партициями

Управление партициями данных в Apache Spark — важная часть оптимизации производительности. Правильное разбиение данных на партиции позволяет эффективнее использовать ресурсы и ускоряет выполнение задач.

Разбиение данных на партиции

Когда данные загружаются в Spark, они автоматически разбиваются на партиции. 

Партиция — это логическая единица данных, которая обрабатывается одним узлом (node) в кластере Spark. Количество партиций по умолчанию зависит от источника данных и конфигурации Spark.

Как правильно настроить количество партиций?

Настройка количества партиций зависит от объёма данных и доступных ресурсов в кластере. Основные рекомендации включают:

• Баланс между партициями. Партиции должны быть примерно одинакового размера, чтобы равномерно распределить нагрузку между узлами.

• Число партиций. Обычное правило — иметь примерно 2-4 партиции на каждый CPU в кластере. Например, если у вас есть кластер с 8 узлами, каждый из которых имеет 4 CPU, то общее число партиций может быть от 64 до 128.

• Размер партиции. Оптимальный размер обычно составляет от 128 MB до 1 GB.

Использование repartition и coalesce

Spark предоставляет методы repartition и coalesce для изменения количества партиций.

• Метод repartition используется для увеличения или уменьшения количества партиций. Он выполняет shuffle данных, что может быть дорогостоящей операцией, но обеспечивает равномерное распределение данных между партициями.

• Метод coalesce используется для уменьшения количества партиций. Он выполняет это без shuffle данных, что делает его менее затратным по сравнению с repartition. Однако coalesce эффективен только для уменьшения количества партиций.

Пример разбиения данных на партиции при загрузке.

from pyspark.sql import SparkSession

# Создание SparkSession
spark = SparkSession.builder.appName("PartitioningExample").getOrCreate()

# Чтение данных из CSV-файла (по умолчанию разбивается на партиции)
df = spark.read.csv("path/to/file.csv", header=True, inferSchema=True)

# Проверка количества партиций
print(f"Number of partitions: {df.rdd.getNumPartitions()}")

Пример настройки количества партиций при загрузке данных.

# Загрузка данных с указанием количества партиций
df = spark.read.csv("path/to/file.csv", header=True, inferSchema=True).repartition(100)

# Проверка количества партиций
print(f"Number of partitions after repartition: {df.rdd.getNumPartitions()}")

repartition.

# Увеличение количества партиций до 100
df_repartitioned = df.repartition(100)

# Проверка количества партиций после repartition
print(f"Number of partitions after repartition: {df_repartitioned.rdd.getNumPartitions()}")

coalesce.

# Уменьшение количества партиций до 10
df_coalesced = df.coalesce(10)

# Проверка количества партиций после coalesce
print(f"Number of partitions after coalesce: {df_coalesced.rdd.getNumPartitions()}")

Пример использования repartition для увеличения числа партиций.

from pyspark.sql import SparkSession

# Создание SparkSession
spark = SparkSession.builder.appName("RepartitionExample").getOrCreate()

# Чтение данных из CSV-файла
df = spark.read.csv("path/to/file.csv", header=True, inferSchema=True)

# Первоначальное количество партиций
print(f"Initial number of partitions: {df.rdd.getNumPartitions()}")

# Увеличение количества партиций до 100
df_repartitioned = df.repartition(100)

# Проверка количества партиций после repartition
print(f"Number of partitions after repartition: {df_repartitioned.rdd.getNumPartitions()}")

# Остановка SparkSession
spark.stop()

Пример использования coalesce для уменьшения числа партиций.

from pyspark.sql import SparkSession

# Создание SparkSession
spark = SparkSession.builder.appName("CoalesceExample").getOrCreate()

# Чтение данных из CSV-файла
df = spark.read.csv("path/to/file.csv", header=True, inferSchema=True)

# Первоначальное количество партиций
print(f"Initial number of partitions: {df.rdd.getNumPartitions()}")

# Уменьшение количества партиций до 10
df_coalesced = df.coalesce(10)

# Проверка количества партиций после coalesce
print(f"Number of partitions after coalesce: {df_coalesced.rdd.getNumPartitions()}")

# Остановка SparkSession
spark.stop()

№2. Оптимизация запросов

Оптимизация запросов в Apache Spark включает выбор правильных операций и понимание принципов, таких как ленивые вычисления. Давайте рассмотрим разницу между трансформациями (например, map, filter) и действиями (например, count, collect), а также то, как правильно выбирать операции для повышения производительности.

Выбор правильных операций

В Spark все операции можно разделить на две категории: трансформации (transformations) и действия (actions).

Трансформации

Это операции, которые создают новое распределенное множество данных (RDD) из существующего, но не выполняют никаких вычислений сразу. Трансформации являются ленивыми. Означает, что они не выполняются до тех пор, пока не будет вызвано действие.

Пример трансформации map.

rdd = sc.parallelize([1, 2, 3, 4, 5])
squared_rdd = rdd.map(lambda x: x * x)

filter, flatMap, reduceByKey, groupByKey — это всё также трансформации.

Действия

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

Примерами действий являются count, collect, take, saveAsTextFile.

Пример действия count.

count = squared_rdd.count()
print(f"Number of elements: {count}")

Принцип ленивых вычислений

Ленивые вычисления (lazy evaluation) — это ключевой принцип Spark, который откладывает выполнение трансформаций до тех пор, пока не будет вызвано действие. Это позволяет Spark оптимизировать план выполнения, объединяя трансформации и минимизируя количество проходов по данным.

Пример.

# Создание SparkSession
spark = SparkSession.builder.appName("LazyEvaluationExample").getOrCreate()

# Чтение данных из CSV-файла
df = spark.read.csv("path/to/file.csv", header=True, inferSchema=True)

# Трансформации (ленивые вычисления)
filtered_df = df.filter(df['age'] > 30)
selected_df = filtered_df.select("name", "age")

# Действие (вызывает выполнение всех предыдущих трансформаций)
result = selected_df.collect()

# Печать результатов
for row in result:
    print(row)

# Остановка SparkSession
spark.stop()

Как выбрать правильные операции? 

Использовать трансформации вместо действий.

По возможности, избегайте вызова действий, которые собирают данные на драйвере (например, collect) на больших наборах данных, так как это может привести к перегрузке памяти. Вместо этого, используйте трансформации, чтобы минимизировать объём данных до вызова действия.

Плохой пример: collect на больших данных.

large_rdd = sc.parallelize(range(1000000))
collected_data = large_rdd.collect()  # Может привести к переполнению памяти

Хороший пример: уменьшение данных перед collect.

filtered_rdd = large_rdd.filter(lambda x: x % 2 == 0)
small_collected_data = filtered_rdd.take(10)  # Безопаснее, так как собирается небольшой объём данных.

Использование map и flatMap

Используйте map и flatMap для преобразования данных:

• map применяется к каждому элементу RDD и возвращает новый RDD того же размера;

• flatMap может возвращать RDD различного размера.

Пример использования map.

rdd = sc.parallelize(["apple", "banana", "cherry"])
length_rdd = rdd.map(lambda x: len(x))
print(length_rdd.collect())

Пример использования flatMap.

words_rdd = rdd.flatMap(lambda x: x.split("a"))
print(words_rdd.collect())

Использование reduceByKey и groupByKey

Для агрегации данных используйте reduceByKey вместо groupByKey, так как reduceByKey агрегирует данные локально на каждом узле перед пересылкой по сети, что уменьшает объём передаваемых данных.

Пример использования reduceByKey.

pairs = sc.parallelize([("a", 1), ("b", 1), ("a", 1)])
reduced_pairs = pairs.reduceByKey(lambda x, y: x + y)
print(reduced_pairs.collect())

Пример использования groupByKey (менее эффективно).

grouped_pairs = pairs.groupByKey()
print([(x, list(y)) for x, y in grouped_pairs.collect()])

Пример оптимизации запроса. 

Рассмотрим пример, где мы хотим вычислить средний возраст пользователей старше 30 лет из CSV-файла.

# Создание SparkSession
spark = SparkSession.builder.appName("QueryOptimizationExample").getOrCreate()

# Чтение данных из CSV-файла
df = spark.read.csv("path/to/file.csv", header=True, inferSchema=True)

# Оптимизация запроса
# Трансформации
filtered_df = df.filter(df['age'] > 30)
age_sum = filtered_df.groupBy().sum("age")
age_count = filtered_df.groupBy().count()

# Действие: выполнение всех трансформаций и получение результата
sum_age = age_sum.collect()[0][0]
count_age = age_count.collect()[0][0]

average_age = sum_age / count_age
print(f"Average age: {average_age}")

# Остановка SparkSession
spark.stop()

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

Порядок выполнения операций

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

Почему важно выполнять фильтрацию и агрегацию до соединений?

• Сокращение объёма данных: Фильтрация и агрегация уменьшают количество данных, которые нужно обрабатывать при последующем соединении. Это снижает затраты на передачу данных по сети и уменьшает объём данных, участвующих в операции соединения.

• Уменьшение ресурсов: Меньшие объёмы данных требуют меньше ресурсов для хранения и обработки, что позволяет более эффективно использовать вычислительные ресурсы.

• Сокращение времени выполнения: Обработка меньших объёмов данных быстрее и эффективнее, что сокращает общее время выполнения запроса.

Пример выполнения фильтрации и агрегации до соединений.

from pyspark.sql import SparkSession

# Создание SparkSession
spark = SparkSession.builder.appName("OptimizationExample").getOrCreate()

# Пример данных
data1 = [("Alice", 34, "HR"), ("Bob", 45, "IT"), ("Charlie", 29, "HR"), ("David", 40, "Finance")]
data2 = [("HR", "Human Resources"), ("IT", "Information Technology"), ("Finance", "Financial Department")]

df1 = spark.createDataFrame(data1, ["name", "age", "dept"])
df2 = spark.createDataFrame(data2, ["dept", "department_name"])

# Фильтрация до соединения
filtered_df1 = df1.filter(df1['age'] > 30)

# Соединение
joined_df = filtered_df1.join(df2, filtered_df1.dept == df2.dept)

# Показ результатов
joined_df.show()

# Остановка SparkSession
spark.stop()

В этом примере сначала фильтруются данные df1, чтобы оставить только записи с возрастом больше 30. Только после этого выполняется соединение с df2. Такая последовательность позволяет уменьшить объём данных, участвующих в соединении, и улучшает производительность.

Стратегии построения запросов от меньших таблиц к большим

Почему это важно?

• Эффективное использование памяти. Начало с меньших таблиц помогает избежать проблем с нехваткой памяти, так как меньшие объёмы данных проще обрабатывать и кэшировать.

• Уменьшение количества shuffle. Соединение меньших таблиц до более крупных уменьшает объём shuffle данных, что улучшает производительность сети и общее время выполнения.

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

Пример стратегии построения запросов.

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

# Создание SparkSession
spark = SparkSession.builder.appName("JoinOptimizationExample").getOrCreate()

# Пример данных
data_small = [("Alice", 34, "HR"), ("Bob", 45, "IT")]
data_medium = [("HR", "Human Resources"), ("IT", "Information Technology"), ("Finance", "Financial Department")]
data_large = [("HR", 1), ("IT", 2), ("Finance", 3), ("HR", 4), ("IT", 5), ("Finance", 6)]

df_small = spark.createDataFrame(data_small, ["name", "age", "dept"])
df_medium = spark.createDataFrame(data_medium, ["dept", "department_name"])
df_large = spark.createDataFrame(data_large, ["dept", "id"])

# Соединение маленькой таблицы с средней
joined_small_medium = df_small.join(df_medium, "dept")

# Фильтрация после первого соединения
filtered_join = joined_small_medium.filter(joined_small_medium['age'] > 30)

# Соединение с большой таблицей
final_join = filtered_join.join(df_large, "dept")

# Показ результатов
final_join.show()

# Остановка SparkSession
spark.stop()

В этом примере:

• Соединение маленькой таблицы со средней. Сначала соединяем df_small и df_medium. Это небольшие таблицы, поэтому операция соединения выполняется быстро и требует меньше ресурсов.

• Фильтрация после первого соединения. После первого соединения выполняем фильтрацию, оставляя только записи с возрастом больше 30. Это ещё больше уменьшает объём данных.

• Соединение с большой таблицей. Только после уменьшения объёма данных соединяем результат с большой таблицей df_large.

Различные стратегии для уменьшения данных до соединения

• Использование агрегации. Выполнение агрегации до соединения, чтобы уменьшить объём данных.

• Использование транзитивных свойств. Например, если у вас есть три таблицы и вам нужно соединить их, сначала соединяйте меньшие таблицы.

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

Пример использования агрегации перед соединением.

# Создание SparkSession
spark = SparkSession.builder.appName("AggregationBeforeJoinExample").getOrCreate()

# Пример данных
data1 = [("Alice", 34, "HR"), ("Bob", 45, "IT"), ("Charlie", 29, "HR"), ("David", 40, "Finance"), ("Eve", 50, "IT")]
data2 = [("HR", "Human Resources"), ("IT", "Information Technology"), ("Finance", "Financial Department")]

df1 = spark.createDataFrame(data1, ["name", "age", "dept"])
df2 = spark.createDataFrame(data2, ["dept", "department_name"])

# Агрегация до соединения (средний возраст по департаментам)
agg_df1 = df1.groupBy("dept").avg("age").alias("avg_age")

# Соединение с таблицей департаментов
joined_df = agg_df1.join(df2, "dept")

# Показ результатов
joined_df.show()

# Остановка SparkSession
spark.stop()

В этом примере мы сначала выполняем агрегацию для вычисления среднего возраста по департаментам, а затем соединяем результаты с таблицей департаментов. Это уменьшает объём данных, участвующих в соединении, и улучшает производительность.

Заключение

Надеюсь, эта статья помогла вам разобраться в основах оптимизации производительности Apache Spark и ответила на некоторые ваши вопросы. Удачи в работе с данными!