apache-spark-data-processing

Apache Spark 数据处理

全面掌握 Apache Spark 数据处理的技能，涵盖从基础 RDD 操作到高级流处理、SQL 和机器学习工作流。学习构建可扩展的分布式数据管道和分析系统。

何时使用此技能

当你需要以下功能时，请使用 Apache Spark：

• 处理大规模数据：处理大到无法在单台机器上处理的数据集（TB 到 PB 级）
• 执行分布式计算：在集群节点间并行执行计算
• 实时流处理：以低延迟处理连续数据流
• 复杂数据分析：运行复杂的分析、聚合和转换
• 大规模机器学习：在海量数据集上训练 ML 模型
• ETL/ELT 管道：构建可靠的数据转换和加载工作流
• 交互式数据分析：对大型数据集进行探索性分析
• 统一数据处理：在一个框架中结合批处理和流处理工作负载

不适合以下场景：

• 可在单台机器内存中处理的小型数据集（<100 GB）
• 简单的 CRUD 操作（使用传统数据库）
• 对延迟要求极低（<10ms）、适合使用专用流处理器的场景
• 需要在分布式数据间进行强 ACID 事务的工作流

核心概念

弹性分布式数据集（RDD）

RDD 是 Spark 的基础数据抽象——不可变、分布式的对象集合，可并行处理。

主要特性：

• 弹性：通过跟踪血缘关系实现容错
• 分布式：分区存储在集群节点上
• 不可变：转换会创建新的 RDD，而非修改现有 RDD
• 惰性求值：转换会构建计算图；操作会触发执行
• 内存计算：缓存中间结果，用于迭代算法

RDD 操作：

• 转换：返回新 RDD 的惰性操作（map、filter、flatMap、reduceByKey）
• 操作：触发计算并返回值的操作（collect、count、reduce、saveAsTextFile）

适合使用 RDD 的场景：

• 需要对数据分布和分区进行底层控制
• 需要自定义分区方案
• 处理非结构化数据（文本文件、二进制数据）
• 从早期 Spark 版本迁移旧代码

尽可能优先使用 DataFrame/Dataset——它们可通过 Catalyst 优化器自动优化。

DataFrame 和 Dataset

DataFrame 是按命名列组织的分布式数据集合，类似数据库表或 pandas DataFrame，但具有强大的优化能力。

DataFrame：

• 具有架构的结构化数据
• 自动查询优化（Catalyst）
• 支持跨语言使用（Python、Scala、Java、R）
• 提供丰富的类 SQL 操作 API

Dataset（仅限 Scala/Java）：

• 具有编译时类型安全的类型化 DataFrame
• 由于 JVM 优化，在 Scala 中性能最佳
• 结合 RDD 的类型安全性与 DataFrame 的优化能力

相比 RDD 的主要优势：

• 查询优化：Catalyst 优化器会重写查询以提升效率
• Tungsten 执行：优化 CPU 和内存使用
• 列式存储：高效的数据表示方式
• 代码生成：编译时生成字节码，以加快执行速度

惰性求值

Spark 使用惰性求值来优化执行：

1. 转换会构建操作的有向无环图（DAG）
2. 操作会触发 DAG 的执行
3. Spark 优化器会分析整个 DAG，并创建优化后的执行计划
4. 工作会分发到各个集群节点

优势：

• 尽可能减少网络中的数据移动
• 将多个操作合并到单个阶段
• 消除不必要的计算
• 优化内存使用

分区

数据会划分为多个分区，以便并行处理：

• 默认分区：通常依据 HDFS 块大小或输入源
• 哈希分区：按键的哈希值分发数据（用于 groupByKey、reduceByKey）
• 范围分区：按键的范围分发数据（适用于已排序的数据）
• 自定义分区：定义自己的分区逻辑

分区数量注意事项：

• 分区过少：集群利用率低，任务执行时间长
• 分区过多：调度开销大，任务执行时间短
• 一般规则：集群中每个 CPU 核心对应 2-4 个分区
• 使用 repartition() 或 coalesce() 调整分区数量

缓存和持久化

将经常访问的数据缓存在内存中，以提升性能：

# Cache DataFrame in memory
df.cache()  # Shorthand for persist(StorageLevel.MEMORY_AND_DISK)

# Different storage levels
df.persist(StorageLevel.MEMORY_ONLY)       # Fast but may lose data if evicted
df.persist(StorageLevel.MEMORY_AND_DISK)   # Spill to disk if memory full
df.persist(StorageLevel.DISK_ONLY)         # Store only on disk
df.persist(StorageLevel.MEMORY_ONLY_SER)   # Serialized in memory (more compact)

# Unpersist when done
df.unpersist()

适合缓存的情况：

• 工作流中会多次使用的数据
• 迭代算法（ML 训练）
• 交互式分析会话
• 下游会重复使用的高开销转换结果

不适合缓存的情况：

• 只使用一次的数据
• 超出集群内存容量的超大型数据集
• 持续产生新数据的流式应用

Spark SQL

Spark SQL 支持使用 SQL 或 DataFrame API 查询结构化数据：

• 在 DataFrame 和表上执行 SQL 查询
• 将 DataFrame 注册为临时视图
• 连接结构化和半结构化数据
• 连接到 Hive 元存储以获取表元数据
• 支持多种数据源（Parquet、ORC、JSON、CSV、JDBC）

性能特性：

• Catalyst Optimizer：基于规则和成本的查询优化
• Tungsten Execution Engine：全阶段代码生成、向量化处理
• Adaptive Query Execution (AQE)：根据统计信息进行运行时优化
• Dynamic Partition Pruning：执行期间跳过无关分区

广播变量和累加器

用于高效分布式计算的共享变量：

广播变量：

• 在每个节点上缓存的只读变量
• 适合共享大型只读数据（查找表、ML 模型）
• 避免在每个任务中发送大型数据

# Broadcast a lookup table
lookup_table = {"key1": "value1", "key2": "value2"}
broadcast_lookup = sc.broadcast(lookup_table)

