环境声明

• Python版本：Python 3.12+（建议使用 3.10 以上版本）
• 开发工具：PyCharm 或 VS Code
• 操作系统：Windows / macOS / Linux（通用）
• PySpark版本：3.5+

---

学习目标

本章学习完成后，你将能够：

1. 理解大数据和分布式计算的基本概念，掌握Spark的核心架构和工作原理
2. 搭建PySpark开发环境，配置本地和集群运行模式
3. 熟练使用RDD（弹性分布式数据集）进行数据转换和行动操作
4. 掌握DataFrame API，使用Spark SQL进行结构化数据处理
5. 了解Spark MLlib机器学习库，实现分布式机器学习任务

---

章节内容

1. 大数据与Spark概述

1.1 什么是大数据

大数据是指规模巨大、类型多样、处理速度要求高的数据集合。传统数据处理工具难以应对，因此需要分布式计算框架。

1.2 Spark的诞生与优势

Spark是由Apache软件基金会开发的快速通用集群计算系统，相比Hadoop MapReduce具有以下优势：

特性	Spark	Hadoop MapReduce
计算速度	内存计算，快10-100倍	磁盘IO频繁，速度较慢
易用性	支持Python、Scala、Java、R	主要使用Java
通用性	批处理、流处理、SQL、机器学习统一	主要面向批处理
容错性	基于RDD的血缘机制	基于任务重试

补充：Spark的创始人Matei Zaharia在UC Berkeley的AMPLab读博期间开发了Spark，2014年成为Apache顶级项目。

1.3 Spark核心架构

+---------------------+
|    Driver Program   |  (驱动程序，运行main函数)
|  (包含SparkContext) |
+----------+----------+
           |
+----------v----------+
|   Cluster Manager   |  (集群管理器：Standalone/YARN/Mesos)
+----------+----------+
           |
    +------+------+
    |             |
+---v---+     +---v---+
|Worker1|     |Worker2|  (工作节点，执行实际任务)
|Executor|    |Executor|
+--------+    +--------+

---

2. PySpark环境搭建

2.1 安装PySpark

# 使用pip安装
pip install pyspark

# 或者使用conda安装
conda install -c conda-forge pyspark

2.2 验证安装

from pyspark.sql import SparkSession

# 创建SparkSession
spark = SparkSession.builder \
    .appName("PySpark入门") \
    .getOrCreate()

# 查看Spark版本
print(f"Spark版本: {spark.version}")

# 停止SparkSession
spark.stop()

---

3. RDD编程基础

3.1 什么是RDD

RDD（Resilient Distributed Dataset，弹性分布式数据集）是Spark最基本的抽象，代表一个不可变、可分区的数据集合。

一句话总结：RDD就像把一个大文件切成很多小块，分散存储在多台机器上，可以并行处理。

RDD的五大核心特性

特性	说明	作用
分区列表	数据被划分为多个分区	实现并行计算的基础
计算函数	每个分区的计算逻辑	定义如何从父RDD计算当前RDD
依赖关系	记录RDD之间的血缘关系	实现容错和故障恢复
分区器	决定数据如何分布到分区	影响Shuffle性能
优先位置	数据本地性信息	优化任务调度，减少网络传输

RDD的不可变性

RDD一旦创建就不能修改，所有转换操作都会生成新的RDD。这种设计带来三个好处：

1. 线程安全：多线程并发访问无需加锁
2. 容错简单：通过重新计算即可恢复丢失数据
3. 优化机会：Spark可以在执行前优化整个计算流程

from pyspark import SparkContext

sc = SparkContext.getOrCreate()

# RDD不可变性示例
rdd1 = sc.parallelize([1, 2, 3, 4, 5])
rdd2 = rdd1.map(lambda x: x * 2)  # rdd1保持不变，返回新RDD
rdd3 = rdd1.filter(lambda x: x > 2)  # rdd1仍然保持不变

print(f"rdd1: {rdd1.collect()}")  # [1, 2, 3, 4, 5]
print(f"rdd2: {rdd2.collect()}")  # [2, 4, 6, 8, 10]
print(f"rdd3: {rdd3.collect()}")  # [3, 4, 5]

3.2 创建RDD

from pyspark import SparkContext

sc = SparkContext.getOrCreate()

# 从列表创建RDD
data = [1, 2, 3, 4, 5]
rdd = sc.parallelize(data)

# 指定分区数创建RDD（推荐大数据场景）
rdd_partitioned = sc.parallelize(data, numSlices=3)
print(f"分区数: {rdd_partitioned.getNumPartitions()}")

# 从文件创建RDD
rdd_from_file = sc.textFile("data.txt")

# 从文件创建RDD并指定最小分区数
rdd_from_file_partitioned = sc.textFile("data.txt", minPartitions=4)

# 从HDFS创建RDD
rdd_hdfs = sc.textFile("hdfs://namenode:9000/path/to/file.txt")

# 从多个文件创建RDD（通配符）
rdd_wildcard = sc.textFile("data/*.csv")

# 从数据库创建RDD（需要JDBC）
# rdd_jdbc = sc.jdbc("jdbc:mysql://localhost/db", "table", properties={"user": "root"})

# 从现有RDD创建新RDD
rdd_from_rdd = rdd.map(lambda x: x * 2)

3.3 RDD依赖关系与Shuffle机制

窄依赖 vs 宽依赖

RDD之间的依赖关系分为两类，这是理解Spark性能的关键：

窄依赖（Narrow Dependency）
+----------------+     +----------------+
|   Parent RDD   |     |   Parent RDD   |
|   Partition 1  |     |   Partition 2  |
+--------+-------+     +--------+-------+
         |                      |
         v                      v
+--------+-------+     +--------+-------+
|   Child RDD    |     |   Child RDD    |
|   Partition 1  |     |   Partition 2  |
+----------------+     +----------------+
特点：父RDD的一个分区对应子RDD的一个分区
操作：map, filter, flatMap, union

宽依赖（Wide Dependency / Shuffle Dependency）
+----------------+     +----------------+
|   Parent RDD   |     |   Parent RDD   |
|   Partition 1  |     |   Partition 2  |
+--------+-------+     +--------+-------+
         |                      |
         +----------+-----------+
                    |
         +----------+-----------+
         |                      |
         v                      v
+--------+-------+     +--------+-------+
|   Child RDD    |     |   Child RDD    |
|   Partition 1  |     |   Partition 2  |
+----------------+     +----------------+
特点：父RDD的一个分区对应子RDD的多个分区
操作：reduceByKey, groupByKey, sortByKey, join

