无可用分区列时如何通过Spark JDBC并发加载MySQL表至DataFrame

我完全懂你的困扰——手里的表全是UUID格式的SESSIONID/PARTID/USERID，想靠Spark JDBC并发提速却找不到合适的partitionColumn，单线程拉取数据慢到离谱，确实让人头大。结合你的场景（已经指定了几个PARTID的查询条件），给你几个实用的解决方案：

方案一：基于PARTID的并行分片（最贴合你的现有查询） #

你当前的查询是用PARTID IN(...)过滤，那可以把每个PARTID拆成单独的JDBC读取任务，Spark会自动并行执行这些任务，相当于手动实现分片。这种方式的优势是每个分片的任务边界清晰，数据量可控。

示例代码：

# 定义要查询的PARTID列表
target_partids = [
    'c59ba81c-5e2f-4760-bf44-24432f1e76fc',
    '992f6369-bf10-4b2e-bd97-b7c99ec4d6f9',
    'd6ee09a5-1a16-4a0a-9e2a-3b9ffd9cf1d0'
]

# 定义基础JDBC配置
jdbc_options = {
    "url": "jdbc:mysql://127.0.0.1:3317/sesdb?useSSL=false",
    "driver": "com.mysql.jdbc.Driver",
    "user": "spark",
    "password": "[PASS]",
    "fetchsize": 5000  # 增大fetchsize减少网络往返
}

# 对每个PARTID生成一个DataFrame
dfs = []
for partid in target_partids:
    query = f"select * from sesdb.session where PARTID = '{partid}'"
    df = spark.read.jdbc(**jdbc_options, query=query)
    dfs.append(df)

# 合并所有分片的DataFrame
from functools import reduce
from pyspark.sql import DataFrame
df_session = reduce(DataFrame.unionAll, dfs)

方案二：动态获取PARTID并并行加载（PARTID取值较多时） #

如果你的目标PARTID不是固定的几个，而是需要查询所有PARTID的数据，可以先获取所有唯一的PARTID，再动态生成分片任务：

# 先获取所有唯一PARTID
partid_df = spark.read.jdbc(**jdbc_options, query="SELECT DISTINCT PARTID FROM sesdb.session")
partid_list = [row.PARTID for row in partid_df.collect()]

# 后续步骤和方案一一致，生成每个PARTID的查询并合并
dfs = []
for partid in partid_list:
    query = f"select * from sesdb.session where PARTID = '{partid}'"
    df = spark.read.jdbc(**jdbc_options, query=query)
    dfs.append(df)

df_session = reduce(DataFrame.unionAll, dfs)

注意：如果PARTID的数量非常多（比如上万级），这种方式会生成大量JDBC任务，可能给MySQL带来连接压力，此时需要通过spark.executor.cores和spark.executor.instances控制并发数。

方案三：退而求其次——用LOGINTIMESTAMP作为分区列 #

你提到试过用LOGINTIMESTAMP但觉得不是必需，其实全量加载时，只要时间范围覆盖所有数据，用时间列做分区是完全可行的，而且能保证分片数据相对均匀。关键是要根据数据分布调整参数：

df_session = spark.read \
    .format("jdbc") \
    .option("url", "jdbc:mysql://127.0.0.1:3317/sesdb?useSSL=false") \
    .option("driver", "com.mysql.jdbc.Driver") \
    .option("user", "spark") \
    .option("password", "[PASS]") \
    .option("query", "select * from sesdb.session where PARTID IN('c59ba81c-5e2f-4760-bf44-24432f1e76fc', '992f6369-bf10-4b2e-bd97-b7c99ec4d6f9', 'd6ee09a5-1a16-4a0a-9e2a-3b9ffd9cf1d0')") \
    .option("numPartitions", 20)  # 根据数据量调整分片数
    .option("fetchsize", 10000)  # 增大fetchsize减少网络开销
    .option("partitionColumn", "LOGINTIMESTAMP") \
    .option("lowerBound", "2018-01-01 00:00:00")  # 覆盖最早的登录时间
    .option("upperBound", "2025-12-31 23:59:59")  # 覆盖最晚的登录时间
    .load()

如果LOGINTIMESTAMP分布不均匀（比如某段时间数据特别多），可以手动拆分时间区间为多个查询，再并行合并，避免单个分片数据过大。

额外优化建议 #

• 增大fetchsize：MySQL默认fetchsize仅为10，设为5000-10000能大幅减少Spark与MySQL的网络往返次数。
• 调整Spark资源：适当增加executor核心数和内存，让Spark能同时执行更多分片任务。
• 过滤无效数据：如果有大量NULL值的行（比如USERID/USERJSON为NULL），可以在查询中提前过滤，减少数据传输量。

备注：内容来源于stack exchange，提问作者user3008410