首页 > 其他分享> 文章详细

Module-Dask并行任务调度

2020-02-01 22:39:57 阅读：407 来源： 互联网

标签：INFO 10.0 distributed worker Module Dask scheduler 任务调度

Dask 并行任务调度

Dask 说明介绍

Dask是用于 Python 中并行计算的灵活库。

达斯由两部分组成：

动态任务调度针对计算进行了优化。这类似于 Airflow，Luigi，Celery 或 Make，但已针对交互式计算工作负载进行了优化。
“大数据”集合（如并行数组，数据帧和列表）将诸如 NumPy，Pandas 或 Python 迭代器之类的通用接口扩展到内存或分布式环境。这些并行集合在动态任务计划程序之上运行。

达斯克强调以下优点：

熟悉：提供并行的 NumPy 数组和 Pandas DataFrame 对象
灵活：提供任务计划界面，以实现更多自定义工作负载并与其他项目集成。
Native：在纯 Python 中启用分布式计算并可以访问 PyData 堆栈。
快速：以低开销，低延迟和快速数值算法所需的最少序列化操作
扩大规模：在具有 1000 个核心的集群上弹性运行
缩小：在单个过程中轻松设置并在笔记本电脑上运行
响应式：在设计时考虑了交互式计算，它提供了快速的反馈和诊断功能，以帮助人类

20191205_Dask_架构图.png

Dask 分析

(env36) [scfan@fdm tools]$ dask-scheduler
(env36) [scfan@fdm ~]$ dask-worker 10.0.2.14:8786

python3

Dask-资源分析
Dask-任务管理

Dask 优缺点

优点

支持单机、分布式环境
类 Pandas 风格,修改成本低

缺点

Dask-DataFrame
- 读取文件不支持 excel。支持 read_csv read_table read_fwf read_parquet read_hdf read_json read_orc

Dask 部署

附件

性能测试

使用自主建模-字段加工节点测试 Pandas & Dask 性能

参考资源

Jupyter-Data Science with Python and Dask

Dask & Pandas 语法差异表

Github-Dask Collections API compatibility

样例

# Dask 没有 pandas.core.series.Series
if data_mode.upper() == 'DASK':
    pass
else:
    if varname.startswith('df') and not isinstance(argls[index], pandas.core.series.Series):
        raise RuntimeError('第%s个参数必须为一列' % (index + 1))

# Dask DataFrame.replace 没有 inplace 参数
if data_mode == 'DASK':
    data = data.replace(to_replace='nan',value='')
else:
    data.replace(to_replace='nan',value='',inplace=True)

# Dask DataFrame.to_csv
# data.to_csv('a1.csv') 会创建目录
# data.to_csv(['a1.csv']) 会创建文件
# data.to_csv('a-*.csv') 会创建分区文件,创建多个文件
if data_mode == 'DASK':
    data.to_csv(['a1.csv'],index=False)
else:
    data.to_csv('a.csv',index=False)

Dask & Pandas 细节语法性能差异

开启程序

Dask-scheduler

开启 dask-scheduler

(env36) [scfan@fdm tools]$ dask-scheduler
distributed.scheduler - INFO - -----------------------------------------------
distributed.dashboard.proxy - INFO - To route to workers diagnostics web server please install jupyter-server-proxy: pip install jupyter-server-proxy
distributed.scheduler - INFO - Local Directory:    /tmp/scheduler-bdk4b7li
distributed.scheduler - INFO - -----------------------------------------------
distributed.scheduler - INFO - Clear task state
distributed.scheduler - INFO -   Scheduler at:      tcp://10.0.2.14:8786
distributed.scheduler - INFO -   dashboard at:                     :8787
distributed.scheduler - INFO - Register tcp://10.0.2.14:30547
distributed.scheduler - INFO - Starting worker compute stream, tcp://10.0.2.14:30547
distributed.core - INFO - Starting established connection
distributed.scheduler - INFO - Register tcp://10.0.2.14:9190
distributed.scheduler - INFO - Starting worker compute stream, tcp://10.0.2.14:9190
distributed.core - INFO - Starting established connection

Dask-Scheduler 可视化界面
20191205_Dask_Scheduler_可视化界面.png

Dask-Worker

开启 Worker

(env36) [scfan@fdm tools]$ dask-worker 10.0.2.14:8786
distributed.nanny - INFO -         Start Nanny at: 'tcp://10.0.2.14:12075'
distributed.diskutils - INFO - Found stale lock file and directory '/home/scfan/project/FISAMS/branches/branch_scfan/src/server/fdm/tools/worker-yyz2l21f', purging
distributed.dashboard.proxy - INFO - To route to workers diagnostics web server please install jupyter-server-proxy: pip install jupyter-server-proxy
distributed.worker - INFO -       Start worker at:      tcp://10.0.2.14:17181
distributed.worker - INFO -          Listening to:      tcp://10.0.2.14:17181
distributed.worker - INFO -          dashboard at:            10.0.2.14:36300
distributed.worker - INFO - Waiting to connect to:       tcp://10.0.2.14:8786
distributed.worker - INFO - -------------------------------------------------
distributed.worker - INFO -               Threads:                          4
distributed.worker - INFO -                Memory:                   10.32 GB
distributed.worker - INFO -       Local Directory: /home/scfan/project/FISAMS/branches/branch_scfan/src/server/fdm/tools/worker-5304u4tp
distributed.worker - INFO - -------------------------------------------------
distributed.worker - INFO -         Registered to:       tcp://10.0.2.14:8786
distributed.worker - INFO - -------------------------------------------------
distributed.core - INFO - Starting established connection

Dask-Worker 可视化界面
20191205_Dask_Worker_可视化界面.png

Dask 对比

Dask 缺点

dataframe
- 不提供 sql 支持,可以使用 dask.dataframe.from_sql
- 支持的数据格式
  - Tabular: Parquet, ORC, CSV, Line Delimited JSON, Avro, text
  - Arrays: HDF5, NetCDF, Zarr, GRIB
  - 不支持 Excel

Dask 优点

Dask 可以抵抗工作节点的故障
Dask 虽然较新 2015 年,但是已经成熟,而且随 Pandas 更新而更新
Dask 是通用并行编程解决方案。类似 Pandsa,使用方便,和 pandas 有细微差异
支持本地 Dask 的概要分析和检查执行情况
- https://docs.dask.org/en/latest/diagnostics-local.html#example

Dask 支持项

支持单机调度程序、分布式调度程序(本地或集群)
dask-worker 资源控制
–resources
用于任务约束的资源，例如“ GPU = 2 MEM = 10e9”。
资源分别应用于每个工作进程（仅在使用“ –nprocs”启动多个工作进程时才相关）。
可视化界面
- http://192.168.172.72:27831/status Scheduler
- http://192.168.172.72:8787/status worker

SuperScfan 发布了41 篇原创文章 · 获赞 10 · 访问量 4万+ 私信关注

标签：INFO,10.0,distributed,worker,Module,Dask,scheduler,任务调度
来源： https://blog.csdn.net/qq_21165007/article/details/104138876

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

ICode9