# 窄依赖示例 - 无需Shuffle
rdd = sc.parallelize([1, 2, 3, 4, 5, 6], 3)
# map是窄依赖，每个分区独立计算
mapped = rdd.map(lambda x: x * 2)
# filter也是窄依赖
filtered = mapped.filter(lambda x: x > 4)

print(f"窄依赖 lineage: {filtered.toDebugString().decode('utf-8')}")

# 宽依赖示例 - 需要Shuffle
pairs = sc.parallelize([
    ("a", 1), ("b", 2), ("a", 3), ("b", 4), ("c", 5)
], 2)
# reduceByKey是宽依赖，需要按Key重新分区
reduced = pairs.reduceByKey(lambda x, y: x + y)

print(f"宽依赖 lineage: {reduced.toDebugString().decode('utf-8')}")

Shuffle机制详解

Shuffle是Spark中最耗时的操作，理解其原理对性能优化至关重要：

Shuffle过程示意图

Map阶段（写入）                    Reduce阶段（读取）
+--------------+                   +--------------+
| Partition 1  |---> [Sort] --->   |              |
| Partition 2  |---> [Sort] --->   |  Shuffle     |---> [Merge] ---> Result
| Partition 3  |---> [Sort] --->   |  Read        |
+--------------+                   +--------------+
       |                                  ^
       |         磁盘文件 / 网络传输       |
       +----------------------------------+

# Shuffle操作对比实验
import time

# 创建测试数据
data = [(i % 100, i) for i in range(100000)]
rdd = sc.parallelize(data, 4)

# 实验1: groupByKey（产生大量Shuffle数据）
start = time.time()
grouped = rdd.groupByKey().mapValues(list)
grouped.count()  # 触发计算
groupby_time = time.time() - start

# 实验2: reduceByKey（在Map端预聚合，减少Shuffle）
start = time.time()
reduced = rdd.reduceByKey(lambda x, y: x + y)
reduced.count()  # 触发计算
reduceby_time = time.time() - start

print(f"groupByKey耗时: {groupby_time:.3f}秒")
print(f"reduceByKey耗时: {reduceby_time:.3f}秒")
print(f"性能提升: {groupby_time / reduceby_time:.1f}倍")

# 实验3: aggregateByKey（更灵活的预聚合）
start = time.time()
# 参数: 初始值, 分区聚合函数, 全局聚合函数
aggregated = rdd.aggregateByKey(
    (0, 0),  # 初始值: (sum, count)
    lambda acc, val: (acc[0] + val, acc[1] + 1),  # 分区聚合
    lambda acc1, acc2: (acc1[0] + acc2[0], acc1[1] + acc2[1])  # 全局聚合
)
avg_result = aggregated.mapValues(lambda x: x[0] / x[1])
avg_result.count()
aggregate_time = time.time() - start

print(f"aggregateByKey耗时: {aggregate_time:.3f}秒")

分区策略详解

# 1. Hash分区（默认）
# 根据Key的hashCode决定分区
hash_partitioned = pairs.partitionBy(numPartitions=4)

# 2. Range分区
# 根据Key的范围划分分区，适合有序数据
from pyspark import RangePartitioner
# pairs.sortByKey() 内部使用Range分区

# 3. 自定义分区器
from pyspark import Partitioner

class CustomPartitioner(Partitioner):
    def __init__(self, num_partitions):
        self.num_partitions = num_partitions
  
    def numPartitions(self):
        return self.num_partitions
  
    def getPartition(self, key):
        # 自定义分区逻辑：按Key的首字母分区
        if isinstance(key, str) and len(key) > 0:
            return ord(key[0].lower()) % self.num_partitions
        return 0

# 使用自定义分区器
# custom_partitioned = pairs.partitionBy(CustomPartitioner(4))

# 4. 查看分区信息
print(f"分区数: {rdd.getNumPartitions()}")
print(f"各分区元素数: {rdd.glom().map(len).collect()}")

# 5. 重新分区
coalesced = rdd.coalesce(2)  # 减少分区（窄依赖，无Shuffle）
repartitioned = rdd.repartition(6)  # 增加分区（宽依赖，有Shuffle）

3.4 RDD转换操作（Transformations）

转换操作是惰性求值的，不会立即执行。

# map: 对每个元素应用函数
rdd = sc.parallelize([1, 2, 3, 4])
mapped_rdd = rdd.map(lambda x: x * 2)  # [2, 4, 6, 8]

# filter: 过滤元素
filtered_rdd = rdd.filter(lambda x: x > 2)  # [3, 4]

# flatMap: 扁平化映射
words_rdd = sc.parallelize(["hello world", "pyspark"])
flat_rdd = words_rdd.flatMap(lambda x: x.split(" "))  # ["hello", "world", "pyspark"]

# reduceByKey: 按键聚合
pairs = sc.parallelize([("a", 1), ("b", 2), ("a", 3)])
reduced = pairs.reduceByKey(lambda x, y: x + y)  # [("a", 4), ("b", 2)]

3.4 RDD行动操作（Actions）

行动操作会触发实际计算。

# collect: 收集所有元素到驱动程序
result = rdd.collect()

# count: 统计元素个数
count = rdd.count()

# take: 取前N个元素
first_n = rdd.take(3)

# reduce: 聚合所有元素
total = rdd.reduce(lambda x, y: x + y)

# saveAsTextFile: 保存到文件
rdd.saveAsTextFile("output_path")

3.5 RDD高级转换操作

# combineByKey: 最灵活的聚合操作
# 参数: createCombiner, mergeValue, mergeCombiners
pairs = sc.parallelize([
    ("apple", 3), ("banana", 2), ("apple", 5),
    ("banana", 4), ("cherry", 1)
])

# 计算每个Key的平均值
avg_by_key = pairs.combineByKey(
    lambda val: (val, 1),  # 初始值转换: value -> (sum, count)
    lambda acc, val: (acc[0] + val, acc[1] + 1),  # 合并新值
    lambda acc1, acc2: (acc1[0] + acc2[0], acc1[1] + acc2[1])  # 合并分区结果
).mapValues(lambda x: x[0] / x[1])

print(avg_by_key.collect())

# foldByKey: 带初始值的reduceByKey
folded = pairs.foldByKey(0, lambda x, y: x + y)

# sortByKey: 按键排序
sorted_pairs = pairs.sortByKey(ascending=True, numPartitions=2)

