Какие знаешь типы операций в Spark?

Типы операций в Apache Spark

Apache Spark — это распределённая система обработки больших данных. Она использует два основных типа операций: Transformations и Actions.

Transformations (Преобразования)

Это ленивые операции, которые создают новый RDD/DataFrame из существующего, но не выполняются сразу. Вычисления откладываются до вызова Action.

Узкие Transformations (Narrow Dependencies)

Каждый элемент выходного RDD зависит только от одного элемента входного RDD. Данные не требуют перекоса между узлами:

# map — преобразует каждый элемент
rdd = sc.parallelize([1, 2, 3, 4])
mapped = rdd.map(lambda x: x * 2)  # [2, 4, 6, 8]

# filter — фильтрует элементы по условию
filtered = rdd.filter(lambda x: x > 2)  # [3, 4]

# flatMap — маппирует и сплющивает результат
flattened = rdd.flatMap(lambda x: [x, x*2])  # [1, 2, 2, 4, 3, 6, 4, 8]

# mapPartitions — применяет функцию к целой партиции
def process_partition(partition):
    return [x * 2 for x in partition]

rdd.mapPartitions(process_partition)

Плюсы: быстрые, не требуют шафла данных, хорошо для конвейеров.

Широкие Transformations (Wide Dependencies)

Выходной элемент зависит от нескольких входных элементов. Требуют перемешивание (shuffle) данных между узлами:

# groupByKey — группирует по ключу
rdd = sc.parallelize([("a", 1), ("b", 1), ("a", 2)])
grouped = rdd.groupByKey()  # [("a", [1, 2]), ("b", [1])]

# reduceByKey — агрегирует по ключу
result = rdd.reduceByKey(lambda a, b: a + b)  # [("a", 3), ("b", 1)]

# join — объединяет два RDD по ключу
rdd1 = sc.parallelize([("a", 1), ("b", 2)])
rdd2 = sc.parallelize([("a", 3), ("b", 4)])
joined = rdd1.join(rdd2)  # [("a", (1, 3)), ("b", (2, 4))]

# distinct — удаляет дубликаты
rdd = sc.parallelize([1, 1, 2, 2, 3])
unique = rdd.distinct()  # [1, 2, 3]

# repartition — изменяет число партиций (требует shuffle)
repartitioned = rdd.repartition(4)

Минусы: медленнее из-за shuffle, требуют сетевого трафика.

Actions (Действия)

Это материализующие операции, которые запускают фактическое вычисление DAG (Directed Acyclic Graph) и возвращают результат в драйвер или запись на диск:

# collect — возвращает все элементы в драйвер (опасно на больших данных!)
result = rdd.collect()  # [1, 2, 3, 4]

# first — возвращает первый элемент
first_elem = rdd.first()  # 1

# count — количество элементов
count = rdd.count()  # 4

# take(n) — первые n элементов
first_three = rdd.take(3)  # [1, 2, 3]

# reduce — агрегирует весь RDD в один элемент
sum_result = rdd.reduce(lambda a, b: a + b)  # 10

# saveAsTextFile — сохраняет в HDFS/S3
rdd.saveAsTextFile("hdfs://path/to/output")

# foreach — выполняет функцию для каждого элемента (обычно для side effects)
rdd.foreach(lambda x: print(x))  # Выполняется на воркерах, не в драйвере!

# max, min, mean, sum — статистические операции
rdd.max()  # 4
rdd.min()  # 1
rdd.mean()  # 2.5

DataFrame Operations

В современном Spark используют DataFrames, которые имеют оптимизированный Catalyst optimizer:

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("example").getOrCreate()

# Создание DataFrame
df = spark.createDataFrame(
    [(1, "Alice"), (2, "Bob")],
    ["id", "name"]
)

# Transformations (ленивые)
df_filtered = df.filter(df.id > 1)
df_selected = df.select("name")
df_grouped = df.groupBy("name").count()

# Actions (выполняют вычисления)
df_filtered.show()  # Выводит результат
df_filtered.collect()  # Возвращает в драйвер
df_filtered.write.csv("output_path")  # Сохраняет

Ключевые характеристики

Best Practices

• Кэширование: используй rdd.cache() или df.cache() для повторного использования
• Избегай collect() на больших датасетах — может вызвать OutOfMemory
• Prefer DataFrames over RDD — лучше оптимизируются
• Используй Spark SQL вместо RDD операций — часто быстрее благодаря Catalyst
• Minimise wide transformations — они требуют shuffle и медленнее

Понимание этого различия критично для написания эффективного Spark кода!