# Use in transformations
rdd.map(lambda x: broadcast_lookup.value.get(x, "default"))

累加器：

• 用于跨任务聚合值的只写变量
• 用于分布式操作中的计数器和求和
• 只有驱动程序可以读取最终累加值

# Create accumulator
error_count = sc.accumulator(0)

# Increment in tasks
rdd.foreach(lambda x: error_count.add(1) if is_error(x) else None)

# Read final value in driver
print(f"Total errors: {error_count.value}")

深入了解 Spark SQL

创建 DataFrame

从各种来源创建 DataFrame：

from pyspark.sql import SparkSession

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

# From structured data
data = [("Alice", 1), ("Bob", 2), ("Charlie", 3)]
columns = ["name", "id"]
df = spark.createDataFrame(data, columns)

# From files
df_json = spark.read.json("path/to/file.json")
df_parquet = spark.read.parquet("path/to/file.parquet")
df_csv = spark.read.option("header", "true").csv("path/to/file.csv")

# From JDBC sources
df_jdbc = spark.read \
    .format("jdbc") \
    .option("url", "jdbc:postgresql://host:port/database") \
    .option("dbtable", "table_name") \
    .option("user", "username") \
    .option("password", "password") \
    .load()

DataFrame 操作

常见的 DataFrame 转换：

# Select columns
df.select("name", "age").show()

# Filter rows
df.filter(df.age > 21).show()
df.where(df["age"] > 21).show()  # Alternative syntax

# Add/modify columns
from pyspark.sql.functions import col, lit
df.withColumn("age_plus_10", col("age") + 10).show()
df.withColumn("country", lit("USA")).show()

# Aggregations
df.groupBy("department").count().show()
df.groupBy("department").agg({"salary": "avg", "age": "max"}).show()

# Sorting
df.orderBy("age").show()
df.orderBy(col("age").desc()).show()

# Joins
df1.join(df2, df1.id == df2.user_id, "inner").show()
df1.join(df2, "id", "left_outer").show()

# Unions
df1.union(df2).show()

SQL 查询

在 DataFrame 上执行 SQL：

# Register DataFrame as temporary view
df.createOrReplaceTempView("people")

# Run SQL queries
sql_result = spark.sql("SELECT name FROM people WHERE age > 21")
sql_result.show()

# Complex queries
result = spark.sql("""
    SELECT
        department,
        COUNT(*) as employee_count,
        AVG(salary) as avg_salary,
        MAX(age) as max_age
    FROM people
    WHERE age > 25
    GROUP BY department
    HAVING COUNT(*) > 5
    ORDER BY avg_salary DESC
""")
result.show()

数据源

Spark SQL 支持多种数据格式：

Parquet（推荐用于分析）：

• 列式存储格式
• 压缩效果出色，查询性能优异
• 文件中嵌入架构
• 支持谓词下推和列裁剪

# Write
df.write.parquet("output/path", mode="overwrite", compression="snappy")

# Read with partition pruning
df = spark.read.parquet("output/path").filter(col("date") == "2025-01-01")

ORC（优化行列式存储）：

• 与 Parquet 类似，压缩效果略好
• 首选用于 Hive 集成
• 内置索引，可加快查询

df.write.orc("output/path", mode="overwrite")
df = spark.read.orc("output/path")

JSON（半结构化数据）：

• 易于阅读，但效率较低
• 读取时推断架构
• 适用于嵌套/复杂数据

# Read with explicit schema
from pyspark.sql.types import StructType, StructField, StringType, IntegerType

schema = StructType([
    StructField("name", StringType(), True),
    StructField("age", IntegerType(), True)
])

df = spark.read.schema(schema).json("data.json")

CSV（旧式/简单数据）：

• 兼容性广，但速度较慢
• 需要推断标题行或指定架构
• 压缩收益极小

df.write.csv("output.csv", header=True, mode="overwrite")
df = spark.read.option("header", "true").option("inferSchema", "true").csv("data.csv")

窗口函数

使用窗口函数进行高级分析：

from pyspark.sql.window import Window
from pyspark.sql.functions import row_number, rank, dense_rank, lag, lead, sum, avg

# Define window specification
window_spec = Window.partitionBy("department").orderBy(col("salary").desc())

# Ranking functions
df.withColumn("rank", rank().over(window_spec)).show()
df.withColumn("row_num", row_number().over(window_spec)).show()
df.withColumn("dense_rank", dense_rank().over(window_spec)).show()

# Aggregate functions over window
df.withColumn("dept_avg_salary", avg("salary").over(window_spec)).show()
df.withColumn("running_total", sum("salary").over(window_spec.rowsBetween(Window.unboundedPreceding, Window.currentRow))).show()

# Offset functions
df.withColumn("prev_salary", lag("salary", 1).over(window_spec)).show()
df.withColumn("next_salary", lead("salary", 1).over(window_spec)).show()

用户定义函数（UDF）

创建自定义转换：

from pyspark.sql.functions import udf
from pyspark.sql.types import StringType, IntegerType

# Python UDF (slower due to serialization overhead)
def categorize_age(age):
    if age < 18:
        return "Minor"
    elif age < 65:
        return "Adult"
    else:
        return "Senior"

categorize_udf = udf(categorize_age, StringType())
df.withColumn("age_category", categorize_udf(col("age"))).show()

# Pandas UDF (vectorized, faster for large datasets)
from pyspark.sql.functions import pandas_udf
import pandas as pd

@pandas_udf(IntegerType())
def square(series: pd.Series) -> pd.Series:
    return series ** 2

df.withColumn("age_squared", square(col("age"))).show()

UDF 性能提示：

• 尽可能使用 Spark 内置函数（始终更快）
• 优先使用 Pandas UDF，而非 Python UDF，以获得更好的性能
• 使用 Scala UDF 以获得最高性能（没有序列化开销）
• 如果会多次使用 DataFrame，请在应用 UDF 前缓存它

转换和操作

