Apache Spark 使用教程：从入门到精通

# 1. 安装 JDK 17
sudo apt update
sudo apt install openjdk-17-jdk
java -version  # 确认输出 17.x

# 2. 下载 Spark 4.0 预编译包（以 4.0.0 为例，with Hadoop 3）
wget https://archive.apache.org/dist/spark/spark-4.0.0/spark-4.0.0-bin-hadoop3.tgz
tar -xzf spark-4.0.0-bin-hadoop3.tgz
sudo mv spark-4.0.0-bin-hadoop3 /opt/spark

# 3. 配置环境变量
echo 'export SPARK_HOME=/opt/spark' >> ~/.bashrc
echo 'export PATH=$PATH:$SPARK_HOME/bin:$SPARK_HOME/sbin' >> ~/.bashrc
source ~/.bashrc

# 4. 验证安装
spark-shell --version

pip install pyspark==4.0.0

pip install pyspark-client

from pyspark.sql import SparkSession

# 创建 SparkSession（Spark 2.0+ 的统一入口）
spark = SparkSession.builder \
    .appName("WordCount") \
    .master("local[*]") \
    .getOrCreate()

# 读取文本文件 -> 分词 -> 计数
counts = spark.read.text("README.md") \
    .selectExpr("explode(split(value, ' ')) as word") \
    .filter("word != ''") \
    .groupBy("word") \
    .count() \
    .orderBy("count", ascending=False)

counts.show(20, truncate=False)

spark.stop()

val spark = SparkSession.builder().appName("WordCount").getOrCreate()

val counts = spark.read.textFile("README.md")
  .flatMap(line => line.split(" "))
  .filter(_ != "")
  .groupBy("value")
  .count()
  .orderBy(desc("count"))

counts.show(20)

-- 在 spark-sql 中注册临时视图后查询
CREATE OR REPLACE TEMP VIEW words AS
SELECT explode(split(value, ' ')) AS word FROM text_data;

SELECT word, COUNT(*) AS cnt
FROM words
WHERE word != ''
GROUP BY word
ORDER BY cnt DESC
LIMIT 20;

from pyspark import SparkContext

sc = SparkContext("local[*]", "RDDDemo")

# 方式一：从集合并行化创建
rdd1 = sc.parallelize([1, 2, 3, 4, 5, 6, 7, 8, 9, 10])

# 方式二：从外部存储读取
rdd2 = sc.textFile("hdfs://localhost:9000/data/log.txt")

# 查看分区数
print(rdd1.getNumPartitions())  # local[*] 模式下默认为 CPU 核数

data = sc.parallelize(range(1, 11))

# 以下都是 Transformation，不会触发计算
evens = data.filter(lambda x: x % 2 == 0)        # [2, 4, 6, 8, 10]
squared = evens.map(lambda x: x ** 2)              # [4, 16, 36, 64, 100]

# Action 触发计算
print(squared.collect())   # [4, 16, 36, 64, 100]
print(squared.count())     # 5
print(squared.reduce(lambda a, b: a + b))  # 220

rdd = sc.textFile("data/large_file.txt") \
    .filter(lambda line: "error" in line.lower())

# 缓存到内存
rdd.cache()  # 等价于 rdd.persist(StorageLevel.MEMORY_ONLY)

# 第一次使用：触发计算并缓存
error_count = rdd.count()

# 第二次使用：直接从内存读取，跳过重新读取文件和过滤
sample_errors = rdd.take(10)

# 广播一份城市编码映射表
city_map = {1: "北京", 2: "上海", 3: "广州", 4: "深圳"}
broadcast_city = sc.broadcast(city_map)

rdd = sc.parallelize([(1, "张三"), (2, "李四"), (3, "王五")])
result = rdd.map(lambda x: (x[1], broadcast_city.value[x[0]]))
print(result.collect())
# [('张三', '北京'), ('李四', '上海'), ('王五', '广州')]