# cogroup: 多RDD分组（类似SQL的FULL OUTER JOIN）
rdd1 = sc.parallelize([("a", 1), ("b", 2)])
rdd2 = sc.parallelize([("a", "x"), ("a", "y"), ("b", "z")])
cogrouped = rdd1.cogroup(rdd2)
print(cogrouped.mapValues(lambda x: (list(x[0]), list(x[1]))).collect())

# join系列操作
rdd3 = sc.parallelize([("a", 10), ("c", 30)])
inner_joined = rdd1.join(rdd3)  # 内连接
left_joined = rdd1.leftOuterJoin(rdd3)  # 左外连接
full_joined = rdd1.fullOuterJoin(rdd3)  # 全外连接

# subtract: 集合差集
rdd_a = sc.parallelize([1, 2, 3, 4])
rdd_b = sc.parallelize([3, 4, 5, 6])
diff = rdd_a.subtract(rdd_b)  # [1, 2]

# intersection: 集合交集
intersect = rdd_a.intersection(rdd_b)  # [3, 4]

# distinct: 去重
unique = sc.parallelize([1, 2, 2, 3, 3, 3]).distinct()

# cartesian: 笛卡尔积（谨慎使用，数据量爆炸）
cartesian = sc.parallelize([1, 2]).cartesian(sc.parallelize(["a", "b"]))
# [(1, "a"), (1, "b"), (2, "a"), (2, "b")]

# zip: 拉链操作（两个RDD分区数和元素数必须相同）
zipped = sc.parallelize([1, 2, 3]).zip(sc.parallelize(["a", "b", "c"]))

# zipWithIndex: 添加索引
with_index = sc.parallelize(["a", "b", "c"]).zipWithIndex()
# [("a", 0), ("b", 1), ("c", 2)]

3.6 RDD缓存与持久化

# 缓存策略对比
from pyspark import StorageLevel

rdd = sc.parallelize(range(1000000), 4)
mapped = rdd.map(lambda x: x * x * x)

# 1. MEMORY_ONLY（默认）
# 仅内存存储，速度快但占用内存
mapped.cache()  # 等价于 persist(StorageLevel.MEMORY_ONLY)

# 2. MEMORY_AND_DISK
# 内存不足时溢写到磁盘
mapped.persist(StorageLevel.MEMORY_AND_DISK)

# 3. MEMORY_ONLY_SER
# 序列化存储（节省内存，但读取时需要反序列化）
mapped.persist(StorageLevel.MEMORY_ONLY_SER)

# 4. DISK_ONLY
# 仅磁盘存储
mapped.persist(StorageLevel.DISK_ONLY)

# 触发缓存
mapped.count()

# 再次使用时会从缓存读取
mapped.reduce(lambda x, y: x + y)

# 取消持久化
mapped.unpersist()

# 检查点（Checkpoint）- 切断血缘，适合迭代算法
sc.setCheckpointDir("/tmp/checkpoints")
long_lineage_rdd = sc.parallelize([1])
for i in range(100):
    long_lineage_rdd = long_lineage_rdd.map(lambda x: x + 1)
    if i % 10 == 0:
        long_lineage_rdd.checkpoint()  # 每10步切断一次血缘

避坑小贴士

• RDD的转换操作是惰性求值的，只有遇到行动操作才会真正执行
• collect()操作会将所有数据拉取到驱动程序，大数据量时慎用，可能导致内存溢出
• 频繁使用collect()会影响性能，尽量在RDD层面完成计算
• 避免使用groupByKey：优先使用reduceByKey或aggregateByKey进行预聚合
• 合理设置分区数：太少会导致任务过大，太多会增加调度开销，一般每个分区100MB-1GB
• 及时unpersist()：缓存的RDD不再使用时及时释放内存
• 慎用cartesian：笛卡尔积会产生N*M的数据量，极易导致OOM

---

4. Spark性能优化专题

4.1 广播变量与累加器

广播变量（Broadcast Variables）

当需要在所有节点上使用同一份较大数据时，使用广播变量可以避免数据重复传输。

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("广播变量示例").getOrCreate()
sc = spark.sparkContext

# 场景：将小表广播到大表进行Join
# 大表（假设有1亿条记录）
large_data = [(i, f"user_{i}", i % 100) for i in range(10000)]
large_rdd = sc.parallelize(large_data, 10)

# 小表（假设只有100条记录）
small_lookup = {i: f"category_{i}" for i in range(100)}

# 不使用广播（每个任务都传输小表）
def join_without_broadcast(record):
    user_id, name, cat_id = record
    return (user_id, name, small_lookup.get(cat_id, "unknown"))

# 使用广播（只传输一次，所有任务共享）
broadcast_lookup = sc.broadcast(small_lookup)

def join_with_broadcast(record):
    user_id, name, cat_id = record
    return (user_id, name, broadcast_lookup.value.get(cat_id, "unknown"))

# 对比性能
import time

start = time.time()
result1 = large_rdd.map(join_without_broadcast).collect()
print(f"不使用广播耗时: {time.time() - start:.3f}秒")

start = time.time()
result2 = large_rdd.map(join_with_broadcast).collect()
print(f"使用广播耗时: {time.time() - start:.3f}秒")

# 释放广播变量
broadcast_lookup.unpersist()

累加器（Accumulators）

累加器用于在任务中进行全局计数或求和，只能在Driver端读取。

# 使用累加器统计异常数据
error_count = sc.accumulator(0)
total_count = sc.accumulator(0)

def process_with_counter(record):
    global error_count, total_count
    total_count += 1
    try:
        # 处理逻辑
        result = int(record) * 2
        return result
    except:
        error_count += 1
        return None

data = sc.parallelize(["1", "2", "abc", "4", "5", "xyz"])
results = data.map(process_with_counter).filter(lambda x: x is not None).collect()

print(f"总记录数: {total_count.value}")
print(f"错误记录数: {error_count.value}")
print(f"错误率: {error_count.value / total_count.value:.2%}")

# 自定义累加器
from pyspark.accumulators import AccumulatorParam

class VectorAccumulatorParam(AccumulatorParam):
    def zero(self, value):
        return [0] * len(value)
  
    def addInPlace(self, val1, val2):
        for i in range(len(val1)):
            val1[i] += val2[i]
        return val1

vector_acc = sc.accumulator([0, 0, 0], VectorAccumulatorParam())

4.2 数据倾斜处理

数据倾斜是指某些Key的数据量远大于其他Key，导致部分任务执行时间过长。

# 识别数据倾斜
from pyspark.sql import functions as F