常见转换

map：对每个元素应用函数

# RDD
rdd = sc.parallelize([1, 2, 3, 4, 5])
squared = rdd.map(lambda x: x * 2)  # [2, 4, 6, 8, 10]

# DataFrame (use select with functions)
from pyspark.sql.functions import col
df.select(col("value") * 2).show()

filter：选择符合谓词条件的元素

# RDD
rdd.filter(lambda x: x > 2).collect()  # [3, 4, 5]

# DataFrame
df.filter(col("age") > 25).show()

flatMap：映射并展平结果

# RDD - Split text into words
lines = sc.parallelize(["hello world", "apache spark"])
words = lines.flatMap(lambda line: line.split(" "))  # ["hello", "world", "apache", "spark"]

reduceByKey：按键聚合值

# Word count example
words = sc.parallelize(["apple", "banana", "apple", "cherry", "banana", "apple"])
word_pairs = words.map(lambda word: (word, 1))
word_counts = word_pairs.reduceByKey(lambda a, b: a + b)
# Result: [("apple", 3), ("banana", 2), ("cherry", 1)]

groupByKey：按键分组值（尽可能避免使用，改用 reduceByKey）

# Less efficient than reduceByKey
word_pairs.groupByKey().mapValues(list).collect()
# Result: [("apple", [1, 1, 1]), ("banana", [1, 1]), ("cherry", [1])]

join：按键合并数据集

# RDD join
users = sc.parallelize([("user1", "Alice"), ("user2", "Bob")])
orders = sc.parallelize([("user1", 100), ("user2", 200), ("user1", 150)])
users.join(orders).collect()
# Result: [("user1", ("Alice", 100)), ("user1", ("Alice", 150)), ("user2", ("Bob", 200))]

# DataFrame join (more efficient)
df_users.join(df_orders, "user_id", "inner").show()

distinct：删除重复项

# RDD
rdd.distinct().collect()

# DataFrame
df.distinct().show()
df.dropDuplicates(["user_id"]).show()  # Drop based on specific columns

coalesce/repartition：更改分区数量

# Reduce partitions (no shuffle, more efficient)
df.coalesce(1).write.parquet("output")

# Increase/decrease partitions (involves shuffle)
df.repartition(10).write.parquet("output")
df.repartition(10, "user_id").write.parquet("output")  # Partition by column

常见操作

collect：将所有数据提取到驱动程序

results = rdd.collect()  # Returns list
# WARNING: Only use on small datasets that fit in driver memory

count：统计元素数量

total = df.count()  # Number of rows

first/take：获取前 N 个元素

first_elem = rdd.first()
first_five = rdd.take(5)

reduce：聚合所有元素

total_sum = rdd.reduce(lambda a, b: a + b)

foreach：对每个元素执行函数

# Side effects only (no return value)
rdd.foreach(lambda x: print(x))

saveAsTextFile：写入文件系统

rdd.saveAsTextFile("hdfs://path/to/output")

show：显示 DataFrame 行（操作）

df.show(20, truncate=False)  # Show 20 rows, don't truncate columns

结构化流处理

使用 DataFrame API 处理连续数据流。

核心概念

流式 DataFrame：

• 持续增长的无界表
• 支持与批处理 DataFrame 相同的操作
• 微批处理（默认）或连续处理

输入源：

• 文件源（JSON、Parquet、CSV、ORC、文本）
• Kafka
• Socket（用于测试）
• Rate 源（用于测试）
• 自定义源

输出模式：

• Append：仅将新增行添加到结果表
• Complete：每次触发时写入整个结果表
• Update：仅写入更新的行

输出接收器：

• 文件接收器（Parquet、ORC、JSON、CSV、文本）
• Kafka
• Console（用于调试）
• Memory（用于测试）
• Foreach/ForeachBatch（自定义逻辑）

基本流处理示例

from pyspark.sql import SparkSession
from pyspark.sql.functions import col

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

# Read stream from JSON files
input_stream = spark.readStream \
    .format("json") \
    .schema(schema) \
    .option("maxFilesPerTrigger", 1) \
    .load("input/directory")

# Transform streaming data
processed = input_stream \
    .filter(col("value") > 10) \
    .select("id", "value", "timestamp")

# Write stream to Parquet
query = processed.writeStream \
    .format("parquet") \
    .option("path", "output/directory") \
    .option("checkpointLocation", "checkpoint/directory") \
    .outputMode("append") \
    .start()

# Wait for termination
query.awaitTermination()

流式数据与静态数据连接

将流式数据与静态参考数据连接：

# Static DataFrame (loaded once)
static_df = spark.read.parquet("reference/data")

# Streaming DataFrame
streaming_df = spark.readStream.format("kafka").load()

# Inner join (supported)
joined = streaming_df.join(static_df, "type")

# Left outer join (supported)
joined = streaming_df.join(static_df, "type", "left_outer")

# Write result
joined.writeStream \
    .format("parquet") \
    .option("path", "output") \
    .option("checkpointLocation", "checkpoint") \
    .start()

窗口聚合

按时间窗口聚合数据：

from pyspark.sql.functions import window, col, count

# 10-minute tumbling window
windowed_counts = streaming_df \
    .groupBy(
        window(col("timestamp"), "10 minutes"),
        col("word")
    ) \
    .count()

# 10-minute sliding window with 5-minute slide
windowed_counts = streaming_df \
    .groupBy(
        window(col("timestamp"), "10 minutes", "5 minutes"),
        col("word")
    ) \
    .count()

# Write to console for debugging
query = windowed_counts.writeStream \
    .outputMode("complete") \
    .format("console") \
    .option("truncate", "false") \
    .start()

使用水印处理迟到数据

使用水印处理迟到数据：

from pyspark.sql.functions import window

# Define watermark (10 minutes tolerance for late data)
windowed_counts = streaming_df \
    .withWatermark("timestamp", "10 minutes") \
    .groupBy(
        window(col("timestamp"), "10 minutes"),
        col("word")
    ) \
    .count()

