管理Dask LocalCluster工作器控制台输出

import subprocess
import time
from dask.distributed import Client, Scheduler

# 启动调度器
scheduler = Scheduler(port=0, dashboard_address=':0') # 随机端口
scheduler.start()
scheduler_address = scheduler.address

print(f"调度器地址: {scheduler_address}")

# 启动工作器并重定向其输出到/dev/null
# 注意：这只是一个概念性示例，实际使用中需要确保dask命令在PATH中
# 并且可能需要更复杂的错误处理和进程管理
worker_cmd = [
    'dask-worker',
    scheduler_address,
    '--nthreads', '1',
    '--nprocs', '1',
    '>', '/dev/null', # 重定向stdout
    '2>', '/dev/null' # 重定向stderr
]

# 在Windows上，重定向需要shell=True，但在Linux/macOS上通常不推荐
# 并且命令行重定向通常由shell解释，直接在subprocess参数列表中可能无效
# 更可靠的方法是使用stdout/stderr参数
with open('worker_output.log', 'w') as log_file:
    worker_process = subprocess.Popen(
        ['dask-worker', scheduler_address, '--nthreads', '1', '--nprocs', '1'],
        stdout=log_file, # 重定向stdout到文件
        stderr=subprocess.STDOUT # 将stderr也重定向到stdout文件
    )

time.sleep(5) # 等待工作器启动并连接

client = Client(scheduler_address)
print(f"Dask Dashboard: {client.dashboard_link}")

# ... 运行Dask任务 ...

client.close()
worker_process.terminate() # 终止工作器进程
scheduler.stop()

import sys
import os
from distributed.diagnostics.plugin import WorkerPlugin
from dask.distributed import Client, LocalCluster
import dask

# 1. 定义一个自定义的WorkerPlugin
class RedirectStdoutPlugin(WorkerPlugin):
    """
    一个Dask工作器插件，用于重定向工作器的标准输出和标准错误。
    """
    def __init__(self, target_file=os.devnull):
        """
        初始化插件。
        Args:
            target_file (str): 重定向输出的目标文件路径。
                                默认为os.devnull，即丢弃所有输出。
                                也可以是具体的日志文件路径。
        """
        self.target_file = target_file
        self._original_stdout = None
        self._original_stderr = None
        self._redirected_file = None

    def setup(self, worker):
        """
        在工作器启动时执行，用于重定向stdout和stderr。
        """
        # 保存原始的stdout和stderr
        self._original_stdout = sys.stdout
        self._original_stderr = sys.stderr

        # 打开目标文件用于写入
        # 'w' 模式会清空文件，'a' 模式会追加
        self._redirected_file = open(self.target_file, 'w')

        # 重定向sys.stdout和sys.stderr
        sys.stdout = self._redirected_file
        sys.stderr = self._redirected_file

        # 可以在这里打印一条信息到工作器自己的日志（不会被重定向）
        worker.logger.info(f"工作器 '{worker.name}' 的stdout/stderr已重定向到 '{self.target_file}'")

    def teardown(self, worker):
        """
        在工作器关闭时执行，用于恢复stdout和stderr。
        """
        # 恢复原始的stdout和stderr
        if self._original_stdout:
            sys.stdout = self._original_stdout
        if self._original_stderr:
            sys.stderr = self._original_stderr

        # 关闭重定向的文件
        if self._redirected_file:
            self._redirected_file.close()

        worker.logger.info(f"工作器 '{worker.name}' 的stdout/stderr已恢复。")

# 示例函数，会在工作器内部产生print输出
def dask_function(i):
    print(f'工作器正在处理任务 {i}，这条信息将被重定向或丢弃！')
    return i**2

if __name__ == "__main__":
    # 2. 启动LocalCluster
    # n_workers=2, processes=True 表示启动两个独立的进程作为工作器
    cluster = LocalCluster(n_workers=2, processes=True, threads_per_worker=1)
    client = Client(cluster)
    print(f"Dask Dashboard: {client.dashboard_link}")

    # 3. 注册插件
    # 将工作器输出重定向到/dev/null（Linux/macOS）或 'nul'（Windows）
    # 如果想将输出保存到文件，可以将 'os.devnull' 替换为 'worker_output.log'
    redirect_plugin = RedirectStdoutPlugin(target_file=os.devnull if sys.platform != "win32" else "nul")
    client.register_worker_plugin(redirect_plugin)

    print("\n--- 运行Dask任务（工作器输出应被重定向）---")
    dask_delays = []
    for i in range(10):
        dask_delays.append(dask.delayed(dask_function)(i))

    # 执行计算
    dask_outs = client.compute(dask_delays).result()
    print(f"计算结果: {dask_outs}")

    # 4. 关闭客户端和集群
    client.close()
    cluster.close()
    print("\n--- Dask集群已关闭 ---")

    # 验证插件是否恢复了stdout，这条信息应该正常打印到控制台
    print("这条信息应该正常打印到控制台，表明主进程的stdout未受影响。")

    # 如果重定向到文件，可以在这里读取文件内容
    # with open('worker_output.log', 'r') as f:
    #     print("\n--- 工作器日志内容 ---")
    #     print(f.read())