从 MySQL 开发到 Spark 实战

01为什么要从 MySQL 转向 Spark

作为 MySQL 开发者，你已经熟悉了关系型数据库的建表、写 SQL、用 ORM 框架操作数据的完整流程。但当数据量从百万级增长到亿级、十亿级时，MySQL 的单机架构会遭遇性能瓶颈——慢查询、锁表、CPU/IO 打满。这就是你需要学习 Spark 的根本原因。

Spark 是什么？

Spark 是通用分布式大数据计算引擎，支持离线批处理、实时流、SQL 分析、机器学习，替代老旧 MapReduce，内存计算速度提升 10~100 倍。它本身不持久化存储任何数据——数据存在 HDFS / 对象存储 / Hive / MySQL / HBase 等外部系统中，Spark 只负责把数据读出来、在集群内存中并行计算、再把结果写回去。

02概念映射表

下面这张表是你最需要的——把已有的 MySQL 知识直接映射到 Spark 概念上，快速建立认知框架。

03环境搭建

MySQL 开发者习惯用 mysql CLI 或 Navicat 连数据库。Spark 的环境稍有不同——你需要一个 Spark 运行环境来执行代码。以下是三种方案，按推荐程度排序。

方案 1：pip 安装 PySpark（推荐新手）

最简单的方式，无需下载 Spark 安装包，一条命令搞定。

前置依赖

• Java 11（必须，Spark 底层 JVM 运行）
• Python 3.8+

# 安装 PySpark
pip install pyspark==3.5.0

# 启动交互式终端（类似 mysql CLI）
pyspark

出现 Welcome to PySpark 3.5.0 即成功。启动后自动生成两个全局变量：

• spark：SparkSession（2.0+ 统一入口，相当于你的数据库连接）
• sc：SparkContext（底层 RDD 上下文）

方案 2：完整 Spark 安装包（Linux/Mac）

# 下载解压
wget https://archive.apache.org/dist/spark/spark-3.5.0/spark-3.5.0-bin-hadoop3.tgz
tar -xzf spark-3.5.0-bin-hadoop3.tgz -C /opt/
ln -s /opt/spark-3.5.0-bin-hadoop3 /opt/spark

# 环境变量
export SPARK_HOME=/opt/spark
export PATH=$SPARK_HOME/bin:$PATH

# 验证
pyspark

方案 3：Windows（避坑要点）

方案 4：在线免安装（Jupyter/Colab）

# Jupyter Notebook 或 Google Colab 中直接执行
pip install pyspark

from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("demo").getOrCreate()

04第一个 Spark 程序

下面用同一个需求——"统计用户行为日志中各页面的 UV"——对比 MySQL 写法和 Spark 写法，让你直观感受差异。

from sqlalchemy import create_engine

engine = create_engine(
    "mysql+pymysql://user:pass@host/db"
)

# SQL 交给 MySQL 执行
sql = """
SELECT page, COUNT(DISTINCT uid) AS uv
FROM user_log
GROUP BY page
"""

result = engine.execute(sql)
for row in result:
    print(row)

from pyspark.sql import SparkSession

# 1. 创建 SparkSession（固定模板）
spark = SparkSession.builder \
    .appName("UV统计") \
    .master("local[*]") \
    .enableHiveSupport() \
    .getOrCreate()

# 2. SQL 交给 Spark 集群执行
spark.sql("""
SELECT page, COUNT(DISTINCT uid) AS uv
FROM ods.user_log
GROUP BY page
""").show()

# 3. 关闭
spark.stop()

Spark 程序固定模板（可复制到 .py 文件运行）

from pyspark.sql import SparkSession

# ===== 固定模板：所有 PySpark 代码的开头 =====
spark = SparkSession.builder \
    .appName("MyApp") \
    .master("local[*]") \
    .getOrCreate()

# ===== 你的业务逻辑 =====
df = spark.read.csv("data.csv", header=True, inferSchema=True)
df.createOrReplaceTempView("my_table")
spark.sql("SELECT * FROM my_table LIMIT 10").show()

# ===== 固定结尾 =====
spark.stop()

运行命令：

spark-submit --master local[*] my_app.py

05SQL 执行引擎的本质区别

这是从 MySQL 转向 Spark 最核心的知识点——理解 SQL 在两种环境下"谁来执行"的根本差异。