# Data arriving more than 10 minutes late will be dropped

水印的优势：

• 丢弃旧的聚合状态，以限制状态大小
• 在容忍时间窗口内处理迟到数据
• 不维护无限状态，从而提升性能

会话窗口

根据不活动间隔将事件分组成会话：

from pyspark.sql.functions import session_window, when

# Dynamic session window based on user
session_window_spec = session_window(
    col("timestamp"),
    when(col("userId") == "user1", "5 seconds")
    .when(col("userId") == "user2", "20 seconds")
    .otherwise("5 minutes")
)

sessionized_counts = streaming_df \
    .withWatermark("timestamp", "10 minutes") \
    .groupBy(session_window_spec, col("userId")) \
    .count()

有状态流处理

在微批次之间维护状态：

from pyspark.sql.functions import expr

# Deduplication using state
deduplicated = streaming_df \
    .withWatermark("timestamp", "1 hour") \
    .dropDuplicates(["user_id", "event_id"])

# Stream-stream joins (stateful)
stream1 = spark.readStream.format("kafka").option("subscribe", "topic1").load()
stream2 = spark.readStream.format("kafka").option("subscribe", "topic2").load()

joined = stream1 \
    .withWatermark("timestamp", "10 minutes") \
    .join(
        stream2.withWatermark("timestamp", "20 minutes"),
        expr("stream1.user_id = stream2.user_id AND stream1.timestamp >= stream2.timestamp AND stream1.timestamp <= stream2.timestamp + interval 15 minutes"),
        "inner"
    )

检查点

使用检查点确保容错性：

# Checkpoint location stores:
# - Stream metadata (offsets, configuration)
# - State information (for stateful operations)
# - Write-ahead logs

query = streaming_df.writeStream \
    .format("parquet") \
    .option("path", "output") \
    .option("checkpointLocation", "checkpoint/dir")  # REQUIRED for production \
    .start()

# Recovery: Restart query with same checkpoint location
# Spark will resume from last committed offset

检查点最佳实践：

• 生产流始终设置 checkpointLocation
• 使用可靠的分布式存储（HDFS、S3）保存检查点
• 流运行时不要删除检查点目录
• 备份检查点以便灾难恢复

使用 MLlib 进行机器学习

Spark 的可扩展机器学习库。

核心组件

MLlib 功能：

• ML Pipelines：串联转换和模型
• 特征化：向量组装器、缩放器、编码器
• 分类与回归：线性模型、基于树的模型、神经网络
• 聚类：K-means、Gaussian Mixture、LDA
• 协同过滤：ALS（交替最小二乘）
• 降维：PCA、SVD
• 模型选择：交叉验证、训练集与测试集拆分、参数调优

ML Pipelines

串联转换器和估算器：

from pyspark.ml import Pipeline
from pyspark.ml.feature import VectorAssembler, StandardScaler
from pyspark.ml.classification import LogisticRegression

# Load data
df = spark.read.format("libsvm").load("data/sample_libsvm_data.txt")

# Define pipeline stages
assembler = VectorAssembler(
    inputCols=["feature1", "feature2", "feature3"],
    outputCol="features"
)

scaler = StandardScaler(
    inputCol="features",
    outputCol="scaled_features",
    withStd=True,
    withMean=True
)

lr = LogisticRegression(
    featuresCol="scaled_features",
    labelCol="label",
    maxIter=10,
    regParam=0.01
)

# Create pipeline
pipeline = Pipeline(stages=[assembler, scaler, lr])

# Split data
train_df, test_df = df.randomSplit([0.8, 0.2], seed=42)

# Train model
model = pipeline.fit(train_df)

# Make predictions
predictions = model.transform(test_df)
predictions.select("label", "prediction", "probability").show()

特征工程

将原始数据转换为特征：

from pyspark.ml.feature import StringIndexer, OneHotEncoder, VectorAssembler, MinMaxScaler

# Categorical encoding
indexer = StringIndexer(inputCol="category", outputCol="category_index")
encoder = OneHotEncoder(inputCol="category_index", outputCol="category_vec")

# Numerical scaling
scaler = MinMaxScaler(inputCol="features", outputCol="scaled_features")

# Assemble features
assembler = VectorAssembler(
    inputCols=["category_vec", "numeric_feature1", "numeric_feature2"],
    outputCol="features"
)

# Text processing
from pyspark.ml.feature import Tokenizer, HashingTF, IDF

tokenizer = Tokenizer(inputCol="text", outputCol="words")
hashing_tf = HashingTF(inputCol="words", outputCol="raw_features", numFeatures=10000)
idf = IDF(inputCol="raw_features", outputCol="features")

流式线性回归

使用流式数据训练模型：

from pyspark.mllib.regression import LabeledPoint
from pyspark.streaming import StreamingContext
from pyspark.streaming.ml import StreamingLinearRegressionWithSGD

# Create StreamingContext
ssc = StreamingContext(sc, batchDuration=1)

# Define data streams
training_stream = ssc.textFileStream("training/data/path")
testing_stream = ssc.textFileStream("testing/data/path")

# Parse streams into LabeledPoint objects
def parse_point(line):
    values = [float(x) for x in line.strip().split(',')]
    return LabeledPoint(values[0], values[1:])

parsed_training = training_stream.map(parse_point)
parsed_testing = testing_stream.map(parse_point)

# Initialize model
num_features = 3
model = StreamingLinearRegressionWithSGD(initialWeights=[0.0] * num_features)

# Train and predict
model.trainOn(parsed_training)
predictions = model.predictOnValues(parsed_testing.map(lambda lp: (lp.label, lp.features)))

# Print predictions
predictions.pprint()

# Start streaming
ssc.start()
ssc.awaitTermination()

模型评估

评估模型性能：

from pyspark.ml.evaluation import BinaryClassificationEvaluator, MulticlassClassificationEvaluator, RegressionEvaluator

