Dask中文教程

8.1 Dask调度器概述

Dask提供了多种调度器来适应不同的计算需求。每种调度器都有其特定的使用场景和优势。

# 示例：使用不同的调度器
import dask
from dask import delayed

@delayed
def process_data(data):
    return data * 2

# 创建任务
tasks = [process_data(i) for i in range(10)]

# 使用不同的调度器
result1 = dask.compute(*tasks, scheduler='synchronous')  # 同步调度器
result2 = dask.compute(*tasks, scheduler='threads')     # 线程调度器
result3 = dask.compute(*tasks, scheduler='processes')   # 进程调度器
                        

8.2 线程调度器

线程调度器是Dask的默认调度器，适用于I/O密集型任务。

工作原理

线程调度器使用Python的线程池来并行执行任务。由于Python的GIL（全局解释器锁），线程调度器适用于以下场景：

• 文件读写操作
• 网络请求
• 数据库操作
• 其他I/O密集型任务

import dask
from dask import delayed
import requests
import time

@delayed
def fetch_url(url):
    response = requests.get(url)
    return len(response.content)

# 创建网络请求任务
urls = ['http://httpbin.org/delay/1'] * 5
tasks = [fetch_url(url) for url in urls]

# 使用线程调度器（默认）
start = time.time()
results = dask.compute(*tasks, scheduler='threads')
print(f"线程调度器耗时: {time.time() - start:.2f}秒")
                    

8.3 进程调度器

进程调度器适用于CPU密集型任务，可以绕过Python的GIL限制。

工作原理

进程调度器使用Python的多进程模块来并行执行任务。每个进程都有独立的Python解释器，因此可以充分利用多核CPU。

import dask
from dask import delayed
import time

@delayed
def cpu_intensive_task(n):
    # CPU密集型计算
    result = 0
    for i in range(n):
        result += i ** 2
    return result

# 创建CPU密集型任务
tasks = [cpu_intensive_task(1000000) for _ in range(4)]

# 使用进程调度器
start = time.time()
results = dask.compute(*tasks, scheduler='processes')
print(f"进程调度器耗时: {time.time() - start:.2f}秒")

# 对比线程调度器
start = time.time()
results = dask.compute(*tasks, scheduler='threads')
print(f"线程调度器耗时: {time.time() - start:.2f}秒")
                    

8.4 分布式调度器

分布式调度器允许在多个机器上分布式执行任务，适用于大规模计算。

设置分布式集群

# 启动分布式调度器
# 在终端中运行: dask scheduler

# 启动工作节点
# 在终端中运行: dask worker localhost:8786

# 在Python代码中连接到集群
from dask.distributed import Client

# 连接到本地集群
client = Client('localhost:8786')

# 或者创建本地集群
client = Client(processes=True, n_workers=4, threads_per_worker=2)

# 查看集群信息
print(client)
                    

使用分布式调度器

from dask.distributed import Client
from dask import delayed

# 创建客户端
client = Client(processes=True, n_workers=4, threads_per_worker=2)

@delayed
def process_data(data):
    return sum(x ** 2 for x in data)

# 创建任务
data_chunks = [range(i*1000, (i+1)*1000) for i in range(10)]
tasks = [process_data(chunk) for chunk in data_chunks]

# 使用分布式调度器
futures = client.compute(tasks)
results = client.gather(futures)

print(results)
                    

8.5 调度器配置

可以通过多种方式配置调度器的行为：

环境变量配置

# 设置默认调度器
export DASK_SCHEDULER='threads'

# 设置线程数
export DASK_NUM_WORKERS=4

# 设置进程数
export DASK_NUM_PROCESSES=2
                    

Python代码配置

import dask

# 配置默认调度器
dask.config.set(scheduler='processes')

# 配置线程数
dask.config.set(num_workers=8)

# 配置临时目录
dask.config.set(temporary_directory='/tmp/dask')
                    

配置文件

# ~/.config/dask/distributed.yaml
distributed:
  scheduler:
    work-stealing: True
    allowed-failures: 3
  worker:
    memory:
      target: 0.6
      spill: 0.7
      pause: 0.8
      terminate: 0.95
    use-file-locking: False
                    

8.6 任务调度优化

优化任务调度可以显著提高计算性能：

任务融合

import dask.array as da

# 创建数组
x = da.random.random((10000, 10000), chunks=(1000, 1000))

# 定义计算链
y = x + 1
z = y * 2
result = z.sum()

# Dask会自动融合任务以减少调度开销
# 可以通过可视化查看优化效果
result.visualize(filename='optimized_graph.png', optimize_graph=True)
                    

工作窃取

from dask.distributed import Client

# 启用工作窃取
client = Client(processes=True, n_workers=4, threads_per_worker=2)

# 工作窃取允许空闲的工作者从忙碌的工作者那里获取任务
# 这可以提高负载均衡和整体性能
                    

8.7 调度器监控

Dask提供了多种方式来监控调度器的性能：

进度条

from dask.diagnostics import ProgressBar
import dask.array as da

# 创建大型计算
x = da.random.random((10000, 10000), chunks=(1000, 1000))
y = x.sum(axis=0)

# 使用进度条监控计算进度
with ProgressBar():
    result = y.compute()
                    

性能分析

from dask.diagnostics import Profiler, ResourceProfiler, CacheProfiler
import dask.array as da

# 创建计算任务
x = da.random.random((5000, 5000), chunks=(500, 500))
y = x.sum()

# 性能分析
with Profiler() as prof, ResourceProfiler() as rprof, CacheProfiler() as cprof:
    result = y.compute()

# 查看分析结果
print(prof.results)
print(rprof.results)
print(cprof.results)
                    

可视化监控

from dask.distributed import Client
import dask.array as da

# 创建客户端
client = Client(processes=True, n_workers=2, threads_per_worker=2)

# 创建计算任务
x = da.random.random((5000, 5000), chunks=(500, 500))
y = x.sum()

# 提交任务
future = client.compute(y)

# 在浏览器中查看仪表板: http://localhost:8787
# 或者在Jupyter中使用
# client.dashboard_link
                    

8.8 故障恢复

Dask具有内置的故障恢复机制：

from dask.distributed import Client
import random

@delayed
def unreliable_task(x):
    # 模拟可能失败的任务
    if random.random() < 0.1:  # 10%的概率失败
        raise Exception("随机失败")
    return x * 2

# 创建任务
tasks = [unreliable_task(i) for i in range(20)]

# Dask会自动重试失败的任务
# 可以配置重试次数
client = Client()
results = client.compute(tasks, retries=3)
                    

8.9 调度器选择指南

根据不同的使用场景选择合适的调度器：

8.10 最佳实践

使用Dask调度器的最佳实践：

# I/O密集型任务可以设置更多线程
dask.config.set(num_workers=8)

# CPU密集型任务设置为CPU核心数
import multiprocessing
dask.config.set(num_workers=multiprocessing.cpu_count())
                            

from dask.diagnostics import ResourceProfiler
import matplotlib.pyplot as plt

with ResourceProfiler() as rprof:
    # 执行计算任务
    result = large_computation.compute()

# 绘制资源使用图表
rprof.visualize()
                            

# 不好的做法：任务过小
small_tasks = [delayed(process)(item) for item in range(100000)]

# 好的做法：合并任务
def process_batch(batch):
    return [process(item) for item in batch]

batches = [range(i, i+1000) for i in range(0, 100000, 1000)]
batch_tasks = [delayed(process_batch)(batch) for batch in batches]