PySpark в масштабе

Введение в PySpark

Benjamin Schmidt

Data Engineer

Использование масштаба

PySpark эффективно работает с гигабайтами и терабайтами данных
Цель PySpark — высокая скорость и эффективная обработка
Понимание плана выполнения открывает дополнительные возможности оптимизации
Используйте broadcast для управления всем кластером

joined_df = large_df.join(broadcast(small_df), 
                          on="key_column", how="inner")
joined_df.show()

Планы выполнения

# Using explain() to view the execution plan
df.filter(df.Age > 40).select("Name").explain()

== Physical Plan ==
*(1) Filter (isnotnull(Age) AND (Age > 30))
+- Scan ExistingRDD[Name:String, Age:Int]

¹ https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.explain.html

Кэширование и персистентность DataFrame

Кэширование: хранит данные в памяти для быстрого доступа — подходит для небольших наборов данных
Персистентность: хранит данные на разных уровнях хранилища — для крупных наборов данных

df = spark.read.csv("large_dataset.csv", header=True, inferSchema=True)

# Cache the DataFrame
df.cache()


# Perform multiple operations on the cached DataFrame
df.filter(df["column1"] > 50).show()
df.groupBy("column2").count().show()

Персистентность DataFrame с разными уровнями хранилища

# Persist the DataFrame with storage level
from pyspark import StorageLevel

df.persist(StorageLevel.MEMORY_AND_DISK)


# Perform transformations
result = df.groupBy("column3").agg({"column4": "sum"})
result.show()

# Unpersist after use
df.unpersist()

Оптимизация PySpark

Малые подвыборки: чем больше данных, тем медленнее операция — выбирайте map() вместо groupby() из-за большей избирательности
Broadcast-соединения: broadcast задействует все вычислительные ресурсы, даже на небольших наборах данных
Избегайте повторных действий: повторные операции над теми же данными требуют времени и ресурсов, не давая никаких преимуществ

Давайте потренируемся!

Введение в PySpark