# Binary classification
binary_evaluator = BinaryClassificationEvaluator(
    labelCol="label",
    rawPredictionCol="rawPrediction",
    metricName="areaUnderROC"
)
auc = binary_evaluator.evaluate(predictions)
print(f"AUC: {auc}")

# Multiclass classification
multi_evaluator = MulticlassClassificationEvaluator(
    labelCol="label",
    predictionCol="prediction",
    metricName="accuracy"
)
accuracy = multi_evaluator.evaluate(predictions)
print(f"Accuracy: {accuracy}")

# Regression
regression_evaluator = RegressionEvaluator(
    labelCol="label",
    predictionCol="prediction",
    metricName="rmse"
)
rmse = regression_evaluator.evaluate(predictions)
print(f"RMSE: {rmse}")

超参数调优

使用交叉验证优化模型参数：

from pyspark.ml.tuning import ParamGridBuilder, CrossValidator
from pyspark.ml.classification import RandomForestClassifier
from pyspark.ml.evaluation import MulticlassClassificationEvaluator

# Define model
rf = RandomForestClassifier(labelCol="label", featuresCol="features")

# Build parameter grid
param_grid = ParamGridBuilder() \
    .addGrid(rf.numTrees, [10, 20, 50]) \
    .addGrid(rf.maxDepth, [5, 10, 15]) \
    .addGrid(rf.minInstancesPerNode, [1, 5, 10]) \
    .build()

# Define evaluator
evaluator = MulticlassClassificationEvaluator(metricName="accuracy")

# Cross-validation
cv = CrossValidator(
    estimator=rf,
    estimatorParamMaps=param_grid,
    evaluator=evaluator,
    numFolds=5,
    parallelism=4
)

# Train
cv_model = cv.fit(train_df)

# Best model
best_model = cv_model.bestModel
print(f"Best numTrees: {best_model.getNumTrees}")
print(f"Best maxDepth: {best_model.getMaxDepth()}")

# Evaluate on test set
predictions = cv_model.transform(test_df)
accuracy = evaluator.evaluate(predictions)
print(f"Test Accuracy: {accuracy}")

分布式矩阵运算

MLlib 提供分布式矩阵表示：

from pyspark.mllib.linalg.distributed import RowMatrix, IndexedRowMatrix, CoordinateMatrix
from pyspark.mllib.linalg import Vectors

# RowMatrix: Distributed matrix without row indices
rows = sc.parallelize([
    Vectors.dense([1.0, 2.0, 3.0]),
    Vectors.dense([4.0, 5.0, 6.0]),
    Vectors.dense([7.0, 8.0, 9.0])
])
row_matrix = RowMatrix(rows)

# Compute statistics
print(f"Rows: {row_matrix.numRows()}")
print(f"Cols: {row_matrix.numCols()}")
print(f"Column means: {row_matrix.computeColumnSummaryStatistics().mean()}")

# IndexedRowMatrix: Matrix with row indices
from pyspark.mllib.linalg.distributed import IndexedRow
indexed_rows = sc.parallelize([
    IndexedRow(0, Vectors.dense([1.0, 2.0, 3.0])),
    IndexedRow(1, Vectors.dense([4.0, 5.0, 6.0]))
])
indexed_matrix = IndexedRowMatrix(indexed_rows)

# CoordinateMatrix: Sparse matrix using (row, col, value) entries
from pyspark.mllib.linalg.distributed import MatrixEntry
entries = sc.parallelize([
    MatrixEntry(0, 0, 1.0),
    MatrixEntry(0, 2, 3.0),
    MatrixEntry(1, 1, 5.0)
])
coord_matrix = CoordinateMatrix(entries)

分层抽样

在保留类别分布的同时对数据进行抽样：

# Scala/Java approach
data = [("a", 1), ("b", 2), ("a", 3), ("b", 4), ("a", 5), ("c", 6)]
rdd = sc.parallelize(data)

# Define sampling fractions per key
fractions = {"a": 0.5, "b": 0.5, "c": 0.5}

# Approximate sample (faster, one pass)
sampled_rdd = rdd.sampleByKey(withReplacement=False, fractions=fractions)

# Exact sample (slower, guaranteed exact counts)
exact_sampled = rdd.sampleByKeyExact(withReplacement=False, fractions=fractions)

print(sampled_rdd.collect())

性能调优

内存管理

内存划分：

• 执行内存：用于 shuffle、join、排序和聚合
• 存储内存：用于缓存和广播变量
• 用户内存：用于用户数据结构和 UDF
• 预留内存：预留给 Spark 内部操作

配置：

spark = SparkSession.builder \
    .appName("MemoryTuning") \
    .config("spark.executor.memory", "4g") \
    .config("spark.driver.memory", "2g") \
    .config("spark.memory.fraction", "0.6")  # Fraction for execution + storage \
    .config("spark.memory.storageFraction", "0.5")  # Fraction of above for storage \
    .getOrCreate()

内存最佳实践：

• 通过 Spark UI 监控内存使用情况
• 为缓存使用合适的存储级别
• 避免将大型数据集收集到 driver
• 对于内存密集型操作，增加 executor 内存
• 使用 kryo 序列化以提高内存效率

Shuffle 优化

Shuffle 是开销很大的操作——应尽量减少：

Shuffle 的原因：

• groupByKey、reduceByKey、aggregateByKey
• join、cogroup
• repartition、coalesce（增加分区数时）
• distinct、intersection
• sortByKey

优化策略：

# 1. Use reduceByKey instead of groupByKey
# Bad: groupByKey shuffles all data
word_pairs.groupByKey().mapValues(sum)

# Good: reduceByKey combines locally before shuffle
word_pairs.reduceByKey(lambda a, b: a + b)

# 2. Broadcast small tables in joins
from pyspark.sql.functions import broadcast
large_df.join(broadcast(small_df), "key")

# 3. Partition data appropriately
df.repartition(200, "user_id")  # Partition by key for subsequent aggregations

# 4. Coalesce instead of repartition when reducing partitions
df.coalesce(10)  # No shuffle, just merge partitions

