如何检测Dask计算过程中当前正在处理的DataFrame

首先明确：你原本设想的通过全局变量current_df_converting标记当前处理DataFrame的思路完全不可行。Dask的计算是延迟触发的，任务真正被执行时会由调度器分配到多个worker/线程上并发运行，全局变量会被不同任务互相覆盖，既没法准确标记当前处理的数据源，还会引发难以排查的并发bug。

核心解决思路 #

根本不需要在运行时检测当前处理的是哪个DataFrame——Dask的map操作会把处理函数直接绑定到对应DataFrame的任务分片上，只要给两个DataFrame分别绑定自带日志标记的独立转换函数，就能从根源上实现独立错误日志的记录，全程没有跨任务的共享状态。

方案1：拆分专用转换函数（直白易维护） #

把通用的字典映射逻辑抽成底层工具函数，再分别为两个DataFrame写带独立日志逻辑的转换入口：

# 通用字典映射逻辑，无额外状态依赖
def _map_identifier(identifier, alias_dict):
    return alias_dict[str(identifier)]

# pileup_df专属转换函数，出错固定写入pileup日志
def convert_pileup(identifier):
    try:
        return _map_identifier(identifier, alias_dict)
    except KeyError as err:
        conversion_error_handler(err.args[0], 'pileup_log')
        return identifier

# lists_df专属转换函数，出错固定写入lists日志
def convert_lists(identifier):
    try:
        return _map_identifier(identifier, alias_dict)
    except KeyError as err:
        conversion_error_handler(err.args[0], 'lists_log')
        return identifier

绑定映射关系时直接对应调用即可：

pileup_df['identifier'] = pileup_df.identifier.map(convert_pileup, meta=('identifier', str))
lists_df['identifier'] = lists_df.identifier.map(convert_lists, meta=('identifier', str))

方案2：闭包生成专属函数（减少重复代码） #

如果不想写两份重复的异常处理逻辑，可以通过闭包动态生成带指定日志标记的转换函数，避免代码冗余：

def build_convert_function(log_target, alias_dict):
    def inner_convert(identifier):
        try:
            return alias_dict[str(identifier)]
        except KeyError as err:
            conversion_error_handler(err.args[0], log_target)
            return identifier
    return inner_convert

# 生成两个独立的转换函数实例，各自绑定对应的日志文件
convert_pileup = build_convert_function('pileup_log', alias_dict)
convert_lists = build_convert_function('lists_log', alias_dict)

注意事项 #

• 上述方案完全适配Dask的延迟计算逻辑，不管后续是通过to_csv还是其他操作触发计算，都不会出现日志错配的问题。每个任务从被调度执行的那一刻起，就自带了对应的日志标记，不需要读取外部全局状态
• 如果你的conversion_error_handler是直接写本地文件，要考虑多任务并发写同一日志文件的问题，可以给日志写入逻辑加线程锁，或者将错误信息收集到任务返回值中，待主进程统一聚合后再写文件，避免文件损坏

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