Luigi任务类间执行时间传递问题：全局变量读取失效求解决方案

你遇到的问题核心是Luigi任务的进程隔离特性：默认情况下，Luigi会为每个任务启动独立进程（或由Worker分配不同进程），全局变量的作用域仅限于当前进程。因此，Pull/Validate/Upload任务中修改的全局变量，只在各自进程中生效，Finalize任务所在进程读取的是自身进程内的初始值0，自然无法获取其他任务的时间数据。

方案1：利用任务输出文件存储执行时间 #

这是Luigi原生支持的最稳妥方式——每个任务将执行时间写入自己的输出文件，Finalize通过依赖链读取这些文件内容：

修改各任务的run方法 #

import time
from datetime import datetime

CUR_DATE = datetime.now().strftime("%Y%m%d")

class Pull(luigi.Task):
    environment = luigi.Parameter(default='dev')
    
    def output(self):
        return luigi.LocalTarget(f'.out/{CUR_DATE}_{self.environment}_Pull.txt')
    
    def run(self):
        start_time = time.time()
        # --- 业务代码开始 ---
        # 此处编写Pull任务逻辑
        # --- 业务代码结束 ---
        pull_time = round(time.time() - start_time, 2)
        
        # 将时间写入输出文件
        with self.output().open('w') as f:
            f.write(str(pull_time))
        print(f"Time to pull = {pull_time}")

class Validate(luigi.Task):
    environment = luigi.Parameter(default='dev')
    
    def requires(self):
        return [Pull(environment=self.environment)]
    
    def output(self):
        return luigi.LocalTarget(f'.out/{CUR_DATE}_{self.environment}_Validate.txt')
    
    def run(self):
        start_time = time.time()
        # --- 业务代码开始 ---
        # 此处编写Validate任务逻辑
        # --- 业务代码结束 ---
        validate_time = round(time.time() - start_time, 2)
        
        with self.output().open('w') as f:
            f.write(str(validate_time))
        print(f"Time to validate = {validate_time}")

class Upload(luigi.Task):
    environment = luigi.Parameter(default='dev')
    
    def requires(self):
        return [Validate(environment=self.environment)]
    
    def output(self):
        return luigi.LocalTarget(f'.out/{CUR_DATE}_{self.environment}_Upload.txt')
    
    def run(self):
        start_time = time.time()
        # --- 业务代码开始 ---
        # 此处编写Upload任务逻辑
        # --- 业务代码结束 ---
        upload_time = round(time.time() - start_time, 2)
        
        with self.output().open('w') as f:
            f.write(str(upload_time))
        print(f"Time to upload = {upload_time}")

在Finalize中读取时间并汇总 #

class Finalize(luigi.Task):
    environment = luigi.Parameter(default='dev')
    
    def requires(self):
        return [Upload(environment=self.environment)]
    
    def output(self):
        return luigi.LocalTarget(f'.out/{CUR_DATE}_{self.environment}_Finalize.txt')
    
    def run(self):
        start_time = time.time()
        # --- 业务代码开始 ---
        # 此处编写Finalize任务逻辑
        # --- 业务代码结束 ---
        finalize_time = round(time.time() - start_time, 2)
        
        # 读取各依赖任务的执行时间
        pull_task = Pull(environment=self.environment)
        with pull_task.output().open('r') as f:
            pull_time = float(f.read().strip())
        
        validate_task = Validate(environment=self.environment)
        with validate_task.output().open('r') as f:
            validate_time = float(f.read().strip())
        
        upload_task = Upload(environment=self.environment)
        with upload_task.output().open('r') as f:
            upload_time = float(f.read().strip())
        
        # 汇总打印并写入MySQL
        print(f"all together now: pull = {pull_time}, validate = {validate_time}, upload = {upload_time}, finalize = {finalize_time}")
        # --- MySQL存储代码 ---
        # 此处编写MySQL插入逻辑，如使用pymysql/sqlalchemy等库

方案2：使用Redis等共享缓存存储时间 #

如果你的环境部署了Redis，可以用它作为跨进程的共享存储，每个任务完成后将时间写入Redis，Finalize直接从Redis读取：

初始化Redis连接 #

import redis
# 根据你的Redis配置修改参数
r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)

修改各任务的run方法 #

class Pull(luigi.Task):
    environment = luigi.Parameter(default='dev')
    
    def run(self):
        start_time = time.time()
        # --- 业务代码 ---
        pull_time = round(time.time() - start_time, 2)
        r.set(f"task_time:{self.environment}:pull", pull_time)
        print(f"Time to pull = {pull_time}")

# Validate和Upload任务同理，分别写入`task_time:{self.environment}:validate`和`task_time:{self.environment}:upload`

Finalize中读取并汇总 #

class Finalize(luigi.Task):
    environment = luigi.Parameter(default='dev')
    
    def run(self):
        start_time = time.time()
        # --- 业务代码 ---
        finalize_time = round(time.time() - start_time, 2)
        
        pull_time = float(r.get(f"task_time:{self.environment}:pull") or 0)
        validate_time = float(r.get(f"task_time:{self.environment}:validate") or 0)
        upload_time = float(r.get(f"task_time:{self.environment}:upload") or 0)
        
        print(f"all together now: pull = {pull_time}, validate = {validate_time}, upload = {upload_time}, finalize = {finalize_time}")
        # --- MySQL存储代码 ---

方案3：使用Luigi事件系统+进程安全存储 #

利用Luigi的任务成功事件，在回调中记录时间，同时使用进程安全的字典存储（适合单机器多进程场景）：

初始化进程安全存储和事件回调 #

from luigi import Event
from luigi.task import Task
from multiprocessing import Manager

# 创建进程安全的字典
task_times = Manager().dict()

def record_task_time(task, event):
    """任务成功后记录执行时间"""
    if hasattr(task, 'execution_time'):
        key = f"{task.__class__.__name__}_{task.environment}"
        task_times[key] = task.execution_time

# 注册事件监听
Task.event_handler(Event.SUCCESS)(record_task_time)

修改各任务的run方法 #

class Pull(luigi.Task):
    environment = luigi.Parameter(default='dev')
    
    def run(self):
        start_time = time.time()
        # --- 业务代码 ---
        self.execution_time = round(time.time() - start_time, 2)
        print(f"Time to pull = {self.execution_time}")

# Validate和Upload任务同理，都设置self.execution_time

Finalize中读取并汇总 #

class Finalize(luigi.Task):
    environment = luigi.Parameter(default='dev')
    
    def run(self):
        start_time = time.time()
        # --- 业务代码 ---
        finalize_time = round(time.time() - start_time, 2)
        
        pull_time = task_times.get(f"Pull_{self.environment}", 0)
        validate_time = task_times.get(f"Validate_{self.environment}", 0)
        upload_time = task_times.get(f"Upload_{self.environment}", 0)
        
        print(f"all together now: pull = {pull_time}, validate = {validate_time}, upload = {upload_time}, finalize = {finalize_time}")
        # --- MySQL存储代码 ---

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