执行流程对比

MySQL 路径：全部计算压在数据库单机

Spark 路径：计算收拢到 Spark 集群，存储只提供原始数据

分维度详细对比

06DataFrame 与 SparkSQL

对 MySQL 开发者来说，DataFrame 就是一张分布式数据表——带字段名、数据类型，兼容标准 SQL，性能最优。这是生产环境的核心，必学。

Spark 三层数据抽象

6.1 手动创建 DataFrame

# 类似 MySQL 的 CREATE TABLE ... VALUES
data = [
    ("Alice", "北京", "2024-01-01", "home"),
    ("Bob", "上海", "2024-01-01", "search"),
    ("Alice", "北京", "2024-01-02", "cart"),
]

df = spark.createDataFrame(data, ["name", "city", "date", "page"])
df.show()
# +-----+----+----------+------+
# | name|city|      date|  page|
# +-----+----+----------+------+
# |Alice|北京|2024-01-01|  home|
# |  Bob|上海|2024-01-01|search|
# |Alice|北京|2024-01-02|  cart|
# +-----+----+----------+------+

6.2 DSL 链式操作（类 pandas / ORM 写法）

# 类似 SQLAlchemy ORM 的链式查询
df.filter(df.city == "北京") \
  .groupBy("page") \
  .count() \
  .show()

# +----+-----+
# |page|count|
# +----+-----+
# |home|    1|
# |cart|    1|
# +----+-----+

6.3 SparkSQL 标准 SQL 查询（最常用）

# 注册临时视图（类似给表起个别名）
df.createOrReplaceTempView("user_log")

# 直接写标准 SQL，语法和 MySQL 基本一致
result = spark.sql("""
SELECT page, COUNT(DISTINCT name) AS uv
FROM user_log
GROUP BY page
ORDER BY uv DESC
""")
result.show()

6.4 读写外部文件

# 读 CSV（类似 MySQL 的 LOAD DATA INFILE）
df = spark.read.csv("hdfs://xxx/data/user_log.csv",
                     header=True, inferSchema=True)

# 读 Parquet（列式存储，Spark 性能最优格式）
df = spark.read.parquet("hdfs://xxx/data/user_log.parquet")

# 写入结果（类似 INSERT INTO ... SELECT）
result.write.mode("overwrite") \
       .parquet("hdfs://xxx/output/uv_result")

07多数据源联邦查询

这是 MySQL 开发者最需要理解的能力差异——MySQL 只能查自己库里的表，而 Spark 可以在一条 SQL 中同时关联 Hive 表、MySQL 表、HBase 表、文件，所有计算在 Spark 集群内完成。

场景：三表关联查询

-- Hive 离线日志 + MySQL 业务维度 + HBase 用户画像 三表关联
SELECT
    a.page,
    b.user_name,
    c.user_tag,
    COUNT(DISTINCT a.uid) AS uv
FROM ods.user_log a
JOIN mysql_dim.user b ON a.uid = b.uid
JOIN hbase_user c ON a.uid = c.uid
GROUP BY a.page, b.user_name, c.user_tag

四种数据源的接入方式

① Hive 表（最常用）

# 开启 Hive 支持，自动读取 Hive Metastore 中的表
spark = SparkSession.builder \
    .appName("app") \
    .enableHiveSupport() \
    .getOrCreate()

# 直接查询，无需额外配置
spark.sql("SELECT * FROM ods.user_log LIMIT 10").show()

② MySQL 表（JDBC 外部数据源）

# 方式 1：临时读取后注册视图
mysql_df = spark.read \
    .format("jdbc") \
    .option("url", "jdbc:mysql://host:3306/db") \
    .option("dbtable", "user") \
    .option("user", "root") \
    .option("password", "xxx") \
    .load()

mysql_df.createOrReplaceTempView("mysql_dim.user")

# 方式 2：创建永久外部表（映射关系存入 Hive Metastore）
spark.sql("""
CREATE TABLE IF NOT EXISTS mysql_dim.user
USING jdbc
OPTIONS (
    url 'jdbc:mysql://host:3306/db',
    dbtable 'user',
    user 'root',
    password 'xxx'
)
""")

③ 原始文件（CSV/Parquet/JSON）

