PySpark多线程执行SQL：成功却无数据+OOM任务异常咨询

开发者阿泽

2026-6-12

问题场景与疑问

手上有大量SQL查询语句，用Python多线程结合PySpark执行，每个线程提交对应SQL。现在遇到两个问题：

部分查询日志显示执行成功且数据集有记录，但最终目标表无数据；
Spark UI中部分阶段任务出现OOM（内存溢出），但任务最终标记为成功。
求相关排查思路和解决建议。

相关代码示例

单SQL执行逻辑

def run_single_rule(self, log):
    try:
        dataset = self.spark.sql(sql_filter)
        result_count = dataset.count()
        print(
            f"""
              Statement for building rule [{log.rule_id}] result:
              {sql_filter}
              dataset after processed contains {result_count} records
              """
        )
        write_to_uc_table(dataset, self.job_output)

    except Exception:
        logger.warning(
            f"""
                       rule_id = {log.rule_id} has failed with log:
                       {full_error_log}
                       """
        )
    return update_values

多线程执行逻辑

with ThreadPoolExecutor(max_workers=10) as executor:
    future_job_runs = {
        executor.submit(self.run_single_rule, query): query
        for query in not_processed_log
    }
    wait(future_job_runs)
    for future in as_completed(future_job_runs):
        log = future_job_runs[future]
        try:
            log_records.append(future.result())
        except Exception as exc:
            print(f"rule_id: {log.rule_id}  generated an exception: {exc}")
        else:
            print(
                f"Finist log_id: {log.rule_id} finised with result: {future.result()}"
            )

写入表逻辑

def write_to_uc_table(df, job_output: JobOutput):
    writer = df.write
    if job_output.partition_columns:
        writer = writer.partitionBy(job_output.partition_columns)
    if job_output.write_mode == WriteMode.OVERWRITE_PARTITION:
        writer = writer.option("partitionOverwriteMode", "dynamic")
        writer = writer.mode("overwrite")
    else:
        writer = writer.mode(job_output.write_mode)
    if job_output.options:
        for k, v in job_output.options.items():
            writer = writer.option(k, v)
    writer.saveAsTable(
        f"{job_output.target_catalog}.{job_output.target_schema}.{job_output.target_table}"
    )

异常案例示例

Statement for building rule [10030006] result:
SELECT ... FROM ...
dataset after processed contains 650048 records

Finist log_id: 10030006 finised with result: {'job_start_time': datetime.datetime(2025, 10, 31, 1, 7, 2, 469565, tzinfo=datetime.timezone.utc), 'log_id': '5763b7d8-b5ee-11f0-a43c-00163eb2a776', 'result_record_count': None, 'state': 'FAILED', 'job_end_time': datetime.datetime(2025, 10, 31, 1, 7, 25, 763043, tzinfo=datetime.timezone.utc)}

问题分析与解决建议

核心问题根源

PySpark与Python多线程的兼容性问题
SparkSession/Py4J网关并非线程安全，多线程共用同一个SparkSession会引发资源竞争，导致任务执行混乱：比如写操作被静默中断、SQL执行上下文被覆盖，最终出现"count有记录但写入无数据"的情况。
懒执行导致的重复计算与OOM
Spark是懒执行模型，dataset.count()会触发一次SQL计算，但后续write_to_uc_table时，若未缓存数据集，会重新执行一遍SQL。如果第二次执行时出现OOM，Spark的任务重试机制可能让整个任务标记为成功，但实际写入过程已失败；或者异常被外层try-except捕获，但日志只记录了count成功的信息，掩盖了写入失败的问题。
异常捕获不完整
当前代码仅在run_single_rule外层捕获异常，write_to_uc_table内部未做详细错误日志，导致写入失败的具体原因无法追踪；同时返回的update_values标记为FAILED，但未关联具体错误信息。

具体解决建议

替换Python多线程为Spark原生并行或多进程
放弃ThreadPoolExecutor，改用Spark的RDD/DataFrame并行执行SQL，或使用multiprocessing（多进程拥有独立Python解释器，避免Py4J线程安全问题）。如果必须保留多线程，需为每个线程创建独立的SparkSession（注意控制资源消耗）。
缓存数据集避免重复计算
在dataset.count()前添加dataset.cache()，确保count和write使用同一份缓存数据，避免重复执行SQL引发的OOM和数据不一致：
```
dataset = self.spark.sql(sql_filter).cache()
result_count = dataset.count()
```
执行完成后可调用dataset.unpersist()释放缓存。

增强异常日志与监控
在write_to_uc_table内部添加try-except块，记录写入过程的详细错误信息，比如：

def write_to_uc_table(df, job_output: JobOutput):
    try:
        writer = df.write
        # 原写入逻辑...
        writer.saveAsTable(...)
        logger.info(f"Successfully wrote to table {job_output.target_table}")
    except Exception as e:
        logger.error(f"Failed to write to table {job_output.target_table}: {str(e)}", exc_info=True)
        raise

确保写入失败的异常能被上层捕获并准确记录，避免静默失败。

调整Spark内存配置与重试策略
针对OOM问题，调大executor内存（spark.executor.memory）、executor堆外内存（spark.executor.memoryOverhead）；同时调整spark.task.maxFailures（默认4），避免多次重试掩盖OOM问题，必要时关闭重试以便快速定位内存瓶颈。
验证写入操作的原子性
若使用动态分区覆盖（partitionOverwriteMode=dynamic），需检查目标表的分区是否完整写入。可在写入后读取目标表对应分区，验证数据量与count结果是否一致；对于关键任务，可先写入临时表，验证后再rename到目标表，确保原子性。

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

PySpark多线程执行SQL：成功却无数据+OOM任务异常咨询

相关代码示例 #

单SQL执行逻辑 #

多线程执行逻辑 #

写入表逻辑 #

异常案例示例 #

问题分析与解决建议 #

核心问题根源 #

具体解决建议 #