使用RQ(Redis Queue)传递DataFrame时报cannot pickle '_thread.lock'错误如何解决

问题根因 #

q.enqueue()返回的是RQ的Job实例，而非任务执行的结果。你直接将Job实例作为参数传入第二个任务时，RQ需要序列化所有任务参数，而Job实例内部包含不可被pickle序列化的_thread.lock对象，因此触发该报错。

解决方案 #

方案1：使用RQ原生任务依赖功能（推荐） #

RQ内置了任务依赖逻辑，无需手动传递前序任务的对象，指定依赖关系后，后序任务会等前序任务执行完成再启动，你可以在后序任务内直接获取前序任务的返回结果：

1. 调整入队逻辑：

# 入队第一个任务，拿到Job实例
data_full_job = q.enqueue(furnish_stops)
# 入队第二个任务，指定依赖为第一个任务，不需要手动传结果参数
daily_count_job = q.enqueue(furnish_daily, depends_on=data_full_job)

2. 在furnish_daily函数内获取前序任务的DataFrame结果：

from rq import get_current_job

def furnish_daily():
    # 获取当前任务实例
    current_job = get_current_job()
    # 取前序依赖任务的返回结果，也就是你生成的DataFrame
    data_full = current_job.dependency.result
    # 后续原有逻辑直接使用data_full即可

方案2：中间存储中转大数据量DataFrame #

如果你的DataFrame体积特别大，不适合存在RQ的任务结果中，可以先将DataFrame序列化后存入Redis/本地存储/对象存储，仅传递存储标识给第二个任务：

# furnish_stops返回结果示例，序列化DataFrame存Redis
import pickle
def furnish_stops():
    # 原有BigQuery拉取逻辑生成df
    key = "bigquery_data:xxx" # 可以用任务ID作为唯一key
    conn.set(key, pickle.dumps(df))
    conn.expire(key, 3600) # 设置过期时间避免资源占用
    return key

# 入队逻辑
data_key_job = q.enqueue(furnish_stops)
daily_count_job = q.enqueue(furnish_daily, depends_on=data_key_job)

# furnish_daily逻辑
def furnish_daily():
    current_job = get_current_job()
    data_key = current_job.dependency.result
    data_full = pickle.loads(conn.get(data_key))
    # 后续处理逻辑

注意事项 #

furnish_stops的返回值只能包含可被pickle序列化的对象，不要返回BigQuery客户端、连接句柄、线程对象这类不可序列化的内容，仅返回纯DataFrame即可。

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