# 方法1: 查看Key分布
key_counts = df.groupBy("key_column").count().orderBy(F.desc("count"))
key_counts.show(20)

# 方法2: 计算倾斜度
total_count = df.count()
distinct_keys = df.select("key_column").distinct().count()
avg_per_key = total_count / distinct_keys

skewed_keys = df.groupBy("key_column") \
    .count() \
    .filter(F.col("count") > avg_per_key * 10) \
    .select("key_column") \
    .rdd.map(lambda x: x[0]) \
    .collect()

print(f"倾斜的Key: {skewed_keys}")

解决方案1: 加盐（Salting）

import random

# 给倾斜的Key添加随机前缀
SALT_COUNT = 10

def add_salt(key):
    if key in skewed_keys:
        return f"{key}_{random.randint(0, SALT_COUNT - 1)}"
    return key

# 两阶段聚合
# 第一阶段：局部聚合（加盐）
phase1 = rdd.map(lambda x: (add_salt(x[0]), x[1])) \
    .reduceByKey(lambda x, y: x + y)

# 第二阶段：去除盐，全局聚合
phase2 = phase1.map(lambda x: (x[0].split("_")[0], x[1])) \
    .reduceByKey(lambda x, y: x + y)

解决方案2: 拆分倾斜Key

# 将倾斜的Key单独处理
normal_keys_rdd = rdd.filter(lambda x: x[0] not in skewed_keys)
skewed_keys_rdd = rdd.filter(lambda x: x[0] in skewed_keys)

# 对倾斜Key使用mapJoin（广播小表）
broadcast_small = sc.broadcast(small_table_dict)
skewed_result = skewed_keys_rdd.map(lambda x: (x[0], (x[1], broadcast_small.value.get(x[0]))))

# 正常Key使用reduceByKey
normal_result = normal_keys_rdd.reduceByKey(lambda x, y: x + y)

# 合并结果
final_result = normal_result.union(skewed_result)

4.3 Spark UI监控与调优

# 配置Spark以启用历史服务器和详细日志
spark = SparkSession.builder \
    .appName("监控示例") \
    .config("spark.eventLog.enabled", "true") \
    .config("spark.eventLog.dir", "/tmp/spark-events") \
    .config("spark.history.fs.logDirectory", "/tmp/spark-events") \
    .getOrCreate()

# 获取Spark UI链接
print(f"Spark UI: http://localhost:4040")

# 关键指标解读
"""
Jobs页面：
- 查看Job执行时间和状态
- 识别慢任务

Stages页面：
- 查看Stage的输入/输出数据量
- 识别Shuffle读写量
- 查看任务执行时间分布

Tasks页面：
- 查看每个任务的执行时间
- 识别数据倾斜（某些任务时间明显更长）
- 查看GC时间和Shuffle时间

Storage页面：
- 查看缓存的RDD/DataFrame
- 内存和磁盘使用情况
"""

# 代码中获取度量信息
sc.statusTracker().getExecutorInfos()

4.4 内存与序列化优化

# 配置序列化方式
spark = SparkSession.builder \
    .appName("序列化优化") \
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") \
    .config("spark.kryo.registrationRequired", "false") \
    .getOrCreate()

# Kryo序列化比Java序列化更快更紧凑
# 对于自定义类，需要注册
# conf.registerKryoClasses([MyClass.class])

# 内存配置
spark = SparkSession.builder \
    .appName("内存优化") \
    .config("spark.executor.memory", "4g") \
    .config("spark.executor.memoryOverhead", "512m") \
    .config("spark.driver.memory", "2g") \
    .config("spark.memory.fraction", "0.8") \
    .config("spark.memory.storageFraction", "0.3") \
    .getOrCreate()

"""
内存配置说明：
- executor.memory: Executor堆内存
- executor.memoryOverhead: Executor非堆内存（用于PySpark等）
- memory.fraction: 用于执行和存储的内存比例（默认0.6）
- memory.storageFraction: 存储内存占fraction的比例（默认0.5）
"""

---

5. Spark Streaming流处理

5.1 DStream基础

from pyspark.streaming import StreamingContext

# 创建StreamingContext，批次间隔2秒
ssc = StreamingContext(sc, 2)

# 从TCP socket接收数据
lines = ssc.socketTextStream("localhost", 9999)

# 处理逻辑：词频统计
words = lines.flatMap(lambda line: line.split(" "))
pairs = words.map(lambda word: (word, 1))
word_counts = pairs.reduceByKey(lambda x, y: x + y)

# 输出结果
word_counts.pprint()

# 启动流处理
ssc.start()
ssc.awaitTermination()

5.2 窗口操作

# 创建带检查点的StreamingContext
ssc = StreamingContext(sc, 1)
ssc.checkpoint("/tmp/streaming-checkpoint")

# 从Kafka接收数据（需要spark-streaming-kafka包）
# from pyspark.streaming.kafka import KafkaUtils
# kafka_stream = KafkaUtils.createStream(ssc, "zookeeper:2181", "consumer-group", {"topic": 1})

lines = ssc.socketTextStream("localhost", 9999)
words = lines.flatMap(lambda line: line.split(" "))
pairs = words.map(lambda word: (word, 1))

# 滑动窗口操作
# 窗口大小30秒，滑动间隔10秒
windowed_counts = pairs.reduceByKeyAndWindow(
    lambda x, y: x + y,  # 聚合函数
    lambda x, y: x - y,  # 逆函数（用于增量计算）
    windowDuration=30,
    slideDuration=10
)

windowed_counts.pprint()

ssc.start()
ssc.awaitTermination()

5.3 Structured Streaming

from pyspark.sql import SparkSession
from pyspark.sql.functions import explode, split, window

spark = SparkSession.builder \
    .appName("StructuredStreaming") \
    .getOrCreate()

# 读取流数据（从socket）
lines = spark.readStream \
    .format("socket") \
    .option("host", "localhost") \
    .option("port", 9999) \
    .load()

# 处理逻辑
words = lines.select(
    explode(split(lines.value, " ")).alias("word")
)
word_counts = words.groupBy("word").count()

# 输出到控制台
query = word_counts.writeStream \
    .outputMode("complete") \
    .format("console") \
    .start()

query.awaitTermination()

Structured Streaming输出模式

# 1. Complete Mode：输出完整结果表（适用于聚合查询）
agg_query = word_counts.writeStream \
    .outputMode("complete") \
    .format("console") \
    .start()