acc = sc.accumulator(0)

def count_error(line):
    global acc
    if "error" in line.lower():
        acc += 1

sc.textFile("logs/app.log").foreach(count_error)
print(f"错误日志总数: {acc.value}")

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DoubleType

spark = SparkSession.builder.appName("DFDemo").master("local[*]").getOrCreate()

# 方式一：从内部数据创建
df = spark.createDataFrame(
    [("张三", 28, 8500.0), ("李四", 35, 12000.0), ("王五", 22, 6000.0)],
    schema=["name", "age", "salary"]
)

# 方式二：指定 Schema 创建（生产环境推荐，避免类型推断出错）
schema = StructType([
    StructField("name", StringType(), False),
    StructField("age", IntegerType(), True),
    StructField("salary", DoubleType(), True)
])
df2 = spark.createDataFrame(
    [("张三", 28, 8500.0), ("李四", 35, 12000.0)],
    schema=schema
)

# 方式三：从文件读取（最常用）
df_json = spark.read.json("data/users.json")
df_csv = spark.read.option("header", True).csv("data/users.csv")
df_parquet = spark.read.parquet("data/users.parquet")  # 性能最佳

from pyspark.sql.functions import col, avg, sum, max, count, when

result = df.filter(col("age") >= 25) \
    .groupBy() \
    .agg(
        count("*").alias("total"),
        avg("salary").alias("avg_salary"),
        max("salary").alias("max_salary")
    )

result.show()

df.createOrReplaceTempView("employees")

result = spark.sql("""
    SELECT COUNT(*) AS total, AVG(salary) AS avg_salary, MAX(salary) AS max_salary
    FROM employees
    WHERE age >= 25
""")

result.show()

from pyspark.sql.functions import col, desc, expr, lit, concat

# 选择与过滤
df.select("name", "salary").show()
df.filter((col("age") > 25) & (col("salary") < 10000)).show()

# 排序与去重
df.orderBy(desc("salary")).show()
df.dropDuplicates(["name"]).show()

# 新增列
df = df.withColumn("bonus", col("salary") * 0.1)
df = df.withColumn("level", when(col("salary") > 10000, "高级").otherwise("普通"))

# 聚合
df.groupBy("level").agg(avg("salary").alias("avg_sal")).show()

# 连接
df_a.join(df_b, df_a.id == df_b.id, "inner").show()     # 内连接
df_a.join(df_b, df_a.id == df_b.id, "left").show()      # 左连接
df_a.join(df_b, df_a.id == df_b.id, "left_anti").show() # 反连接：在 A 不在 B

# 窗口函数
from pyspark.sql.window import Window
w = Window.partitionBy("department").orderBy(desc("salary"))
df.withColumn("rank", row_number().over(w)).show()

# 读取 JSON（支持自动推断 Schema）
df = spark.read.option("multiline", True).json("data/nested.json")

# 读取 CSV（推荐显式指定 Schema 和选项）
df = spark.read \
    .option("header", True) \
    .option("inferSchema", True) \
    .option("sep", ",") \
    .csv("data/sales.csv")

# 读取 Parquet（列式存储，性能最佳）
df = spark.read.parquet("data/sales.parquet")

# 写入时按列分区
df.write \
    .mode("overwrite") \
    .partitionBy("year", "month") \
    .parquet("output/sales_partitioned")

# 写入时指定压缩
df.write \
    .mode("append") \
    .option("compression", "snappy") \
    .parquet("output/sales_compressed")

-- 创建数据库
CREATE DATABASE IF NOT EXISTS analytics
LOCATION '/warehouse/analytics';

-- 创建表
CREATE TABLE IF NOT EXISTS analytics.sales (
    id BIGINT,
    product_name STRING,
    amount DECIMAL(10, 2),
    sale_date DATE
)
PARTITIONED BY (sale_date)
STORED AS PARQUET;

