从动态 sql 列表运行 python 多进程答案

【问题标题】：Running python multiprocesses from a dynamic sql list从动态 sql 列表运行 python 多进程
【发布时间】：2021-06-16 15:38:45
【问题描述】：

嗨，我试图让我的代码更加动态和智能，为此我想通过动态列表调用我想要运行的函数，而不是让它们硬编码。这将清理代码并帮助自动重新运行失败的脚本。

下面是我一直在处理的代码的 sn-p。运行时我通常会收到此错误

TypeError: 'str' 对象不可调用

 cursor = conn.cursor()
    cursor.execute(
        f'''select distinct caller from {db}.log a where a.log_text like 'Failed:%' and a.log_time > DATE_TRUNC('DAY', NOW()) and caller not in (select caller from {db}.log a where a.log_text like 'Done' and a.log_time > DATE_TRUNC('DAY', NOW()))''')
    df = as_pandas(cursor)
    print('The following scripts will be rerun')
    print(df)

    c = df['caller']

    processes = []
    # Loop over failed scripts/modules
    for mod in (c):  
        print(f'Rerun of {c}')
        p = multiprocessing.Process(target=mod, args=(db,))
        time.sleep(10)
        p.start()
        processes.append(p)

    for process in processes:
        process.join()

完整的回溯错误

Traceback（最近一次调用最后一次）： _bootstrap 中的文件“/home/xxx/anaconda3/lib/python3.6/multiprocessing/process.py”，第 258 行自我运行（）文件“/home/xxx/anaconda3/lib/python3.6/multiprocessing/process.py”，第 93 行，运行中 self._target(*self._args, **self._kwargs) TypeError：“str”对象不可调用流程流程-2：回溯（最近一次通话最后）： _bootstrap 中的文件“/home/xxx/anaconda3/lib/python3.6/multiprocessing/process.py”，第 258 行自我运行（）文件“/home/xxx/anaconda3/lib/python3.6/multiprocessing/process.py”，第 93 行，运行中 self._target(*self._args, **self._kwargs) TypeError：“str”对象不可调用进程以退出代码 0 结束

【问题讨论】：

新子进程的target 必须是函数对象，但您只提供函数的名称。
我可以把名字转换成同名的函数对象吗？
最好找到名称所指的函数或创建一个字典来将函数名称映射到对象。子进程应该调用的函数在哪里，与所示代码相同的模块，其他模块，分散在多个模块中？
它们分散在许多不同的脚本/模块中。通常我只是将它们命名为“ for mod in (foo_1,foo_2......):” 但是对于这段代码，每次运行我想要调用的脚本或函数都会有所不同，这取决于在运行期间可能失败的脚本或函数第一次运行。
那么我认为以函数名作为键、函数对象作为值的字典是解决这个问题的最佳方法。

标签： python sql loops impala multiprocess

【解决方案1】：

所以我找到了解决问题的方法，并将其发布在这里，以防将来有人可以使用它。

基本上我使用 getattr 和 importlib.import_module 函数。然后我通过 for 循环启动它们。

所以我使用 sql 来获取所有失败模块的名称，然后将这些模块加载到 df 列表中，然后通过 for 循环进行迭代，该循环将启动失败的模块。

cursor.execute(
            f'''select distinct caller from {db}.sch_log_python a where a.log_text like 'Failed:%' and lower(caller) in ('st_%','ctrl_%','ar%') and a.log_time > DATE_TRUNC('DAY', NOW()) and caller not in (select caller from {db}.sch_log_python a where a.log_text like 'Done' and a.log_time > DATE_TRUNC('DAY', NOW()))''')
        df = as_pandas(cursor)
        print('The following scripts will be rerun')
        print(df)

        sch_log_func(caller, 'Rerun of failed scripts', db)

        c = df
        # Loop over failed scripts/modules
        for i in (c):
            cstr = c.to_string(index=False, header=False)
            print(f'Initialise ' + cstr)
            cstr = cstr.lower()
            print(cstr + ' Module to import')
            print('Trying to run ' + cstr)
            cls = getattr(importlib.import_module(cstr), cstr)
            cls(db)
            print(f'Started {cstr}')

【讨论】：