# 2. Append Mode：只输出新增行（适用于无聚合查询）
append_query = words.writeStream \
    .outputMode("append") \
    .format("console") \
    .start()

# 3. Update Mode：只输出更新的行
update_query = word_counts.writeStream \
    .outputMode("update") \
    .format("console") \
    .start()

流数据Join

# 流-静态Join
static_df = spark.read.parquet("/path/to/static/data")
stream_df = spark.readStream.format("kafka").option("subscribe", "topic").load()

joined = stream_df.join(static_df, "join_key")

# 流-流Join（需要水印）
from pyspark.sql.functions import expr

stream1 = spark.readStream.format("kafka").option("subscribe", "topic1").load()
stream2 = spark.readStream.format("kafka").option("subscribe", "topic2").load()

# 添加水印处理延迟数据
stream1_with_watermark = stream1.withWatermark("timestamp", "10 minutes")
stream2_with_watermark = stream2.withWatermark("timestamp", "10 minutes")

joined_streams = stream1_with_watermark.join(
    stream2_with_watermark,
    expr("""
        join_key = join_key AND
        timestamp1 >= timestamp2 - interval 5 minutes AND
        timestamp1 <= timestamp2 + interval 5 minutes
    """),
    "inner"
)

---

6. DataFrame与Spark SQL

4.1 DataFrame简介

DataFrame是Spark 2.0+推荐的编程接口，以命名列的方式组织分布式数据，类似于pandas的DataFrame，但支持分布式计算。

4.2 创建DataFrame

from pyspark.sql import SparkSession
from pyspark.sql import Row

spark = SparkSession.builder.appName("DataFrame示例").getOrCreate()

# 从RDD创建
rdd = spark.sparkContext.parallelize([
    Row(name="张三", age=25, city="北京"),
    Row(name="李四", age=30, city="上海"),
    Row(name="王五", age=28, city="广州")
])
df = spark.createDataFrame(rdd)

# 从列表创建
data = [("张三", 25, "北京"), ("李四", 30, "上海")]
columns = ["name", "age", "city"]
df = spark.createDataFrame(data, columns)

# 从JSON/CSV文件创建
df_json = spark.read.json("data.json")
df_csv = spark.read.csv("data.csv", header=True, inferSchema=True)

4.3 DataFrame常用操作

# 显示数据
df.show()

# 查看Schema
df.printSchema()

# 选择列
df.select("name", "age").show()

# 过滤数据
df.filter(df.age > 25).show()
df.filter("age > 25").show()

# 分组聚合
df.groupBy("city").count().show()
df.groupBy("city").agg({"age": "avg"}).show()

# 排序
df.orderBy(df.age.desc()).show()

4.4 Spark SQL

# 注册临时视图
df.createOrReplaceTempView("people")

# 执行SQL查询
result = spark.sql("""
    SELECT city, AVG(age) as avg_age
    FROM people
    GROUP BY city
    ORDER BY avg_age DESC
""")
result.show()

---

5. Spark MLlib机器学习

5.1 MLlib简介

MLlib是Spark的可扩展机器学习库，提供常用的机器学习算法和工具。

5.2 特征工程

from pyspark.ml.feature import VectorAssembler, StandardScaler

# 特征向量化
assembler = VectorAssembler(
    inputCols=["feature1", "feature2", "feature3"],
    outputCol="features"
)
df_vector = assembler.transform(df)

# 标准化
scaler = StandardScaler(
    inputCol="features",
    outputCol="scaledFeatures"
)
scaler_model = scaler.fit(df_vector)
df_scaled = scaler_model.transform(df_vector)

5.3 机器学习示例（逻辑回归）

from pyspark.ml.classification import LogisticRegression
from pyspark.ml.evaluation import BinaryClassificationEvaluator

# 划分训练集和测试集
train_df, test_df = df_scaled.randomSplit([0.8, 0.2], seed=42)

# 创建模型
lr = LogisticRegression(
    featuresCol="scaledFeatures",
    labelCol="label",
    maxIter=10
)

# 训练模型
model = lr.fit(train_df)

# 预测
predictions = model.transform(test_df)

# 评估
evaluator = BinaryClassificationEvaluator()
accuracy = evaluator.evaluate(predictions)
print(f"模型准确率: {accuracy}")

---

6. 实战项目：电商用户行为分析

本项目将使用PySpark完成一个完整的电商用户行为分析系统，涵盖数据清洗、用户画像、RFM分析和推荐系统。

6.1 项目概述与数据准备

项目结构
├── data/
│   ├── user_behavior.csv    # 用户行为数据
│   ├── user_info.csv        # 用户基本信息
│   └── product_info.csv     # 商品信息
├── src/
│   ├── data_cleaning.py     # 数据清洗
│   ├── user_profile.py      # 用户画像
│   ├── rfm_analysis.py      # RFM分析
│   └── recommendation.py    # 推荐系统
└── output/                  # 输出结果

数据生成脚本

import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import random

# 设置随机种子
np.random.seed(42)
random.seed(42)

# 生成用户行为数据
def generate_user_behavior(n_records=100000):
    """生成用户行为日志"""
    user_ids = np.random.randint(1000, 1100, n_records)  # 100个用户
    product_ids = np.random.randint(10000, 10500, n_records)  # 500个商品
  
    # 行为类型：浏览、收藏、加购、购买
    behaviors = np.random.choice(
        ['view', 'fav', 'cart', 'buy'],
        n_records,
        p=[0.7, 0.1, 0.15, 0.05]
    )
  
    # 生成时间戳（最近30天）
    base_time = datetime(2024, 1, 1)
    timestamps = [
        base_time + timedelta(
            days=random.randint(0, 29),
            hours=random.randint(0, 23),
            minutes=random.randint(0, 59)
        )
        for _ in range(n_records)
    ]
  
    # 生成金额（购买行为才有）
    amounts = [
        round(random.uniform(10, 1000), 2) if b == 'buy' else 0
        for b in behaviors
    ]
  
    df = pd.DataFrame({
        'user_id': user_ids,
        'product_id': product_ids,
        'behavior': behaviors,
        'timestamp': timestamps,
        'amount': amounts
    })
  
    return df

# 生成用户基本信息
def generate_user_info():
    """生成用户基本信息"""
    user_ids = range(1000, 1100)
    ages = np.random.randint(18, 60, 100)
    genders = np.random.choice(['M', 'F'], 100)
    cities = np.random.choice(['北京', '上海', '广州', '深圳', '杭州'], 100)
    register_dates = [
        datetime(2023, 1, 1) + timedelta(days=random.randint(0, 365))
        for _ in range(100)
    ]
  
    df = pd.DataFrame({
        'user_id': user_ids,
        'age': ages,
        'gender': genders,
        'city': cities,
        'register_date': register_dates
    })
  
    return df