-- 查看表结构（Spark 4.0 支持以 JSON 格式描述表）
DESCRIBE TABLE EXTENDED analytics.sales;

-- 基本聚合
SELECT
    product_name,
    COUNT(*) AS order_cnt,
    SUM(amount) AS total_amount,
    AVG(amount) AS avg_amount
FROM analytics.sales
WHERE sale_date >= '2026-01-01'
GROUP BY product_name
HAVING SUM(amount) > 10000
ORDER BY total_amount DESC;

-- 窗口函数
SELECT
    product_name,
    sale_date,
    amount,
    ROW_NUMBER() OVER (PARTITION BY product_name ORDER BY amount DESC) AS rn,
    SUM(amount) OVER (PARTITION BY product_name) AS product_total
FROM analytics.sales;

-- 多表连接
SELECT s.product_name, s.amount, c.category_name
FROM sales s
JOIN product_catalog c ON s.product_name = c.name
WHERE c.category_name = '电子产品';

# DataFrame 注册为视图
df.createOrReplaceTempView("temp_employees")

# 用 SQL 查询，结果仍是 DataFrame
high_salary = spark.sql("SELECT * FROM temp_employees WHERE salary > 10000")

# 继续用 API 处理
high_salary.groupBy("department").count().show()

-- 定义会话变量
DECLARE report_date DATE DEFAULT CURRENT_DATE;
DECLARE threshold DOUBLE DEFAULT 10000.0;

-- 在查询中使用
SELECT product_name, SUM(amount) AS total
FROM sales
WHERE sale_date = report_date
GROUP BY product_name
HAVING SUM(amount) > threshold;

-- 用 SQL 定义函数
CREATE FUNCTION calc_discount(price DOUBLE, rate DOUBLE)
RETURNS DOUBLE
RETURN price * (1 - rate);

-- 使用函数
SELECT product_name, calc_discount(amount, 0.15) AS discounted FROM sales;

from pyspark.sql.functions import col, from_json, window
from pyspark.sql.types import StructType, StructField, StringType, TimestampType

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

# 定义日志 Schema
log_schema = StructType([
    StructField("timestamp", TimestampType()),
    StructField("level", StringType()),
    StructField("message", StringType())
])

# 从 Kafka 读取流
stream = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "app-logs") \
    .load()

# 解析 JSON 并按窗口聚合
parsed = stream.selectExpr("CAST(value AS STRING) as json") \
    .select(from_json(col("json"), log_schema).alias("log")) \
    .select("log.*")

windowed = parsed \
    .withWatermark("timestamp", "10 minutes") \
    .groupBy(
        window(col("timestamp"), "5 minutes"),
        col("level")
    ) \
    .count()

# 写入控制台
query = windowed.writeStream \
    .outputMode("update") \
    .format("console") \
    .option("truncate", False) \
    .trigger(processingTime="30 seconds") \
    .start()

query.awaitTermination()

# 写入文件（仅支持 append 模式）
query = result.writeStream \
    .format("parquet") \
    .option("path", "output/stream") \
    .option("checkpointLocation", "checkpoint/stream") \
    .outputMode("append") \
    .start()

# 写入 Kafka
query = result.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)") \
    .writeStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("topic", "aggregated-results") \
    .option("checkpointLocation", "checkpoint/kafka") \
    .start()

# Foreach 写入外部系统（如 Redis、MySQL）
def process_row(row):
    # 自定义写入逻辑
    pass

query = result.writeStream \
    .foreachBatch(lambda df, batch_id: df.foreach(process_row)) \
    .option("checkpointLocation", "checkpoint/foreach") \
    .start()

from pyspark.sql.functions import col
from pyspark.sql.streaming.state import StateValueProcessor

class SessionAggregator(StateValueProcessor):
    def process(self, key, row, state):
        current = state.getValue() or 0
        new_value = current + row["amount"]
        state.update(new_value)
        # 设置 TTL，10 分钟后状态自动清理
        state.setTimeoutDuration("10 minutes")

