PySpark中如何展示applyInPandas异常的完整Python堆栈跟踪

用户提供的PySpark代码 #

import pandas as pd
from pyspark.sql.functions import pandas_udf, ceil 

df = spark.createDataFrame(
    [(1, 1.0), (1, 2.0), (2, 3.0), (2, 5.0), (2, 10.0)],
    ("id", "v"))   

def normalize(pdf):
    v = pdf.v
    return pdf.assign(v=(v - v.mean()) / v.std()) 

df.groupby("id").applyInPandas(normalize, schema="id long, v double").show()

用户的问题 #

当applyInPandas这类Python代码出现错误时，只能获取到Java代码的堆栈跟踪，无法获取Python代码的堆栈信息。是否可以通过配置Spark的log4j2配置来获取Python的堆栈跟踪？

用户当前的日志配置 #

logging.config.fileConfig("logging.conf")
logger = logging.getLogger(__name__)
log4jLogger = spark.sparkContext._jvm.org.apache.log4j
logger = log4jLogger.LogManager.getLogger("Logexample")
logger.setLevel(log4jLogger.Level.DEBUG)

解决方案 #

1. 配置log4j2捕获Python Worker日志 #

Spark的Python Worker输出的日志由org.apache.spark.api.python Logger负责处理，你需要在log4j2配置文件中添加对应配置，将级别设为DEBUG，确保捕获Python层的异常信息：

<Logger name="org.apache.spark.api.python" level="DEBUG" additivity="false">
    <AppenderRef ref="console"/> <!-- 根据需求选择输出目标 -->
    <AppenderRef ref="fileAppender"/>
</Logger>

2. 让Spark加载自定义log4j2配置 #

初始化SparkSession时，通过配置参数指定log4j2配置文件的绝对路径，确保Driver和Executor都使用该配置：

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("YourAppName") \
    .config("spark.driver.extraJavaOptions", "-Dlog4j.configurationFile=file:/your/path/log4j2.xml") \
    .config("spark.executor.extraJavaOptions", "-Dlog4j.configurationFile=file:/your/path/log4j2.xml") \
    .getOrCreate()

3. 在Python函数中手动打印堆栈 #

如果log4j2配置后仍有信息遗漏，可以在applyInPandas的处理函数中手动捕获异常并打印完整堆栈，确保信息输出到Worker日志：

import pandas as pd
import traceback

def normalize(pdf):
    try:
        v = pdf.v
        return pdf.assign(v=(v - v.mean()) / v.std())
    except Exception as e:
        traceback.print_exc()  # 打印完整Python堆栈到日志
        raise  # 重新抛出异常让Spark捕获

4. 调整Spark日志级别 #

确保Spark的Driver和Executor日志级别设置为DEBUG，可以通过Spark配置参数添加：

spark = SparkSession.builder \
    .appName("YourAppName") \
    .config("spark.driver.log.level", "DEBUG") \
    .config("spark.executor.log.level", "DEBUG") \
    .getOrCreate()

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