# 生成商品信息
def generate_product_info():
    """生成商品信息"""
    product_ids = range(10000, 10500)
    categories = np.random.choice(
        ['电子产品', '服装', '食品', '家居', '图书'],
        500
    )
    prices = np.random.uniform(10, 1000, 500).round(2)
    brands = np.random.choice(['品牌A', '品牌B', '品牌C', '品牌D', '品牌E'], 500)
  
    df = pd.DataFrame({
        'product_id': product_ids,
        'category': categories,
        'price': prices,
        'brand': brands
    })
  
    return df

# 保存数据
if __name__ == "__main__":
    behavior_df = generate_user_behavior(100000)
    user_df = generate_user_info()
    product_df = generate_product_info()
  
    behavior_df.to_csv("data/user_behavior.csv", index=False)
    user_df.to_csv("data/user_info.csv", index=False)
    product_df.to_csv("data/product_info.csv", index=False)
  
    print("数据生成完成！")

6.2 数据清洗与ETL

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.types import *

# 初始化Spark
spark = SparkSession.builder \
    .appName("电商用户行为分析") \
    .config("spark.sql.adaptive.enabled", "true") \
    .getOrCreate()

# 读取数据
behavior_df = spark.read.csv("data/user_behavior.csv", header=True, inferSchema=True)
user_df = spark.read.csv("data/user_info.csv", header=True, inferSchema=True)
product_df = spark.read.csv("data/product_info.csv", header=True, inferSchema=True)

print("原始数据统计：")
print(f"行为数据: {behavior_df.count()} 条")
print(f"用户数据: {user_df.count()} 条")
print(f"商品数据: {product_df.count()} 条")

# 数据清洗
class DataCleaner:
    def __init__(self, df):
        self.df = df
        self.cleaning_log = []
  
    def remove_duplicates(self):
        """去除重复数据"""
        before_count = self.df.count()
        self.df = self.df.dropDuplicates()
        after_count = self.df.count()
        self.cleaning_log.append(f"去除重复: {before_count - after_count} 条")
        return self
  
    def handle_missing_values(self):
        """处理缺失值"""
        # 查看缺失值统计
        missing_stats = []
        for col_name in self.df.columns:
            null_count = self.df.filter(F.col(col_name).isNull()).count()
            if null_count > 0:
                missing_stats.append(f"{col_name}: {null_count}")
      
        if missing_stats:
            self.cleaning_log.append(f"缺失值统计: {', '.join(missing_stats)}")
            # 删除包含缺失值的行
            self.df = self.df.dropna()
        else:
            self.cleaning_log.append("无缺失值")
        return self
  
    def filter_invalid_data(self):
        """过滤无效数据"""
        before_count = self.df.count()
      
        # 过滤金额为负的记录（购买行为）
        self.df = self.df.filter(
            ~((F.col("behavior") == "buy") & (F.col("amount") <= 0))
        )
      
        after_count = self.df.count()
        self.cleaning_log.append(f"过滤无效数据: {before_count - after_count} 条")
        return self
  
    def standardize_timestamp(self):
        """标准化时间戳"""
        self.df = self.df.withColumn(
            "timestamp",
            F.to_timestamp(F.col("timestamp"))
        ).withColumn(
            "date",
            F.to_date(F.col("timestamp"))
        ).withColumn(
            "hour",
            F.hour(F.col("timestamp"))
        )
        self.cleaning_log.append("时间戳标准化完成")
        return self
  
    def get_clean_data(self):
        return self.df
  
    def get_log(self):
        return self.cleaning_log

# 执行数据清洗
cleaner = DataCleaner(behavior_df)
cleaned_behavior = cleaner.remove_duplicates() \
    .handle_missing_values() \
    .filter_invalid_data() \
    .standardize_timestamp() \
    .get_clean_data()

print("\n数据清洗日志：")
for log in cleaner.get_log():
    print(f"  - {log}")

print(f"\n清洗后数据: {cleaned_behavior.count()} 条")

6.3 用户画像构建

class UserProfileBuilder:
    def __init__(self, behavior_df, user_df, product_df):
        self.behavior_df = behavior_df
        self.user_df = user_df
        self.product_df = product_df
  
    def build_behavior_features(self):
        """构建行为特征"""
        # 统计各类行为次数
        behavior_pivot = self.behavior_df.groupBy("user_id").pivot("behavior").count().fillna(0)
        behavior_pivot = behavior_pivot.select(
            "user_id",
            F.col("view").alias("view_count"),
            F.col("fav").alias("fav_count"),
            F.col("cart").alias("cart_count"),
            F.col("buy").alias("buy_count")
        )
      
        # 计算转化率
        behavior_features = behavior_pivot.withColumn(
            "fav_rate", F.col("fav_count") / F.col("view_count")
        ).withColumn(
            "cart_rate", F.col("cart_count") / F.col("view_count")
        ).withColumn(
            "buy_rate", F.col("buy_count") / F.col("view_count")
        ).fillna(0)
      
        return behavior_features
  
    def build_consumption_features(self):
        """构建消费特征"""
        consumption = self.behavior_df.filter(F.col("behavior") == "buy") \
            .groupBy("user_id") \
            .agg(
                F.sum("amount").alias("total_amount"),
                F.avg("amount").alias("avg_amount"),
                F.count("*").alias("order_count"),
                F.max("amount").alias("max_amount"),
                F.min("amount").alias("min_amount")
            ).fillna(0)
      
        return consumption
  
    def build_preference_features(self):
        """构建偏好特征"""
        # 关联商品信息
        behavior_with_product = self.behavior_df.join(
            self.product_df,
            "product_id",
            "left"
        )
      
        # 统计各类目偏好
        category_pref = behavior_with_product.groupBy("user_id", "category") \
            .count() \
            .groupBy("user_id") \
            .pivot("category") \
            .sum("count") \
            .fillna(0)
      
        # 找出最偏好的类目
        category_cols = [c for c in category_pref.columns if c != "user_id"]
        category_pref = category_pref.withColumn(
            "favorite_category",
            F.greatest(*category_cols)
        )
      
        return category_pref
  
    def build_activity_features(self):
        """构建活跃度特征"""
        activity = self.behavior_df.groupBy("user_id").agg(
            F.countDistinct("date").alias("active_days"),
            F.min("date").alias("first_active"),
            F.max("date").alias("last_active"),
            F.count("*").alias("total_actions")
        )
      
        # 计算活跃天数占比（假设数据跨度30天）
        activity = activity.withColumn(
            "activity_rate", F.col("active_days") / 30
        )
      
        return activity
  
    def build_full_profile(self):
        """构建完整用户画像"""
        behavior_features = self.build_behavior_features()
        consumption_features = self.build_consumption_features()
        preference_features = self.build_preference_features()
        activity_features = self.build_activity_features()
      
        # 合并所有特征
        profile = self.user_df \
            .join(behavior_features, "user_id", "left") \
            .join(consumption_features, "user_id", "left") \
            .join(preference_features, "user_id", "left") \
            .join(activity_features, "user_id", "left") \
            .fillna(0)
      
        return profile

