python中multiprocessing.pool和pandas如何结合使用？

Question

我有一个DataFrame表"all_urls_df"，有十亿条数据，一共两列（title列，url列）。对url列进行检测，在DF表中新增status列，如果检测返回值是200，status列值为1，否则为0。

我想采用多进程的方式来实现该功能，部分代码如下：

def pandas_data_washed(df):
pool = multiprocessing.Pool(processes=35)
all_urls_df = df
all_urls_df["status"] = all_urls_df.apply(lambda x: 1 if pool.apply_async(check_url_ok, (x.url,)) else 0, axis=1)
pool.close()
pool.join()


def check_url_ok(url):
    """检测连接是否可用"""
    useragent = "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/86.0.4240.75 Safari/537.36"
    # print("正在检查URL %s" % url)

    try:

        result = requests.get(url, headers={"User-Agent": useragent}, timeout=(5, 5))
        if result.status_code == 200:

            print("访问正常 %s" % url)
            return True
        else:
            print("访问超时 %s" % url)
            return False
    except requests.exceptions.ConnectionError as e:
        # print("URL %s 访问超时" % url)
        # print("URL访问超时")
        print("访问超时 %s" % url)
        return False

现在遇到的问题如下，如果我用pool.apply，"all_urls_df"的status列出来的结果是正确的，但是数据是逐条检测，无法实现进程池并行效果；如果我用pool.apply_async，可以实现并行检测，但是status列的结果却全部都是1。请问我的代码是哪里写错了呢，该如何调整呢？

Answer 1

apply_async 返回的是对象，不能直接if pool.apply_async...