result = stream.groupByKey().transformWithState(
    SessionAggregator(),
    outputMode="update",
    timeoutConf="ProcessingTimeTimeout"
)

from pyspark.ml import Pipeline
from pyspark.ml.feature import Tokenizer, StopWordsRemover, HashingTF, IDF, StringIndexer
from pyspark.ml.classification import LogisticRegression

# 准备数据
training = spark.createDataFrame([
    (0, "Spark 性能很好 推荐使用", 1.0),
    (1, "这个产品太差了 不推荐", 0.0),
    (2, "Spark 集群部署简单", 1.0),
    (3, "速度太慢 体验很差", 0.0),
], ["id", "text", "label"])

# 构建处理阶段
tokenizer = Tokenizer(inputCol="text", outputCol="words")
remover = StopWordsRemover(inputCol="words", outputCol="filtered")
hashingTF = HashingTF(inputCol="filtered", outputCol="rawFeatures", numFeatures=10000)
idf = IDF(inputCol="rawFeatures", outputCol="features")
lr = LogisticRegression(maxIter=10, regParam=0.01)

# 组装 Pipeline
pipeline = Pipeline(stages=[tokenizer, remover, hashingTF, idf, lr])

# 训练
model = pipeline.fit(training)

# 预测
test = spark.createDataFrame([
    (4, "Spark 真的很好用 速度快"),
    (5, "太慢了 不行"),
], ["id", "text"])

prediction = model.transform(test)
prediction.select("text", "prediction", "probability").show(truncate=False)

from pyspark.ml.evaluation import BinaryClassificationEvaluator
from pyspark.ml.tuning import CrossValidator, ParamGridBuilder

# 评估
evaluator = BinaryClassificationEvaluator(
    rawPredictionCol="rawPrediction",
    labelCol="label",
    metricName="areaUnderROC"
)
auc = evaluator.evaluate(prediction)
print(f"AUC: {auc:.4f}")

# 网格搜索调参
paramGrid = ParamGridBuilder() \
    .addGrid(hashingTF.numFeatures, [1000, 5000, 10000]) \
    .addGrid(lr.regParam, [0.01, 0.1, 0.5]) \
    .build()

cv = CrossValidator(
    estimator=pipeline,
    estimatorParamMaps=paramGrid,
    evaluator=evaluator,
    numFolds=5
)

cvModel = cv.fit(training)
bestModel = cvModel.bestModel

-- 创建含 VARIANT 列的表
CREATE TABLE events (
    id BIGINT,
    payload VARIANT
) USING PARQUET;

-- 插入 JSON 数据（自动转为 VARIANT）
INSERT INTO events VALUES
    (1, parse_json('{"user":"alice","action":"click","amount":99.5}'));

-- 直接查询半结构化字段，无需预定义 Schema
SELECT
    id,
    payload:user::STRING AS user_name,
    payload:action::STRING AS action,
    payload:amount::DOUBLE AS amount
FROM events;

-- 动态提取未知字段
SELECT payload:*.amount FROM events;

# 使用轻量级 Python 客户端连接远程 Spark
from pyspark.sql.connect import SparkSession

spark = SparkSession.builder \
    .remote("sc://my-spark-server:15002") \
    .getOrCreate()

# 之后的 API 用法与本地完全一致
df = spark.read.parquet("hdfs://namenode/data/sales")
df.groupBy("product").sum("amount").show()

from pyspark.sql.datasource import DataSource, DataSourceReader

class MyCustomSource(DataSource):
    @classmethod
    def reader(cls, schema):
        return MyCustomReader(schema)

class MyCustomReader(DataSourceReader):
    def __init__(self, schema):
        self.schema = schema
    def read(self, partition):
        # 自定义读取逻辑，返回数据行
        yield (1, "alice"), (2, "bob")

# 注册并使用
spark.dataSource.register(MyCustomSource, "my_source")
df = spark.read.format("my_source").load()

