Python多线程调用Spark SQL拉取数据报错求助

我是Python线程开发新手，首次尝试实现多线程。因为单线程拉取需要扫描70GB数据的Spark SQL查询耗时太长，想通过threading实现多线程来提升性能。写了下面的代码，但运行时出现了Py4JError和PermissionError错误，麻烦帮忙看看！

我的代码 #

from pyspark import sql, SparkConf, SparkContext,SQLContext
import threading
#conf = SparkConf().setMaster('local[*]').setAppName('appname')
#conf.set('spark.driver.allowMultipleContexts',True)
def task(sc, i):
    dataframe_mysql = sqlContext.read.format("jdbc")\
        .option("url", "jdbc:mysql://path.rds.amazonaws.com:3306/dbname")\
        .option("driver", "com.mysql.jdbc.Driver")\
        .option("dbtable", "test")\
        .option("user", "user")\
        .option("password", "pwd")\
        .load()
    dataframe_mysql.registerTempTable("data")
    data_2015_00=sqlContext.sql("select * from data where date='2015-01-15 00:00:00' LIMIT 1000")
    print(sc.parallelize(data_2015_00).show())
def run_multiple_jobs():
    conf = SparkConf().setMaster('local[*]').setAppName('appname')
    # Set scheduler to FAIR: http://spark.apache.org/docs/latest/job-scheduling.html#scheduling-within-an-application
    conf.set('spark.scheduler.mode', 'FAIR')
    conf.set('spark.driver.allowMultipleContexts', True)
    sc = SparkContext.getOrCreate()
    sqlContext=SQLContext(sc)
    for i in range(4):
        t = threading.Thread(target=task, args=(sc, i))
        t.start()
run_multiple_jobs()

报错信息 #

py4j.protocol.Py4JError: An error occurred while calling o249.getstate.
Trace:
py4j.Py4JException: Method getstate([]) does not exist
at py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:318)
at py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:326)
at py4j.Gateway.invoke(Gateway.java:272)
at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
at py4j.commands.CallCommand.execute(CallCommand.java:79)
at py4j.GatewayConnection.run(GatewayConnection.java:214)
at java.lang.Thread.run(Thread.java:748)
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "D:\Users\vk\AppData\Local\Continuum\Anaconda3\envs\py35\lib\threading.py", line 914, in _bootstrap_inner
self.run()
File "D:\Users\vk\AppData\Local\Continuum\Anaconda3\envs\py35\lib\threading.py", line 862, in run
self._target(*self._args, **self._kwargs)
File "", line 8, in task
print(sc.parallelize(data_2015_00).show())
File "D:\Users\vk\spark\spark-2.1.0-bin-hadoop2.7\python\pyspark\context.py", line 470, in parallelize
os.unlink(tempFile.name)
PermissionError: [WinError 32] The process cannot access the file because it is being used by another process: 'D:\Users\vk\AppData\Local\Temp\spark-7741cdcd-f85d-4f9c-8a41-50bf6f293dbc\pyspark-fb73a703-e797-44c7-b56a-0e08999aea3a\tmp4w72p_jj'

问题根源分析 #

首先得明确：Spark的核心对象（比如SparkContext、SQLContext）不是线程安全的，你不能直接在Python多线程环境里共享这些对象来执行任务。你遇到的两个错误都是这个原因导致的：

1. Py4JError：PySpark底层靠Py4J调用Java虚拟机中的Spark对象，而Java侧的Spark组件不支持跨线程的序列化和访问，子线程调用这些对象的方法时就会触发这个错误；
2. PermissionError：多个线程同时操作Spark的临时文件，引发了文件资源竞争，导致文件被占用无法删除，这是典型的线程安全问题。

另外你的代码还有几个小问题：

• sqlContext是主线程定义的全局变量，子线程直接使用会引发线程安全风险；
• sc.parallelize(data_2015_00)完全多余，data_2015_00本身就是DataFrame，直接调用.show()即可；
• spark.driver.allowMultipleContexts这个配置官方不推荐使用，容易引发更多不可控的问题。

正确的实现方式 #

其实Spark本身就支持多任务并行执行，根本不需要你手动用Python threading来实现。只需要开启FAIR调度模式，Spark就会自动帮你并行处理多个Job。

修改后的代码如下：

from pyspark.sql import SparkSession
from concurrent.futures import ThreadPoolExecutor

def task(spark, task_id):
    # 每个任务使用传入的SparkSession进行操作
    df_mysql = spark.read.format("jdbc")\
        .option("url", "jdbc:mysql://path.rds.amazonaws.com:3306/dbname")\
        .option("driver", "com.mysql.jdbc.Driver")\
        .option("dbtable", "test")\
        .option("user", "user")\
        .option("password", "pwd")\
        .load()
    
    # 用createOrReplaceTempView替代已废弃的registerTempTable
    df_mysql.createOrReplaceTempView("data")
    result_df = spark.sql("select * from data where date='2015-01-15 00:00:00' LIMIT 1000")
    
    # 直接调用show()，不需要转成RDD
    result_df.show()
    print(f"任务 {task_id} 执行完成")

def run_multiple_jobs():
    # 用SparkSession替代旧的SparkContext+SQLContext，这是Spark 2.x及以后的推荐方式
    spark = SparkSession.builder \
        .master('local[*]') \
        .appName('multi-job-demo') \
        .config('spark.scheduler.mode', 'FAIR') \
        .getOrCreate()
    
    # 使用ThreadPoolExecutor异步提交任务，SparkSession在Spark 2.x中是线程安全的，可以安全共享
    with ThreadPoolExecutor(max_workers=4) as executor:
        for i in range(4):
            executor.submit(task, spark, i)
    
    spark.stop()

run_multiple_jobs()

额外优化建议 #

1. 放弃手动线程操作：Spark的调度器已经做了充分的并行优化，手动用Python线程操作Spark核心对象几乎一定会踩坑；
2. JDBC读取分区优化：如果你的MySQL数据量很大，建议在读取时就按date字段做分区（比如用partitionColumn、lowerBound、upperBound、numPartitions参数），让Spark并行读取数据，效率比后续并行查询更高；
3. 避免全局变量：尽量不要在多线程环境中使用全局的Spark对象，最好通过参数传递的方式使用；
4. 优先Linux环境：Windows的文件锁机制比较严格，Spark在Windows上运行时容易出现临时文件占用的问题，生产环境尽量用Linux。

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