df = spark.read.csv("hdfs://xxx/data/log.csv", header=True)
df.createOrReplaceTempView("file_log")

④ HBase 表

df = spark.read \
    .format("org.apache.phoenix.spark") \
    .option("table", "user_profile") \
    .option("zkUrl", "host:2181") \
    .load()
df.createOrReplaceTempView("hbase_user")

08表名前缀的真相

MySQL 开发者看到 ods.user_log、mysql_dim.user 这样的表名，自然会问：这些前缀是 Spark 的标准约定吗？看到 ods. 就知道是 Hive？看到 mysql_dim. 就知道是 MySQL？

ods. 的来源：数仓分层标准（Hive 专属）

ods 是离线数仓分层标准中的贴源原始数据层（Operational Data Store），所有从 MySQL、日志同步过来的原始数据统一存在 Hive 的 ods 库下。这是行业通用的人工命名规范，不是 Spark 内核逻辑。

mysql_dim. 是企业自定义规则

企业为了一眼区分"Hive 维度"和"MySQL 业务维度"做的命名区分：

• dim.xxx：Hive 内部存储的静态维度表（HDFS）
• mysql_dim.xxx：映射 MySQL 库的维度表
• hbase_xxx：映射 HBase 表
• file_temp：文件注册的临时视图

09Spark 如何识别数据源

既然不靠表名前缀，那 Spark 怎么知道 ods.user_log 是 Hive 表、mysql_dim.user 是 MySQL 表？答案是：靠"这张表是在哪种元数据载体里注册的"。

先分清两种表对象

逐表识别逻辑

ods.user_log → Hive 数据源

前提：代码开启了 enableHiveSupport()。哪怕你把库改名成 aaa.bbb，只要元数据在 Hive Metastore，Spark 依旧走 HDFS 读取。

mysql_dim.user → MySQL 数据源

hbase_user → HBase 数据源

同样是临时视图模式，Spark 内存记录该视图绑定的是 HBase 连接器，查询时走 HBase ZK 读取 HFile 数据。

Spark 3.x 正规方案：Multi Catalog

企业正式环境不靠视图和命名区分，而是配置多 Catalog，语法天然隔离：

# spark-defaults.conf 配置
spark.sql.catalog.hive_catalog  org.apache.spark.sql.hive.HiveCatalog
spark.sql.catalog.mysql_catalog com.mysql.jdbc.MySQLCatalog
spark.sql.catalog.mysql_catalog.url jdbc:mysql://host:3306

# SQL 中通过 Catalog 名天然隔离数据源
SELECT *
FROM hive_catalog.ods.user_log a
JOIN mysql_catalog.dim.user b ON a.uid = b.uid

此时 . 前面第一层是 Catalog 名称，Spark 配置文件绑定了 Catalog 与对应数据源，是程序层面的官方区分方案。

10完整实战

需求：读取用户行为日志（CSV），统计各页面访问 UV，写入结果文件。这是从 MySQL 开发者视角的端到端 Spark 实战。

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

# ===== 1. 创建 SparkSession =====
spark = SparkSession.builder \
    .appName("PageUV") \
    .master("local[*]") \
    .getOrCreate()

# ===== 2. 读取数据（类似 LOAD DATA INFILE）=====
df = spark.read.csv(
    "hdfs://xxx/data/user_log.csv",
    header=True,
    inferSchema=True
)

# ===== 3. 注册临时视图 =====
df.createOrReplaceTempView("user_log")

# ===== 4. 执行 SQL（和 MySQL 写法一致）=====
result = spark.sql("""
SELECT
    page,
    COUNT(DISTINCT uid) AS uv,
    COUNT(*) AS pv
FROM user_log
WHERE date >= '2024-01-01'
GROUP BY page
ORDER BY uv DESC
""")

# ===== 5. 查看结果 =====
result.show()

# +------+---+---+
# |  page| uv| pv|
# +------+---+---+
# |  home|150|300|
# |search|120|250|
# |  cart| 80|100|
# +------+---+---+

# ===== 6. 写入结果文件（类似 INSERT INTO ... SELECT）=====
result.write.mode("overwrite") \
       .parquet("hdfs://xxx/output/uv_result")

# ===== 7. 关闭 =====
spark.stop()

11常见误区与 FAQ

Spark 与生态组件分工总结

12进阶学习路线