# 直接在分布式 DataFrame 上绘图
df = spark.read.parquet("data/sales")
plot = df.groupBy("category").sum("amount").plot.bar(x="category", y="sum(amount)")
plot.show()

# 自动广播（小于 spark.sql.autoBroadcastJoinThreshold 的表会自动广播）
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", 50 * 1024 * 1024)  # 50MB

# 手动指定广播
from pyspark.sql.functions import broadcast
big_df.join(broadcast(small_df), "id").show()

from pyspark.sql.functions import concat, lit, rand, split, col

# 给大表的倾斜 key 加盐
salted_big = big_df.withColumn("salted_id",
    concat(col("id"), lit("_"), (rand() * 10).cast("int")))

# 给小表扩展 10 倍并对应加盐
from pyspark.sql.functions import explode, array
small_exploded = small_df.withColumn("salt",
    explode(array([lit(i) for i in range(10)]))) \
    .withColumn("salted_id", concat(col("id"), lit("_"), col("salt")))

# 用加盐 key join
result = salted_big.join(small_exploded, "salted_id")

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

# 分区数直接影响并行度，通常建议每个分区 128MB 左右数据
spark.conf.set("spark.sql.shuffle.partitions", 200)  # 默认 200，按集群规模调整

# 读取时指定分区数
df = spark.read.option("maxPartitions", 100).parquet("data/large")

# 处理后重分区
df = df.repartition(200, "partition_key")  # 按 key 重分布
df = df.coalesce(10)  # 合并分区（避免 Shuffle，用于写小文件场景）

# Executor 内存与核心配置
spark = SparkSession.builder \
    .config("spark.executor.memory", "8g") \
    .config("spark.executor.cores", 4) \
    .config("spark.executor.instances", 20) \
    .getOrCreate()

# 明确指定存储级别（生产环境推荐 MEMORY_AND_DISK）
from pyspark.sql import StorageLevel
df.persist(StorageLevel.MEMORY_AND_DISK)

# 查看缓存情况
print(spark.catalog.cacheTablesExist())  # 检查表是否缓存

# 用完及时释放
df.unpersist()

# 使用列式存储 + 压缩，读写性能最优
spark.conf.set("spark.sql.parquet.compression.codec", "snappy")  # 读写快
spark.conf.set("spark.sql.parquet.compression.codec", "zstd")   # 压缩比高

# 避免小文件问题：写入前合并分区
df.coalesce(1).write.parquet("output/single_file")  # 单文件（慎用，丢失并行度）
df.repartition("key").write.parquet("output/by_key")  # 按 key 分文件

spark-submit \
  --master yarn \
  --deploy-mode cluster \
  --name SalesAnalytics \
  --executor-memory 8g \
  --executor-cores 4 \
  --num-executors 20 \
  --driver-memory 4g \
  --conf spark.sql.adaptive.enabled=true \
  --conf spark.sql.shuffle.partitions=400 \
  --packages org.apache.spark:spark-sql-kafka-0-10_2.13:4.0.0 \
  /path/to/your_app.py

# 提交到 K8s
spark-submit \
  --master k8s://https://<k8s-api-server>:6443 \
  --deploy-mode cluster \
  --name spark-on-k8s \
  --conf spark.kubernetes.container.image=apache/spark:v4.0.0 \
  --conf spark.kubernetes.namespace=spark \
  --conf spark.executor.instances=10 \
  --conf spark.kubernetes.authenticate.driver.serviceAccountName=spark \
  local:///opt/spark/examples/jars/spark-examples.jar

# SparkApplication CRD 示例
apiVersion: sparkoperator.k8s.io/v1beta2
kind: SparkApplication
metadata:
  name: sales-analytics
spec:
  type: Python
  mode: cluster
  image: apache/spark:v4.0.0
  mainApplicationFile: s3a://bucket/apps/sales_analytics.py
  driver:
    cores: 2
    memory: 4g
  executor:
    cores: 4
    instances: 20
    memory: 8g

