PySpark利用百万行TXT文件批量查询MySQL的实现方案咨询

错误根因 #

你触发TypeError: can't pickle thread lock objects错误的核心原因是：在executor端运行的RDD map算子中直接引用了driver端创建的DataFrame对象。DataFrame包含大量不可序列化的会话资源、线程锁对象，无法被序列化分发到executor节点执行，因此会触发序列化异常。
同时你原来逐行过滤的逻辑性能极低，百万级TXT行的场景下完全不可用。

最优实现方案 #

直接将TXT解析为结构化DataFrame，和MySQL读取的业务表做分布式关联查询，全流程走Spark的优化引擎，性能比逐行处理高几个量级，核心逻辑如下：

1. 把TXT文件解析为包含id、time、word_list三个字段的结构化DataFrame，其中word_list为数组类型
2. 用explode函数将word_list数组拆分为单行单个word的格式，得到TXT侧的查询条件表
3. 将MySQL业务表与查询条件表按time和word两个字段做内连接，得到的结果就是完全匹配所有查询条件的数据集，可直接用于后续运算

完整代码示例 #

from pyspark.sql import SparkSession
from pyspark.sql.functions import explode

# 初始化SparkSession
spark = SparkSession.builder.appName('test').master('local[*]').getOrCreate()
sc = spark.sparkContext

# 1. 解析TXT为结构化DataFrame
txt_df = sc.textFile("<my_huge_text_file>") \
    .map(lambda x: get_rows_in_correct_format(x)) \
    .toDF(["id", "time", "word_list"])

# 2. 拆分word数组为单行单word的查询条件表
txt_condition_df = txt_df.select("time", explode("word_list").alias("word"))

# 3. 读取MySQL业务表
mysql_df = spark.read.format("jdbc") \
    .option("url","jdbc:mysql://localhost/<my_database>") \
    .option("driver","com.mysql.jdbc.Driver") \
    .option("dbtable","<my_table>") \
    .option("user","<my_username>") \
    .option("password","<my_password>") \
    .load()

# 4. 关联查询得到匹配结果
result_df = mysql_df.join(
    txt_condition_df,
    (mysql_df.time == txt_condition_df.time) & (mysql_df.word == txt_condition_df.word),
    "inner"
)

# 后续直接对result_df执行你的处理逻辑即可

可选优化 #

如果MySQL业务表数据量远大于TXT侧的条件数据，可以把JDBC读取的dbtable参数替换为子查询，提前过滤不需要的字段、裁剪分区，减少从MySQL拉取的数据量：

.option("dbtable","(SELECT time, word, 需要的其他字段 FROM <my_table>) as t")

内容的提问来源于stack exchange，提问作者gibbone