Databricks中如何在Spark RDD映射函数中打印并高效查看调试信息？

针对你遇到的问题——想查看parallelized_func中每行的唯一标识符但被Spark系统日志淹没，这里有几个实用的解决方案，按推荐程度排序：

1. 将唯一标识符与结果一起返回，在Driver端查看 #

这是最直接的调试方式，不用依赖执行器日志，所有信息都集中在Driver控制台：

修改你的函数，把唯一标识符作为返回结果的一部分：

def parallelized_func(a,b,c):
    # 替换成你实际的唯一标识符字段，比如a中的某个列
    unique_id = a["your_unique_id_col"]  # 或a.id，根据数据结构调整
    # 原函数逻辑
    ####FUNCTION BODY#####
    return {'df1': df1, 'df2': df2, 'unique_id': unique_id}

然后在collect()之后，直接打印或筛选这些ID：

data_3 = data_2.rdd.map(lambda x: parallelized_func(x, **args)).collect()

# 打印所有处理过的唯一ID
for result in data_3:
    print(f"Processed row ID: {result['unique_id']}")

# 如果要定位有问题的行，可以结合错误处理，比如检查df1/df2是否异常
problem_rows = [r for r in data_3 if r['df1'] is None or r['df2'] is None]
print("Problematic rows IDs:", [r['unique_id'] for r in problem_rows])

这种方法完全避开了执行器的日志噪音，所有调试信息都在Driver端可控。

2. 降低Spark日志级别，屏蔽无关日志 #

如果一定要在执行器日志里看打印信息，可以通过设置Spark日志级别，过滤掉大量INFO/DEBUG级别的系统日志：

在笔记本开头添加这段代码：

from pyspark import SparkContext
sc.setLogLevel("ERROR")

这样Spark只会输出ERROR及以上级别的日志，大部分任务执行的冗余日志会被屏蔽，你的print信息会更容易在执行器的stdout里找到。如果觉得还是有多余日志，可以尝试设置为"FATAL"，但注意这会屏蔽大部分错误提示，仅保留致命错误，调试时谨慎使用。

3. 使用logging模块定向日志到文件 #

用Python的logging替代print，将日志输出到DBFS的文件中，避免和Spark系统日志混在一起：

import logging

# 在Driver端配置日志（执行器会自动继承该配置）
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(row_id)s - %(message)s',
    handlers=[
        # 输出到DBFS文件，方便后续查看
        logging.FileHandler('/dbfs/tmp/row_processing.log'),
        # 同时输出到控制台
        logging.StreamHandler()
    ]
)

def parallelized_func(a,b,c):
    unique_id = a["your_unique_id_col"]
    # 用logging.info替代print，带上唯一ID
    logging.info("Processing row", extra={'row_id': unique_id})
    ####FUNCTION BODY#####
    return {'df1': df1, 'df2': df2}

运行完任务后，你可以直接在Databricks中查看日志文件：

display(dbutils.fs.head("/tmp/row_processing.log"))

所有你的自定义日志都会单独存在这个文件里，不会被Spark的系统日志干扰。

4. 用累加器收集问题行ID（针对错误排查） #

如果是想定位处理出错的行，可以用Spark的累加器来收集这些行的唯一ID，不用翻日志就能直接获取：

# 创建一个累加器，用于存储有问题的行ID
problem_row_ids = sc.accumulator([])

def parallelized_func(a,b,c):
    unique_id = a["your_unique_id_col"]
    try:
        ####FUNCTION BODY#####
        return {'df1': df1, 'df2': df2}
    except Exception as e:
        # 捕获错误，将ID加入累加器
        problem_row_ids.add(unique_id)
        # 可选：记录错误详情
        logging.error(f"Failed to process row {unique_id}: {str(e)}")
        # 返回默认值或抛出异常，根据你的需求调整
        return {'df1': None, 'df2': None}

data_3 = data_2.rdd.map(lambda x: parallelized_func(x, **args)).collect()

# 直接查看所有出错的行ID
print("Rows with errors:", problem_row_ids.value)

这种方法能快速定位到故障行，非常适合调试数据异常的场景。

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