# 配置历史服务器，查看已结束作业的 UI
export SPARK_HISTORY_OPTS="-Dspark.history.fs.logDirectory=hdfs://namenode/spark-logs"
$SPARK_HOME/sbin/start-history-server.sh

# 访问 http://<host>:18080 查看历史作业

spark-submit \
  --conf spark.log.structuredLogging.enabled=true \
  --conf spark.log.structuredLogging.format=json \
  your_app.py

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum, count, avg, to_date, current_date

spark = SparkSession.builder \
    .appName("EcommerceBatch") \
    .config("spark.sql.adaptive.enabled", "true") \
    .config("spark.sql.shuffle.partitions", 200) \
    .getOrCreate()

# 读取当日交易数据
transactions = spark.read.parquet("hdfs://namenode/data/transactions") \
    .filter(col("ts") >= to_date(current_date()))

# 用户维度报表
user_report = transactions.groupBy("user_id").agg(
    count("*").alias("order_count"),
    sum("amount").alias("total_spent"),
    avg("amount").alias("avg_order_value")
)

# 商品维度报表
product_report = transactions.groupBy("product_id", "category").agg(
    count("*").alias("sales_count"),
    sum("amount").alias("revenue")
).orderBy(col("revenue").desc())

# 写入数据仓库（分区存储）
product_report.write \
    .mode("overwrite") \
    .partitionBy("category") \
    .parquet("warehouse/product_daily_report")

from pyspark.sql.functions import window, col, from_json
from pyspark.sql.types import StructType, StructField, StringType, DoubleType, TimestampType

# Kafka 流
schema = StructType([
    StructField("user_id", StringType()),
    StructField("product_id", StringType()),
    StructField("amount", DoubleType()),
    StructField("ts", TimestampType())
])

stream = spark.readStream.format("kafka") \
    .option("kafka.bootstrap.servers", "kafka:9092") \
    .option("subscribe", "transactions") \
    .load() \
    .select(from_json(col("value").cast("string"), schema).alias("d")) \
    .select("d.*")

# 实时销售额（1 分钟窗口）
realtime_sales = stream \
    .withWatermark("ts", "5 minutes") \
    .groupBy(window(col("ts"), "1 minute")) \
    .agg(sum("amount").alias("sales"), count("*").alias("orders"))

# 写入 Redis 供大屏读取
def write_to_redis(batch_df, batch_id):
    rows = batch_df.collect()
    for row in rows:
        # 写入 Redis 逻辑
        pass

query = realtime_sales.writeStream \
    .foreachBatch(write_to_redis) \
    .outputMode("update") \
    .option("checkpointLocation", "hdfs://namenode/checkpoint/realtime_sales") \
    .trigger(processingTime="30 seconds") \
    .start()

query.awaitTermination()

from pyspark.ml import Pipeline
from pyspark.ml.feature import VectorAssembler, StandardScaler
from pyspark.ml.clustering import KMeans

# 准备特征
features = user_report.select(
    "order_count", "total_spent", "avg_order_value"
)

assembler = VectorAssembler(
    inputCols=["order_count", "total_spent", "avg_order_value"],
    outputCol="raw_features"
)
scaler = StandardScaler(
    inputCol="raw_features",
    outputCol="features",
    withMean=True,
    withStd=True
)
kmeans = KMeans(k=4, seed=42, featuresCol="features", predictionCol="cluster")

pipeline = Pipeline(stages=[assembler, scaler, kmeans])
model = pipeline.fit(features)

# 输出分群结果
clustered = model.transform(user_report)
clustered.groupBy("cluster").agg(
    count("*").alias("user_cnt"),
    avg("total_spent").alias("avg_spent")
).show()

# 保存模型，供流处理层实时打标
model.write().overwrite().save("hdfs://namenode/models/user_segmentation")