最佳实践
内容
最佳实践¶
开始使用 Dask delayed 很容易，但要*很好地*使用它确实需要一些经验。本页包含最佳实践的建议，并包括常见问题的解决方案。
在函数上调用延迟，而不是结果¶
Dask delayed 操作于函数，如 dask.delayed(f)(x, y)，而不是操作于它们的结果，如 dask.delayed(f(x, y))。当你执行后者时，Python 首先计算 f(x, y)，然后 Dask 才有机会介入。
不要 |
做 |
# This executes immediately
dask.delayed(f(x, y))
|
# This makes a delayed function, acting lazily
dask.delayed(f)(x, y)
|
一次性进行大量计算¶
为了提高并行性，您希望在每次计算调用中包含大量的计算。理想情况下，您希望进行多次 dask.delayed 调用来定义您的计算，然后在最后调用 dask.compute。在中途调用 dask.compute 也是可以的，但 Dask 会在继续执行您的代码之前计算这些结果，从而导致计算停止。
不要 |
做 |
# Avoid calling compute repeatedly
results = []
for x in L:
y = dask.delayed(f)(x)
results.append(y.compute())
results
|
# Collect many calls for one compute
results = []
for x in L:
y = dask.delayed(f)(x)
results.append(y)
results = dask.compute(*results)
|
在循环中调用 y.compute() 会每次等待计算结果，从而抑制并行性。
不要改变输入¶
你的函数不应直接改变输入。
不要 |
做 |
# Mutate inputs in functions
@dask.delayed
def f(x):
x += 1
return x
|
# Return new values or copies
@dask.delayed
def f(x):
x = x + 1
return x
|
如果你需要使用一个可变操作，那么首先在你的函数中进行复制：
@dask.delayed
def f(x):
x = copy(x)
x += 1
return x
避免全局状态¶
理想情况下，您的操作不应依赖于全局状态。如果仅使用线程，使用全局状态 可能 有效，但当您转向多进程或分布式计算时，您可能会遇到令人困惑的错误。
不要 |
L = []
# This references global variable L
@dask.delayed
def f(x):
L.append(x)
|
不要依赖副作用¶
延迟函数只有在计算时才会执行某些操作。您始终需要将输出传递给最终调用计算的内容。
不要 |
做 |
# Forget to call compute
dask.delayed(f)(1, 2, 3)
...
|
# Ensure delayed tasks are computed
x = dask.delayed(f)(1, 2, 3)
...
dask.compute(x, ...)
|
在第一种情况下，什么都不会发生，因为 compute() 从未被调用。
将计算分解为多个部分¶
从 Dask 的角度来看，每个 dask.delayed 函数调用都是一个单独的操作。您通过拥有许多延迟调用来实现并行性，而不是仅使用一个：Dask 不会查看带有 @dask.delayed 装饰的函数内部并并行化该代码。要实现这一点，它需要您的帮助来找到分解计算的好地方。
不要 |
做 |
# One giant task
def load(filename):
...
def process(data):
...
def save(data):
...
@dask.delayed
def f(filenames):
results = []
for filename in filenames:
data = load(filename)
data = process(data)
result = save(data)
results.append(result)
return results
dask.compute(f(filenames))
|
# Break up into many tasks
@dask.delayed
def load(filename):
...
@dask.delayed
def process(data):
...
@dask.delayed
def save(data):
...
def f(filenames):
results = []
for filename in filenames:
data = load(filename)
data = process(data)
result = save(data)
results.append(result)
return results
dask.compute(f(filenames))
|
第一个版本只有一个延迟任务，因此无法并行化。
避免过多任务¶
每个延迟任务都有几百微秒的开销。通常这没什么问题，但如果你过于精细地应用 dask.delayed，这可能会成为一个问题。在这种情况下，通常最好将许多任务分成批次，或者使用 Dask 集合之一来帮助你。
不要 |
做 |
# Too many tasks
results = []
for x in range(10000000):
y = dask.delayed(f)(x)
results.append(y)
|
# Use collections
import dask.bag as db
b = db.from_sequence(range(10000000), npartitions=1000)
b = b.map(f)
...
|
在这里，我们使用 dask.bag 来自动批量应用我们的函数。我们也可以如下构造自己的批处理。
def batch(seq):
sub_results = []
for x in seq:
sub_results.append(f(x))
return sub_results
batches = []
for i in range(0, 10000000, 10000):
result_batch = dask.delayed(batch)(range(i, i + 10000))
batches.append(result_batch)
在这里，我们构建批次，其中每个延迟的函数调用为原始输入中的许多数据点进行计算。
避免在延迟函数中调用延迟函数¶
通常，如果你是 Dask delayed 的新手，你会在各处放置 dask.delayed 调用，并希望一切顺利。虽然这可能确实有效，但它通常很慢，并且会导致难以理解的解决方案。
通常你不会在 dask.delayed 函数内部调用 dask.delayed。
不要 |
做 |
# Delayed function calls delayed
@dask.delayed
def process_all(L):
result = []
for x in L:
y = dask.delayed(f)(x)
result.append(y)
return result
|
# Normal function calls delayed
def process_all(L):
result = []
for x in L:
y = dask.delayed(f)(x)
result.append(y)
return result
|
因为普通函数只执行延迟工作，所以它非常快，因此没有理由延迟它。
不要在其他 Dask 集合上调用 dask.delayed¶
当你将一个 Dask 数组或 Dask DataFrame 放入一个延迟调用中时，该函数将接收到等效的 NumPy 或 Pandas 对象。请注意，如果你的数组很大，那么这可能会导致你的工作节点崩溃。
相反，更常见的是使用 da.map_blocks 这样的方法
不要 |
做 |
# Call delayed functions on Dask collections
import dask.dataframe as dd
df = dd.read_csv('/path/to/*.csv')
dask.delayed(train)(df)
|
# Use mapping methods if applicable
import dask.dataframe as dd
df = dd.read_csv('/path/to/*.csv')
df.map_partitions(train)
|
或者，如果该过程不适合映射，您总是可以将数组或数据帧转换为 许多 延迟对象，例如
partitions = df.to_delayed()
delayed_values = [dask.delayed(train)(part)
for part in partitions]
然而，如果你不介意将你的 Dask 数组/DataFrame 转换为单个块，那么这是可以的。
dask.delayed(train)(..., y=df.sum())
避免在延迟调用中重复放入大量输入¶
每次你传递一个具体的结果（任何不是延迟的内容）Dask 会默认对其进行哈希处理以赋予其名称。这相当快（大约 500 MB/s），但如果反复执行可能会变慢。相反，最好是也将你的数据延迟处理。
在使用分布式集群时，这一点尤为重要，以避免为每次函数调用单独发送数据。
不要 |
做 |
x = np.array(...) # some large array
results = [dask.delayed(train)(x, i)
for i in range(1000)]
|
x = np.array(...) # some large array
x = dask.delayed(x) # delay the data once
results = [dask.delayed(train)(x, i)
for i in range(1000)]
|
每次调用 dask.delayed(train)(x, ...) 都必须对 NumPy 数组 x 进行哈希处理，这会减慢速度。