# 5. Tune shuffle partitions
spark.conf.set("spark.sql.shuffle.partitions", 200)  # Default is 200

Shuffle 配置：

spark = SparkSession.builder \
    .config("spark.sql.shuffle.partitions", 200) \
    .config("spark.default.parallelism", 200) \
    .config("spark.sql.adaptive.enabled", "true")  # Enable AQE \
    .config("spark.sql.adaptive.coalescePartitions.enabled", "true") \
    .getOrCreate()

分区策略

分区数指南：

• 过少：集群利用率不足，出现 OOM 错误
• 过多：任务调度开销增加
• 合适范围：CPU 核数的 2-4 倍
• 对于大型 shuffle：100-200 个以上的分区

按列分区：

# Partition writes by date for easy filtering
df.write.partitionBy("date", "country").parquet("output")

# Read with partition pruning (only reads relevant partitions)
spark.read.parquet("output").filter(col("date") == "2025-01-15").show()

自定义分区：

from pyspark.rdd import portable_hash

# Custom partitioner for RDD
def custom_partitioner(key):
    return portable_hash(key) % 100

rdd.partitionBy(100, custom_partitioner)

缓存策略

何时缓存：

# Iterative algorithms (ML)
training_data.cache()
for i in range(num_iterations):
    model = train_model(training_data)

# Multiple aggregations on same data
base_df.cache()
result1 = base_df.groupBy("country").count()
result2 = base_df.groupBy("city").avg("sales")

# Interactive analysis
df.cache()
df.filter(condition1).show()
df.filter(condition2).show()
df.groupBy("category").count().show()

存储级别：

from pyspark import StorageLevel

# Memory only (fastest, but may lose data)
df.persist(StorageLevel.MEMORY_ONLY)

# Memory and disk (spill to disk if needed)
df.persist(StorageLevel.MEMORY_AND_DISK)

# Serialized in memory (more compact, slower access)
df.persist(StorageLevel.MEMORY_ONLY_SER)

# Disk only (slowest, but always available)
df.persist(StorageLevel.DISK_ONLY)

# Replicated (fault tolerance)
df.persist(StorageLevel.MEMORY_AND_DISK_2)  # 2 replicas

广播 Join

使用小表优化 Join：

from pyspark.sql.functions import broadcast

# Automatic broadcast (tables < spark.sql.autoBroadcastJoinThreshold)
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", 10 * 1024 * 1024)  # 10 MB

# Explicit broadcast hint
large_df.join(broadcast(small_df), "key")

# Benefits:
# - No shuffle of large table
# - Small table sent to all executors once
# - Much faster for small dimension tables

自适应查询执行（AQE）

启用运行时查询优化：

spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")

# AQE Benefits:
# - Dynamically coalesce partitions after shuffle
# - Handle skewed joins by splitting large partitions
# - Optimize join strategy at runtime

数据格式选择

性能对比：

1. Parquet（最适合分析）：列式存储、压缩，查询速度快
2. ORC（最适合 Hive）：与 Parquet 类似，压缩率略高
3. Avro（最适合行式存储）：适用于写入密集型工作负载
4. JSON（速度最慢）：可读性好，但效率低
5. CSV（旧格式）：兼容性好，但速度慢且没有 schema

建议：

• 大多数分析型工作负载使用 Parquet
• 启用压缩（snappy、gzip、lzo）
• 按常用筛选列分区
• 对于读取密集型工作负载，使用列式格式

Catalyst 优化器

了解查询优化：

# View physical plan
df.explain(mode="extended")

# Optimizations include:
# - Predicate pushdown: Push filters to data source
# - Column pruning: Read only required columns
# - Constant folding: Evaluate constants at compile time
# - Join reordering: Optimize join order
# - Partition pruning: Skip irrelevant partitions

生产环境部署

集群管理器

Standalone：

• 简单、内置的集群管理器
• 易于为开发和小型集群设置
• 不与其他框架共享资源

# Start master
$SPARK_HOME/sbin/start-master.sh

# Start workers
$SPARK_HOME/sbin/start-worker.sh spark://master:7077

# Submit application
spark-submit --master spark://master:7077 app.py

YARN：

• Hadoop 的资源管理器
• 与 MapReduce、Hive 等共享集群资源
• 两种模式：cluster（driver 在 YARN 上）和 client（driver 在本地机器上）

# Cluster mode (driver runs on YARN)
spark-submit --master yarn --deploy-mode cluster app.py

# Client mode (driver runs locally)
spark-submit --master yarn --deploy-mode client app.py

Kubernetes：

• 现代容器编排
• 动态资源分配
• 云原生部署

spark-submit \
  --master k8s://https://k8s-master:443 \
  --deploy-mode cluster \
  --name spark-app \
  --conf spark.executor.instances=5 \
  --conf spark.kubernetes.container.image=spark:latest \
  app.py

Mesos：

• 通用集群管理器
• 支持细粒度或粗粒度资源共享

应用程序提交

基本的 spark-submit：

spark-submit \
  --master yarn \
  --deploy-mode cluster \
  --driver-memory 4g \
  --executor-memory 8g \
  --executor-cores 4 \
  --num-executors 10 \
  --conf spark.sql.shuffle.partitions=200 \
  --py-files dependencies.zip \
  --files config.json \
  application.py

配置选项：

• --master：集群管理器 URL
• --deploy-mode：driver 的运行位置（client 或 cluster）
• --driver-memory：driver 进程的内存
• --executor-memory：每个 executor 的内存
• --executor-cores：每个 executor 的核心数
• --num-executors：executor 数量
• --conf：Spark 配置属性
• --py-files：Python 依赖项
• --files：要分发的其他文件

资源分配

一般指南：

• Driver 内存：1-4 GB（除非要收集大量结果）
• Executor 内存：每个 executor 4-16 GB
• Executor 核心数：每个 executor 4-5 个核心（超过 5 个后收益递减）
• Executor 数量：充分利用集群容量，同时为操作系统和其他服务预留资源
• 并行度：总核心数的 2-4 倍

