PySpark EssentialsPySpark: основы

PySpark is the Python API for Apache Spark, a distributed engine for processing data that's too big for one machine. You write Python; Spark splits the work across many machines (a cluster) and runs it in parallel. The main thing you touch is the DataFrame: a distributed table with named, typed columns — like a pandas DataFrame, but spread over the cluster and computed lazily.

PySpark — это Python-API для Apache Spark, движка распределённой обработки данных, которые не помещаются на одной машине. Ты пишешь Python; Spark разбивает работу по множеству машин (кластер) и выполняет параллельно. Главное, с чем работаешь — DataFrame: распределённая таблица с именованными типизированными колонками — как pandas DataFrame, но размазанная по кластеру и вычисляемая лениво.

This is the single most important PySpark idea, and a near-guaranteed interview question. Operations come in two kinds:

Это самая важная идея PySpark и почти гарантированный вопрос на интервью. Операции бывают двух видов:

Transformations (lazy)	Actions (eager)
Describe a new DataFrame; nothing runs yet	Trigger actual computation
select, filter, withColumn, join, groupBy, orderBy	show, collect, count, write, take
Just builds a logical plan (a DAG)	Submits a job, Spark runs the plan

Трансформации (ленивые)	Действия (немедленные)
Описывают новый DataFrame; пока ничего не выполняется	Запускают реальное вычисление
select, filter, withColumn, join, groupBy, orderBy	show, collect, count, write, take
Просто строит логический план (DAG)	Отправляет job, Spark выполняет план

# nothing runs here — just building a plan
df2 = (df
    .filter(df.country == "DE")
    .withColumn("amount_eur", df.amount * 1.0)
    .select("user_id", "amount_eur"))

df2.show()   # ACTION → now Spark optimizes & executes the whole chain

The SparkSession is your entry point — the handle to the cluster. Everything starts from spark.

SparkSession — твоя точка входа, ручка к кластеру. Всё начинается с spark.

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

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

# create from Python data (handy for tests)
df = spark.createDataFrame(
    [(1, "Amal", "DE"), (2, "Tanya", "UK")],
    ["id", "name", "country"])

df.printSchema()   # inspect column names & types
df.show()          # pretty-print first 20 rows

schema = StructType([
    StructField("id", IntegerType(), False),
    StructField("name", StringType(), True),
    StructField("country", StringType(), True),
])
df = spark.read.schema(schema).csv("s3://bucket/users.csv", header=True)

Three ways to reference a column: "name" (string), df.name (attribute), F.col("name") (function). Use F.col when you need expressions.

Три способа сослаться на колонку: "name" (строка), df.name (атрибут), F.col("name") (функция). Используй F.col, когда нужны выражения.

# select & rename
df.select("id", F.col("name").alias("user_name"))

# filter (where == filter)
df.filter((F.col("country") == "DE") & (F.col("id") > 0))

# add / replace a column
df.withColumn("is_de", F.col("country") == "DE")

# conditional logic
df.withColumn("tier",
    F.when(F.col("amount") > 100, "hi")
     .when(F.col("amount") > 10, "mid")
     .otherwise("lo"))

# drop, distinct, dedup, sort, limit
df.drop("tmp").dropDuplicates(["id"]).orderBy(F.col("amount").desc()).limit(100)

# nulls
df.na.fill({"amount": 0}).na.drop(subset=["id"])

# aggregation
(df.groupBy("country")
   .agg(
       F.count("*").alias("n"),
       F.sum("amount").alias("total"),
       F.avg("amount").alias("avg_amt"),
       F.countDistinct("user_id").alias("users"),
   ))

Joins — know the how values and the duplicate-column gotcha:

Join — знай значения how и подвох с дублирующимися колонками:

# join on a shared column name → no duplicate "country" column
orders.join(users, on="user_id", how="left")

# join on a condition → both id columns survive (ambiguous!)
orders.join(users, orders.user_id == users.id, "inner")

# how: inner | left | right | outer | left_semi | left_anti
# left_anti = rows in orders with NO match in users (great for "missing")

from pyspark.sql.window import Window
w = Window.partitionBy("user_id").orderBy(F.col("ts").desc())
latest = (df
    .withColumn("rn", F.row_number().over(w))
    .filter("rn = 1")
    .drop("rn"))

# read
df = spark.read.parquet("s3://bucket/events/")
df = spark.read.option("header", True).csv("path.csv")
df = spark.read.json("path.json")

# write — note mode & partitioning
(df.write
   .mode("overwrite")            # overwrite | append | ignore | error
   .partitionBy("ds")            # physical folder per partition value
   .parquet("s3://bucket/out/"))

They're the same engine — pick whichever reads cleaner. Both go through Catalyst, so performance is identical.

Это один движок — выбирай, что читается чище. Оба идут через Catalyst, поэтому производительность одинакова.

# register a temp view, then write plain SQL
df.createOrReplaceTempView("orders")
top = spark.sql("""
    SELECT country, SUM(amount) AS total
    FROM orders
    WHERE amount > 0
    GROUP BY country
    ORDER BY total DESC
""")

Term	Plain meaning
SparkSession	Entry point / handle to the cluster (spark).
DataFrame	Distributed table with named, typed columns.
Transformation	Lazy op that builds a plan (select, filter, join).
Action	Eager op that runs the plan (show, count, write).
Catalyst	Spark's query optimizer (plans, pushdowns, reorders).
Partition	A chunk of the data processed by one task.
Driver	Coordinator process running your code & the plan.
Executor	Worker process that runs tasks on partitions.

Термин	Простой смысл
SparkSession	Точка входа / ручка к кластеру (spark).
DataFrame	Распределённая таблица с именованными типизированными колонками.
Трансформация	Ленивая операция, строящая план (select, filter, join).
Действие (action)	Немедленная операция, запускающая план (show, count, write).
Catalyst	Оптимизатор запросов Spark (планы, pushdown, перестановки).
Партиция	Кусок данных, обрабатываемый одной задачей (task).
Driver	Процесс-координатор, выполняющий твой код и план.
Executor	Рабочий процесс, выполняющий задачи на партициях.