# 构建用户画像
profile_builder = UserProfileBuilder(cleaned_behavior, user_df, product_df)
user_profile = profile_builder.build_full_profile()

print("用户画像字段：")
for col in user_profile.columns:
    print(f"  - {col}")

user_profile.show(5)

6.4 RFM模型分析

from pyspark.sql import Window

class RFMAnalysis:
    def __init__(self, behavior_df, reference_date=None):
        self.behavior_df = behavior_df
        self.reference_date = reference_date or F.current_date()
  
    def calculate_rfm(self):
        """计算RFM指标"""
        # 只考虑购买行为
        purchase_df = self.behavior_df.filter(F.col("behavior") == "buy")
      
        # R: 最近一次购买距今天数
        recency_df = purchase_df.groupBy("user_id").agg(
            F.max("date").alias("last_purchase_date")
        ).withColumn(
            "recency",
            F.datediff(self.reference_date, F.col("last_purchase_date"))
        )
      
        # F: 购买频率
        frequency_df = purchase_df.groupBy("user_id").count() \
            .withColumnRenamed("count", "frequency")
      
        # M: 购买金额
        monetary_df = purchase_df.groupBy("user_id").agg(
            F.sum("amount").alias("monetary")
        )
      
        # 合并RFM
        rfm = recency_df.join(frequency_df, "user_id", "outer") \
            .join(monetary_df, "user_id", "outer") \
            .fillna(0)
      
        return rfm
  
    def score_rfm(self, rfm_df):
        """RFM评分（1-5分）"""
        # 使用分位数进行评分
        rfm_stats = rfm_df.select(
            F.percentile_approx("recency", [0.2, 0.4, 0.6, 0.8]).alias("r_percentiles"),
            F.percentile_approx("frequency", [0.2, 0.4, 0.6, 0.8]).alias("f_percentiles"),
            F.percentile_approx("monetary", [0.2, 0.4, 0.6, 0.8]).alias("m_percentiles")
        ).collect()[0]
      
        r_thresholds = rfm_stats["r_percentiles"]
        f_thresholds = rfm_stats["f_percentiles"]
        m_thresholds = rfm_stats["m_percentiles"]
      
        # R评分（越小越好，所以反向）
        rfm_scored = rfm_df.withColumn(
            "r_score",
            F.when(F.col("recency") <= r_thresholds[0], 5)
            .when(F.col("recency") <= r_thresholds[1], 4)
            .when(F.col("recency") <= r_thresholds[2], 3)
            .when(F.col("recency") <= r_thresholds[3], 2)
            .otherwise(1)
        )
      
        # F评分（越大越好）
        rfm_scored = rfm_scored.withColumn(
            "f_score",
            F.when(F.col("frequency") >= f_thresholds[3], 5)
            .when(F.col("frequency") >= f_thresholds[2], 4)
            .when(F.col("frequency") >= f_thresholds[1], 3)
            .when(F.col("frequency") >= f_thresholds[0], 2)
            .otherwise(1)
        )
      
        # M评分（越大越好）
        rfm_scored = rfm_scored.withColumn(
            "m_score",
            F.when(F.col("monetary") >= m_thresholds[3], 5)
            .when(F.col("monetary") >= m_thresholds[2], 4)
            .when(F.col("monetary") >= m_thresholds[1], 3)
            .when(F.col("monetary") >= m_thresholds[0], 2)
            .otherwise(1)
        )
      
        # 综合RFM得分
        rfm_scored = rfm_scored.withColumn(
            "rfm_score",
            F.concat(F.col("r_score"), F.col("f_score"), F.col("m_score"))
        )
      
        return rfm_scored
  
    def segment_users(self, rfm_scored):
        """用户分群"""
        segments = rfm_scored.withColumn(
            "segment",
            F.when((F.col("r_score") >= 4) & (F.col("f_score") >= 4) & (F.col("m_score") >= 4), "重要价值客户")
            .when((F.col("r_score") >= 4) & (F.col("f_score") >= 4) & (F.col("m_score") < 4), "一般价值客户")
            .when((F.col("r_score") >= 4) & (F.col("f_score") < 4) & (F.col("m_score") >= 4), "重要发展客户")
            .when((F.col("r_score") >= 4) & (F.col("f_score") < 4) & (F.col("m_score") < 4), "新客户")
            .when((F.col("r_score") < 4) & (F.col("f_score") >= 4) & (F.col("m_score") >= 4), "重要保持客户")
            .when((F.col("r_score") < 4) & (F.col("f_score") >= 4) & (F.col("m_score") < 4), "一般保持客户")
            .when((F.col("r_score") < 4) & (F.col("f_score") < 4) & (F.col("m_score") >= 4), "重要挽留客户")
            .otherwise("流失客户")
        )
      
        return segments

# 执行RFM分析
rfm_analyzer = RFMAnalysis(cleaned_behavior)
rfm = rfm_analyzer.calculate_rfm()
rfm_scored = rfm_analyzer.score_rfm(rfm)
rfm_segments = rfm_analyzer.segment_users(rfm_scored)

print("RFM分析结果：")
rfm_segments.select("user_id", "recency", "frequency", "monetary", 
                    "r_score", "f_score", "m_score", "rfm_score", "segment").show(10)

# 分群统计
segment_stats = rfm_segments.groupBy("segment").agg(
    F.count("*").alias("user_count"),
    F.avg("monetary").alias("avg_monetary"),
    F.avg("frequency").alias("avg_frequency")
).orderBy(F.desc("user_count"))

print("\n用户分群统计：")
segment_stats.show()

6.5 协同过滤推荐

from pyspark.ml.feature import StringIndexer
from pyspark.ml.recommendation import ALS
from pyspark.ml.evaluation import RegressionEvaluator