计算示例：

Cluster: 10 nodes, 32 cores each, 128 GB RAM each

Option 1: Many small executors
  - 30 executors (3 per node)
  - 10 cores per executor
  - 40 GB memory per executor
  - Total: 300 cores

Option 2: Fewer large executors (RECOMMENDED)
  - 50 executors (5 per node)
  - 5 cores per executor
  - 24 GB memory per executor
  - Total: 250 cores

动态分配

根据工作负载自动扩缩 executor：

spark = SparkSession.builder \
    .appName("DynamicAllocation") \
    .config("spark.dynamicAllocation.enabled", "true") \
    .config("spark.dynamicAllocation.minExecutors", 2) \
    .config("spark.dynamicAllocation.maxExecutors", 100) \
    .config("spark.dynamicAllocation.initialExecutors", 10) \
    .config("spark.dynamicAllocation.executorIdleTimeout", "60s") \
    .getOrCreate()

优势：

• 更高的资源利用率
• 根据工作负载变化自动扩缩
• 降低云环境中的成本

监控和日志记录

Spark UI：

• Web 界面：http://driver:4040
• 阶段、任务、存储、环境、执行器
• SQL 查询计划和执行详情
• 识别瓶颈和性能问题

历史服务器：

# Start history server
$SPARK_HOME/sbin/start-history-server.sh

# Configure event logging
spark.conf.set("spark.eventLog.enabled", "true")
spark.conf.set("spark.eventLog.dir", "hdfs://namenode/spark-logs")

指标：

# Enable metrics collection
spark.conf.set("spark.metrics.conf.*.sink.console.class", "org.apache.spark.metrics.sink.ConsoleSink")
spark.conf.set("spark.metrics.conf.*.sink.console.period", 10)

日志记录：

# Configure log level
spark.sparkContext.setLogLevel("WARN")  # ERROR, WARN, INFO, DEBUG

# Custom logging
import logging
logger = logging.getLogger(__name__)
logger.info("Custom log message")

容错

自动恢复：

• 任务失败：自动重试失败的任务
• 执行器失败：在其他执行器上重新调度任务
• Driver 失败：从检查点恢复（流处理）
• 节点失败：根据血缘关系重新计算丢失的分区

检查点：

# Set checkpoint directory
spark.sparkContext.setCheckpointDir("hdfs://namenode/checkpoints")

# Checkpoint RDD (breaks lineage for very long chains)
rdd.checkpoint()

# Streaming checkpoint (required for production)
query = streaming_df.writeStream \
    .option("checkpointLocation", "hdfs://namenode/streaming-checkpoint") \
    .start()

推测执行：

# Enable speculative execution for slow tasks
spark.conf.set("spark.speculation", "true")
spark.conf.set("spark.speculation.multiplier", 1.5)
spark.conf.set("spark.speculation.quantile", 0.75)

数据本地性

优化数据放置以提升性能：

本地性级别：

1. PROCESS_LOCAL：数据与任务位于同一 JVM 中（最快）
2. NODE_LOCAL：数据位于同一节点，但属于不同进程
3. RACK_LOCAL：数据位于同一机架
4. ANY：数据位于不同机架（最慢）

提升本地性：

# Increase locality wait time
spark.conf.set("spark.locality.wait", "10s")
spark.conf.set("spark.locality.wait.node", "5s")
spark.conf.set("spark.locality.wait.rack", "3s")

# Partition data to match cluster topology
df.repartition(num_nodes * cores_per_node)

最佳实践

代码组织

1. 模块化设计：分离数据加载、转换和输出逻辑
2. 配置管理：将配置外置（使用配置文件）
3. 错误处理：实现健壮的错误处理和日志记录
4. 测试：对转换进行单元测试，对流水线进行集成测试
5. 文档：记录复杂的转换和业务逻辑

性能

1. 避免 Shuffle：使用 reduceByKey，而不是 groupByKey
2. 合理缓存：仅缓存会多次重用的数据
3. 广播小表：对小型参考数据使用广播联接
4. 合理分区：分区数设为 CPU 核心数的 2-4 倍，并按经常用于筛选的列分区
5. 使用 Parquet：分析型工作负载采用列式格式
6. 启用 AQE：利用自适应查询执行进行运行时优化
7. 调整内存：平衡执行器内存和核心数
8. 监控：使用 Spark UI 识别瓶颈

开发工作流

1. 从小处着手：先使用示例数据在本地开发
2. 尽早分析：从一开始就监控性能
3. 迭代：根据指标逐步优化
4. 规模化测试：部署前使用生产规模的数据进行验证
5. 版本控制：跟踪代码、配置和架构

数据质量

1. 架构验证：读写时强制执行架构
2. 空值处理：明确处理空值
3. 数据验证：检查是否符合预期范围、格式和约束
4. 去重：根据业务逻辑移除重复项
5. 审计日志：跟踪数据血缘和转换

安全

1. 身份验证：为 YARN/HDFS 启用 Kerberos
2. 授权：使用 ACL 控制数据访问
3. 加密：加密静态和传输中的数据
4. 密钥管理：使用安全的凭据提供程序
5. 审计跟踪：记录数据访问和修改

成本优化

1. 合理配置资源：不要过度配置执行器
2. 动态分配：根据工作负载扩缩执行器
3. 竞价实例：在云环境中使用竞价/可抢占实例
4. 数据压缩：使用高效格式（Parquet、ORC）
5. 分区：跳过不必要的数据读取
6. 自动关闭：终止空闲集群

常见模式

ETL 流水线模式

