Dask 工作笔记

Dask 发布 0.17.2

这项工作得到了 Anaconda Inc. 和 Moore Foundation 数据驱动发现计划的支持。

我很高兴宣布 Dask 版本 0.17.2 发布。这是一个包含新功能和稳定性改进的小版本。这篇博文概述了自 2 月 12 日 0.17.0 版本发布以来的显著变化。

您可以使用 conda 安装 Dask

conda install dask

或使用 pip 从 PyPI 安装

pip install dask[complete] --upgrade

完整的更新日志可在此处查看

下面是一些值得注意的变化

Tornado 5.0

Tornado 是一个流行的并发网络编程框架，Dask 对其依赖很大。Tornado 最近发布了一个主要版本更新，其中包含了对 Dask 非常重要的功能，同时也修复了一些错误。

新的 IOStream.read_into 方法允许 Dask 通信（或任何使用此 API 的人）更有效地通过网络传输大型数据集，减少复制次数。这使得 Dask 能够利用现代超级计算机上可用的高性能网络。在我们进行测试的 Cheyenne 系统上，通过这项更改（在使用少量 worker 进程时），我们能够充分利用 Infiniband 网络提供的 3GB/s 带宽。

非常感谢 Antoine Pitrou 和 Ben Darnell 在这方面做出的努力。

同时，更新到 Tornado 5.0 也出现了一些意料之外的问题。更广泛地使用 bytearrays 而不是 bytes 导致了与 Snappy 等压缩库以及 Python 2 的兼容问题，因为它们不期望这些类型。在 distributed.__version__ == 1.21.3 中有一个短暂的窗口，如果存在 Tornado 5.0 则会启用此功能，但如果同时存在 Snappy 则会行为异常。

HTTP 文件系统

Dask 利用一种 类似文件系统的协议 来访问远程数据。这使得像下面这样的命令能够工作

import dask.dataframe as dd

df = dd.read_parquet('s3://...')
df = dd.read_parquet('hdfs://...')
df = dd.read_parquet('gcs://...')

我们现在添加了 http 和 https 文件系统，用于直接从 Web 服务器读取数据。如果 Web 服务器支持范围查询，这些文件系统也支持随机访问。

df = dd.read_parquet('https://...')

与 S3, HDFS, GCS 等一样，您也可以在 Dask 开发之外使用这些工具。这里我们读取 Pandas 许可证的前二十个字节

from dask.bytes.http import HTTPFileSystem
http = HTTPFileSystem()
with http.open('https://raw.githubusercontent.com/pandas-dev/pandas/master/LICENSE') as f:
    print(f.read(20))

b'BSD 3-Clause License'

感谢完成这项工作并负责 Dask 字节处理的 Martin Durant。更多信息请参阅远程数据文档。

修复了 Dask dataframe shuffle 中的一个正确性错误

我们发现并解决了 dask.dataframe 的 shuffle 中的一个正确性错误，该错误导致在进行连接 (joins) 和 groupby-applies 等具有许多分区的复杂操作时丢失一些行。

更多信息请参阅 dask/dask #3201。

集群超类和智能自适应部署

有许多 Python 子项目可以帮助您在不同的集群资源管理器上部署 Dask，例如 Yarn、SGE、Kubernetes、PBS 等。这些项目或多或少都具有相同的 API，我们现在已将其整合到一个一致的接口中，下游项目可以在 distributed.deploy.Cluster 中继承该接口。

现在我们有了一个一致的接口，我们开始投入更多精力来改进这些系统的整体接口和智能化程度。这包括如下所示的令人愉快的 IPython 小部件

以及围绕自适应部署的改进逻辑。自适应部署允许集群根据当前工作负载自动进行伸缩。如果您最近提交了大量工作，调度程序将估算其持续时间并请求适量的 worker 以快速完成计算。计算完成后，调度程序将释放 worker 返回系统以释放资源。

这里的逻辑已得到显著改进，包括以下几点

  • 您可以为自适应性指定最小和最大限制
  • 调度程序估算计算持续时间并适当地请求 worker
  • 在归还 worker 时会有一些额外的延迟，以避免滞后或重复请求和返回 worker 的情况。

一些来自相关项目的新闻

  • 新生的 daskernetes 项目已更名为 dask-kubernetes。这取代了之前一个用于在 Google Cloud Platform 上启动 Dask 的项目（该项目尚未发布）。那个项目已更名为 dask-gke。
  • 一个新项目 dask-jobqueue 已启动，用于处理在传统的批处理队列系统（如 PBS、SLURM、SGE、TORQUE 等）上启动 Dask 集群。该项目源自 Pangeo 合作。
  • 一个 Dask Helm chart 已添加到 Helm 的稳定通道。

致谢

以下人员自 2 月 12 日 0.17.0 版本发布以来为 dask/dask 仓库做出了贡献：

  • Anderson Banihirwe
  • Dan Collins
  • Dieter Weber
  • Gabriele Lanaro
  • John Kirkham
  • James Bourbeau
  • Julien Lhermitte
  • Matthew Rocklin
  • Martin Durant
  • Max Epstein
  • nkhadka
  • okkez
  • Pangeran Bottor
  • Rich Postelnik
  • Scott M. Edenbaum
  • Simon Perkins
  • Thrasibule
  • Tom Augspurger
  • Tor E Hagemann
  • Uwe L. Korn
  • Wes Roach

以下人员自 2 月 12 日 1.21.0 版本发布以来为 dask/distributed 仓库做出了贡献：

  • Alexander Ford
  • Andy Jones
  • Antoine Pitrou
  • Brett Naul
  • Joe Hamman
  • John Kirkham
  • Loïc Estève
  • Matthew Rocklin
  • Matti Lyra
  • Sven Kreiss
  • Thrasibule
  • Tom Augspurger
博客评论由 Disqus 提供支持