class RecommendationEngine:
    def __init__(self, behavior_df):
        self.behavior_df = behavior_df
        self.model = None
  
    def prepare_data(self):
        """准备评分数据"""
        # 将行为转换为评分权重
        behavior_weights = {
            'view': 1,
            'fav': 3,
            'cart': 5,
            'buy': 10
        }
      
        # 计算用户对商品的评分
        ratings = self.behavior_df.groupBy("user_id", "product_id").agg(
            F.sum(
                F.when(F.col("behavior") == "view", behavior_weights['view'])
                .when(F.col("behavior") == "fav", behavior_weights['fav'])
                .when(F.col("behavior") == "cart", behavior_weights['cart'])
                .when(F.col("behavior") == "buy", behavior_weights['buy'])
            ).alias("rating")
        ).filter(F.col("rating") > 0)
      
        # 索引化用户ID和商品ID
        user_indexer = StringIndexer(inputCol="user_id", outputCol="user_idx")
        product_indexer = StringIndexer(inputCol="product_id", outputCol="product_idx")
      
        ratings_indexed = user_indexer.fit(ratings).transform(ratings)
        ratings_indexed = product_indexer.fit(ratings_indexed).transform(ratings_indexed)
      
        return ratings_indexed.select(
            F.col("user_idx").cast("int"),
            F.col("product_idx").cast("int"),
            F.col("rating").cast("float")
        )
  
    def train_model(self, ratings_df):
        """训练ALS模型"""
        # 划分训练集和测试集
        train, test = ratings_df.randomSplit([0.8, 0.2], seed=42)
      
        # 创建ALS模型
        als = ALS(
            maxIter=10,
            regParam=0.01,
            userCol="user_idx",
            itemCol="product_idx",
            ratingCol="rating",
            coldStartStrategy="drop"
        )
      
        # 训练模型
        self.model = als.fit(train)
      
        # 评估模型
        predictions = self.model.transform(test)
        evaluator = RegressionEvaluator(
            metricName="rmse",
            labelCol="rating",
            predictionCol="prediction"
        )
        rmse = evaluator.evaluate(predictions)
        print(f"模型RMSE: {rmse}")
      
        return self.model
  
    def get_recommendations(self, n_recommendations=10):
        """为所有用户生成推荐"""
        if self.model is None:
            raise ValueError("请先训练模型")
      
        # 为每个用户推荐商品
        user_recs = self.model.recommendForAllUsers(n_recommendations)
      
        return user_recs
  
    def get_similar_products(self, product_idx, n_similar=10):
        """获取相似商品"""
        if self.model is None:
            raise ValueError("请先训练模型")
      
        # 获取商品特征向量
        product_factors = self.model.itemFactors
      
        # 计算余弦相似度
        target_vector = product_factors.filter(F.col("id") == product_idx).collect()[0]["features"]
      
        # 广播目标向量
        broadcast_vector = spark.sparkContext.broadcast(target_vector)
      
        def cosine_similarity(features):
            import numpy as np
            v1 = np.array(broadcast_vector.value)
            v2 = np.array(features)
            return float(np.dot(v1, v2) / (np.linalg.norm(v1) * np.linalg.norm(v2)))
      
        from pyspark.sql.types import FloatType
        similarity_udf = F.udf(cosine_similarity, FloatType())
      
        similar_products = product_factors \
            .filter(F.col("id") != product_idx) \
            .withColumn("similarity", similarity_udf(F.col("features"))) \
            .orderBy(F.desc("similarity")) \
            .limit(n_similar)
      
        return similar_products

# 训练推荐模型
rec_engine = RecommendationEngine(cleaned_behavior)
ratings = rec_engine.prepare_data()
rec_engine.train_model(ratings)

# 生成推荐
recommendations = rec_engine.get_recommendations(n_recommendations=5)
print("\n用户推荐结果（前5个用户）：")
recommendations.show(5, truncate=False)

6.6 结果输出与可视化

# 保存分析结果
def save_results():
    """保存所有分析结果"""
    # 用户画像
    user_profile.write.parquet("output/user_profile.parquet")
  
    # RFM分群结果
    rfm_segments.write.parquet("output/rfm_segments.parquet")
  
    # 推荐结果
    recommendations.write.parquet("output/recommendations.parquet")
  
    # 导出为CSV（小数据量）
    segment_stats.toPandas().to_csv("output/segment_stats.csv", index=False)
  
    print("结果已保存到output目录")

# 生成分析报告
def generate_report():
    """生成分析报告"""
    report = []
    report.append("=" * 50)
    report.append("电商用户行为分析报告")
    report.append("=" * 50)
  
    # 数据概览
    report.append(f"\n【数据概览】")
    report.append(f"  总用户数: {user_df.count()}")
    report.append(f"  总商品数: {product_df.count()}")
    report.append(f"  总行为记录: {cleaned_behavior.count()}")
  
    # 行为统计
    behavior_stats = cleaned_behavior.groupBy("behavior").count().collect()
    report.append(f"\n【行为分布】")
    for row in behavior_stats:
        report.append(f"  {row['behavior']}: {row['count']}")
  
    # RFM分群
    report.append(f"\n【用户分群】")
    for row in segment_stats.collect():
        report.append(f"  {row['segment']}: {row['user_count']}人")
  
    report.append("\n" + "=" * 50)
  
    return "\n".join(report)

# 打印报告
print(generate_report())

# 关闭Spark
spark.stop()

---

7. 本章小结

本章介绍了PySpark大数据处理的基础知识：

1. Spark架构：理解Driver、Executor和Cluster Manager的关系
2. RDD编程：掌握转换操作和行动操作的区别与使用
3. DataFrame API：使用结构化数据进行高效处理
4. Spark SQL：通过SQL语句查询数据
5. MLlib：使用分布式机器学习算法

大数据处理的核心思想是分而治之——将大任务拆分成小任务，在多台机器上并行执行，最后汇总结果。PySpark让Python开发者能够轻松驾驭大数据处理。

---

本章内容编写中

本章正在持续完善中，将涵盖以下内容：

• PySpark基础概念与环境配置详解
• RDD高级操作与性能优化
• DataFrame API深入实践
• Spark SQL复杂查询案例
• 分布式机器学习实战项目
• Spark Streaming流处理入门
• 生产环境部署与调优技巧

敬请期待完整内容更新…

---

如果你觉得本章内容对你有帮助，欢迎点赞、收藏、评论交流！