PySpark分布式架构下使用logging模块记录作业时间与时长的疑问

核心结论 #

直接在driver端用logging记录作业生命周期的话，只会输出单个代表整个作业完成的end_time；但如果把logging逻辑写到分布式执行的函数（如map、foreachPartition）里，每个worker上的任务都会打印自己的end_time。

具体说明 #

1. Driver端记录全局作业时长 #

Spark的作业控制逻辑完全在driver节点串行执行，你只需要在driver代码最开始记录start_time，等所有触发作业执行的Spark操作（比如df.collect()、df.write.save()）完成后，再记录end_time并通过logging输出即可。这种方式和Pandas里的用法一致，只会生成一条对应整个作业周期的开始/结束日志。

示例代码：

import logging
import time
from pyspark.sql import SparkSession

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

if __name__ == "__main__":
    spark = SparkSession.builder.appName("JobDurationLogging").getOrCreate()
    
    # 记录作业启动时间
    start_time = time.time()
    logger.info(f"作业启动，开始时间: {time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(start_time))}")
    
    # 执行Spark作业逻辑
    df = spark.read.csv("path/to/your/data.csv", header=True)
    df.groupBy("category").count().write.parquet("path/to/output")
    
    # 记录作业结束时间与总时长
    end_time = time.time()
    total_duration = end_time - start_time
    logger.info(f"作业完成，结束时间: {time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(end_time))}")
    logger.info(f"作业总时长: {total_duration:.2f} 秒")
    
    spark.stop()

这里所有logging操作都在driver端执行，和分布式worker无关，只会输出一次完整的作业生命周期日志。

2. Worker端记录任务级时长 #

如果把logging代码嵌入到要分发到worker执行的函数中（比如处理RDD分区的函数），那么每个worker上的每个任务都会执行这段代码，自然会打印多个end_time——对应每个单独任务的完成时间。

示例代码：

def process_partition(partition):
    import logging
    import time
    logger = logging.getLogger(__name__)
    part_start = time.time()
    # 处理分区内的数据
    for row in partition:
        # 具体处理逻辑
        pass
    part_end = time.time()
    logger.info(f"分区处理完成，耗时: {part_end - part_start:.2f} 秒")

df.rdd.foreachPartition(process_partition)

这种场景下，每个worker节点上的任务都会输出独立的日志，适合排查单个任务的性能瓶颈，但无法代表整个作业的完成时间。

注意事项 #

• 要确保worker节点的logging配置正确（比如在Spark配置中调整日志级别、设置日志滚动策略），否则可能无法看到worker端的日志。
• 统计整个作业的精确时长时，优先使用driver端的记录，它对应作业从启动到所有任务完成的完整周期。

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