def etl_pipeline(spark, input_path, output_path):
    # Extract
    raw_df = spark.read.parquet(input_path)

    # Transform
    cleaned_df = raw_df \
        .dropDuplicates(["id"]) \
        .filter(col("value").isNotNull()) \
        .withColumn("processed_date", current_date())

    # Enrich
    enriched_df = cleaned_df.join(broadcast(reference_df), "key")

    # Aggregate
    aggregated_df = enriched_df \
        .groupBy("category", "date") \
        .agg(
            count("*").alias("count"),
            sum("amount").alias("total_amount"),
            avg("value").alias("avg_value")
        )

    # Load
    aggregated_df.write \
        .partitionBy("date") \
        .mode("overwrite") \
        .parquet(output_path)

增量处理模式

def incremental_process(spark, input_path, output_path, checkpoint_path):
    # Read last processed timestamp
    last_timestamp = read_checkpoint(checkpoint_path)

    # Read new data
    new_data = spark.read.parquet(input_path) \
        .filter(col("timestamp") > last_timestamp)

    # Process
    processed = transform(new_data)

    # Write
    processed.write.mode("append").parquet(output_path)

    # Update checkpoint
    max_timestamp = new_data.agg(max("timestamp")).collect()[0][0]
    write_checkpoint(checkpoint_path, max_timestamp)

缓慢变化维度（SCD）模式

def scd_type2_upsert(spark, dimension_df, updates_df):
    # Mark existing records as inactive if updated
    inactive_records = dimension_df \
        .join(updates_df, "business_key") \
        .select(
            dimension_df["*"],
            lit(False).alias("is_active"),
            current_date().alias("end_date")
        )

    # Add new records
    new_records = updates_df \
        .withColumn("is_active", lit(True)) \
        .withColumn("start_date", current_date()) \
        .withColumn("end_date", lit(None))

    # Union unchanged, inactive, and new records
    result = dimension_df \
        .join(updates_df, "business_key", "left_anti") \
        .union(inactive_records) \
        .union(new_records)

    return result

窗口分析模式

def calculate_running_metrics(df):
    from pyspark.sql.window import Window
    from pyspark.sql.functions import row_number, lag, sum, avg

    # Define window
    window_spec = Window.partitionBy("user_id").orderBy("timestamp")

    # Calculate metrics
    result = df \
        .withColumn("row_num", row_number().over(window_spec)) \
        .withColumn("prev_value", lag("value", 1).over(window_spec)) \
        .withColumn("running_total", sum("value").over(window_spec.rowsBetween(Window.unboundedPreceding, Window.currentRow))) \
        .withColumn("moving_avg", avg("value").over(window_spec.rowsBetween(-2, 0)))

    return result

故障排除

内存不足错误

症状：

• java.lang.OutOfMemoryError
• 执行器失败
• 垃圾回收缓慢

解决方案：

# Increase executor memory
spark.conf.set("spark.executor.memory", "8g")

# Increase driver memory (if collecting data)
spark.conf.set("spark.driver.memory", "4g")

# Reduce memory pressure
df.persist(StorageLevel.MEMORY_AND_DISK)  # Spill to disk
df.coalesce(100)  # Reduce partition count
spark.conf.set("spark.sql.shuffle.partitions", 400)  # Increase shuffle partitions

# Avoid collect() on large datasets
# Use take() or limit() instead
df.take(100)

Shuffle 性能问题

症状：

• Shuffle 读写耗时较长
• 分区大小不均
• 任务执行落后

解决方案：

# Increase shuffle partitions
spark.conf.set("spark.sql.shuffle.partitions", 400)

# Handle skew with salting
df_salted = df.withColumn("salt", (rand() * 10).cast("int"))
result = df_salted.groupBy("key", "salt").agg(...)

# Use broadcast for small tables
large_df.join(broadcast(small_df), "key")

# Enable AQE for automatic optimization
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")

流处理作业失败

症状：

• 流式查询已停止
• 检查点损坏
• 处理延迟不断增加

解决方案：

# Increase executor memory for stateful operations
spark.conf.set("spark.executor.memory", "8g")

# Tune watermark for late data
.withWatermark("timestamp", "15 minutes")

# Increase trigger interval to reduce micro-batch overhead
.trigger(processingTime="30 seconds")

# Monitor lag and adjust parallelism
spark.conf.set("spark.sql.shuffle.partitions", 200)

# Recover from checkpoint corruption
# Delete checkpoint directory and restart (data loss possible)
# Or implement custom state recovery logic

数据倾斜

症状：

• 少数任务的耗时远超其他任务
• 分区大小不均衡
• 执行器出现 OOM 错误

解决方案：

# 1. Salting technique (add random prefix to keys)
from pyspark.sql.functions import concat, lit, rand

df_salted = df.withColumn("salted_key", concat(col("key"), lit("_"), (rand() * 10).cast("int")))
result = df_salted.groupBy("salted_key").agg(...)

# 2. Repartition by skewed column
df.repartition(200, "skewed_column")

# 3. Isolate skewed keys
skewed_keys = df.groupBy("key").count().filter(col("count") > threshold).select("key")
skewed_df = df.join(broadcast(skewed_keys), "key")
normal_df = df.join(broadcast(skewed_keys), "key", "left_anti")

# Process separately
skewed_result = process_with_salting(skewed_df)
normal_result = process_normally(normal_df)
final = skewed_result.union(normal_result)

Context7 代码集成

此技能集成了 Apache Spark 官方代码库中的真实代码示例。EXAMPLES.md 文件中的所有代码片段均来自 Context7 的 Apache Spark 库文档，确保采用可用于生产环境的模式和最佳实践。

版本与兼容性

• Apache Spark 版本：3.x（兼容 2.4+）
• Python：3.7+
• Scala：2.12+
• Java：8+
• R：3.5+

参考资料

• 官方文档：https://spark.apache.org/docs/latest/
• API 参考：https://spark.apache.org/docs/latest/api.html
• GitHub 代码库：https://github.com/apache/spark
• Databricks 博客：https://databricks.com/blog
• Context7 库：/apache/spark

---

技能版本：1.0.0 最后更新：2025 年 10 月 技能类别：大数据、分布式计算、数据工程、机器学习 Context7 集成：/apache/spark，包含 8000 个 token 的文档