📚 课程目录第3章批量入湖基础

3. 批量入湖基础：使用 Apache Spark 读取关系型数据库

好，咱们进入实战环节。前面聊了 Lakehouse 的架构理念和入湖的整体策略，今天得真刀真枪地干一件事——用 Spark 把关系型数据库里的数据拉到 Lakehouse 里。

说白了，这是批量入湖最经典、也最绕不开的一步。你想想看，企业里 80% 的核心数据是不是还在 Oracle、MySQL、PostgreSQL 里躺着？不把它们弄出来，湖里就没水。

3.1 JDBC 连接：打通第一公里

Spark 读关系型数据库，底层走的是 JDBC。这玩意儿就像个万能转接头，只要数据库厂商提供了 JDBC 驱动，Spark 就能连上。

我个人习惯，先把驱动 jar 包准备好。比如连 MySQL，你得下载 mysql-connector-java。然后启动 Spark 时带上它：

spark-submit --jars mysql-connector-java-8.0.30.jar your_app.py

或者直接在代码里指定：

df = spark.read \
    .format("jdbc") \
    .option("url", "jdbc:mysql://host:3306/mydb") \
    .option("dbtable", "orders") \
    .option("user", "admin") \
    .option("password", "****") \
    .load()

嗯，这里要注意。密码硬编码在代码里是生产环境的大忌。我建议用参数化配置或者密钥管理服务，比如 AWS Secrets Manager 或 HashiCorp Vault。

小提示： 如果你的数据库在 VPC 内网，记得确保 Spark 集群能访问到。我曾经遇到过网络策略没开，折腾了半天才发现是安全组的问题。

3.2 并行读取策略：别让 Spark 变单线程

直接 .load() 能跑，但性能堪忧。为什么？因为默认情况下，Spark 只用一个 JDBC 连接去拉数据，说白了就是单线程读取。你想想看，几亿条数据，一个连接慢慢拖，黄花菜都凉了。

Spark 提供了并行读取的机制，核心是四个参数：

参数	含义	我的建议
`partitionColumn`	用于分区的列，通常是数字或时间	选主键或索引列，效果最好
`lowerBound`	分区下界	略小于最小值，避免漏数据
`upperBound`	分区上界	略大于最大值
`numPartitions`	分区数，即并行度	根据集群资源和数据库负载来定

举个例子，我要并行读取 orders 表，按 id 列分区：

df = spark.read \
    .format("jdbc") \
    .option("url", "jdbc:mysql://host:3306/mydb") \
    .option("dbtable", "orders") \
    .option("user", "admin") \
    .option("password", "****") \
    .option("partitionColumn", "id") \
    .option("lowerBound", "1") \
    .option("upperBound", "100000000") \
    .option("numPartitions", "10") \
    .load()

这样 Spark 会生成 10 个并行任务，每个任务读取一段数据。比如任务 1 读 id 1~1000 万，任务 2 读 1000 万~2000 万，以此类推。

  核心原理： Spark 内部会生成类似 SELECT * FROM orders WHERE id >= ? AND id < ? 的 SQL，每个分区一个查询，并行执行。

3.3 避坑指南：分区列选不好，全盘皆输

我曾经在一个项目里，选了 created_at 时间戳作为分区列。结果呢？数据分布极不均匀——最近一个月的数据占了 80%，其他月份稀稀拉拉。导致 8 个任务跑完了，2 个任务还在慢慢爬。

选分区列有几个原则：

数据分布均匀：主键、自增 ID 是最好的选择
有索引：数据库能快速执行范围查询
非空：空值会导致分区边界混乱

如果表没有合适的数字列，可以用 WHERE 条件配合多个查询来模拟并行：

# 手动构造多个查询，并行读取
conditions = ["year=2020", "year=2021", "year=2022"]
df_list = [spark.read.jdbc(url, "sales", predicates=[cond], properties=props) 
           for cond in conditions]
df = df_list[0].unionAll(df_list[1]).unionAll(df_list[2])

这招虽然笨，但有时候很管用。

3.4 连接池与 fetchsize：别把数据库压垮

并行度设高了，数据库那边可能扛不住。每个分区都是一个 JDBC 连接，10 个分区就是 10 个连接。如果数据库连接池上限是 20，你这边一跑，其他应用可能就连不上了。

我建议：

numPartitions 不要超过数据库连接池的 1/3
设置 fetchsize：控制每次网络传输的行数，默认是 1000，可以调大到 10000

.option("fetchsize", "10000")

还有一个容易被忽略的参数 sessionInitStatement，可以在建立连接后执行初始化 SQL。比如设置数据库的时区或隔离级别：

.option("sessionInitStatement", "SET time_zone='+00:00'")

3.5 实战：从 MySQL 批量入湖到 Delta Lake

好了，理论说完了，咱们来个完整的例子。把 MySQL 的 orders 表读到 Spark，然后写入 Delta Lake。

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("mysql_to_delta") \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
    .getOrCreate()

# 读取 MySQL
jdbc_df = spark.read \
    .format("jdbc") \
    .option("url", "jdbc:mysql://host:3306/mydb") \
    .option("dbtable", "orders") \
    .option("user", "admin") \
    .option("password", "****") \
    .option("partitionColumn", "id") \
    .option("lowerBound", "1") \
    .option("upperBound", "50000000") \
    .option("numPartitions", "8") \
    .option("fetchsize", "10000") \
    .load()

# 写入 Delta Lake
jdbc_df.write \
    .format("delta") \
    .mode("overwrite") \
    .save("/lakehouse/raw/orders")

这段代码跑完，MySQL 的数据就进了 Lakehouse。下次再跑，可以把 mode 改成 append，实现增量入湖。

注意： 首次全量入湖时，建议在业务低峰期执行。我曾经在白天跑全量，结果数据库 CPU 飙到 90%，被 DBA 骂了一顿。

3.6 知识体系：批量 JDBC 入湖全景

下面这张图，是我梳理的批量 JDBC 入湖的核心逻辑。你可以把它当作操作手册。

3.7 总结一下

批量 JDBC 入湖，说白了就三件事：连得上、读得快、写得稳。连得上靠驱动和网络配置，读得快靠并行策略，写得稳靠参数调优和时机选择。

我个人经验是，第一次跑全量时，先用小数据集测试并行度，观察数据库负载和 Spark 任务耗时。调优到最佳后再跑全量，能省不少时间。

嗯，这一章就到这儿。记住，入湖只是第一步，数据质量、增量同步、Schema 演化这些坑，咱们后面慢慢填。