在 Jupyter Notebook 中实现并行评估队列

本文将探讨如何在 Jupyter Notebook 环境中实现并行任务处理，以避免长时间运行的函数阻塞 Notebook 的交互体验。我们将利用 Python 的 concurrent.futures 模块和 ipywidgets 库，创建一个可以在后台执行任务的线程池，并将任务的输出实时显示在 Notebook 中。

使用 concurrent.futures.ThreadPoolExecutor 实现并行

concurrent.futures 模块提供了一个高级接口，用于异步执行可调用对象。其中，ThreadPoolExecutor 类允许我们在线程池中执行任务，从而实现并行处理。

使用 ipywidgets.Output 显示任务输出

ipywidgets 库提供了一系列交互式控件，可以方便地在 Notebook 中显示和操作数据。Output 控件可以捕获标准输出和标准错误，并将其显示在 Notebook 中，这对于显示后台任务的执行进度和结果非常有用。

实现代码

以下代码展示了如何使用 concurrent.futures.ThreadPoolExecutor 和 ipywidgets.Output 来实现并行任务处理：

import sys
import asyncio
import concurrent.futures

import ipywidgets

threadpool = concurrent.futures.ThreadPoolExecutor(4)

def run(fn, *args, **kwds):
    "run fn in threadpool"
    out = ipywidgets.Output()

    def print(*args, file=sys.stdout):
        line = ' '.join(map(str, args)) + '\n'
        if file is sys.stderr:
            out.append_stderr(line)
        else:
            out.append_stdout(line)

    def done(fut: asyncio.Future):
        try:
            result = fut.result()
        except asyncio.CancelledError:
            print("cancelled", fut, file=sys.stderr)
        except Exception:
            print("failed", fut, file=sys.stderr)
        else:
            print("completed", fut)

    async def go():
        loop = asyncio.get_running_loop()
        return await loop.run_in_executor(
            threadpool,
            lambda: fn(print, *args, **kwds),
        )

    task = asyncio.create_task(go())
    task.add_done_callback(done)

    return out

这段代码定义了一个 run 函数，它接受一个可调用对象 fn 和一些参数，并在线程池中执行该对象。run 函数还创建了一个 ipywidgets.Output 控件，用于显示任务的输出。

在 run 函数中，我们定义了一个 print 函数，它将输出重定向到 Output 控件。我们还定义了一个 done 函数，它在任务完成后被调用，用于显示任务的完成状态。

天谱乐

唱鸭旗下AI音乐创作平台，为您提供个性化音乐创作体验！

514

查看详情

最后，我们使用 asyncio.create_task 创建一个异步任务，并在线程池中执行该任务。

示例

以下代码展示了如何使用 run 函数来执行一个长时间运行的函数：

import time

def cpu_bound(print, dt, fail=False):
    for i in range(10):
        time.sleep(dt)
        print(i, time.time())
    if fail:
        1 / 0
    return "done"

run(cpu_bound, 0.1)

这段代码定义了一个 cpu_bound 函数，它模拟一个长时间运行的任务。run 函数将 cpu_bound 函数放入线程池执行，并将输出显示在 Notebook 中。

注意事项

• 线程池的大小应该根据实际情况进行调整。如果线程池过小，可能会导致任务排队等待执行；如果线程池过大，可能会导致系统资源耗尽。
• 在执行长时间运行的任务时，应该注意任务的资源消耗，避免占用过多的 CPU 和内存资源。
• ipywidgets.Output 控件依赖于 asyncio 循环，确保在 Jupyter Notebook 中正确运行。

总结

通过使用 concurrent.futures.ThreadPoolExecutor 和 ipywidgets.Output，我们可以在 Jupyter Notebook 中实现并行任务处理，而不会阻塞 Notebook 本身。这种方法可以提高 Notebook 的交互体验，并允许我们同时执行多个任务。

以上就是在 Jupyter Notebook 中实现并行评估队列的详细内容，更多请关注php中文网其它相关文章！