分布式机器学习

Dask

Dask是一个数据分析的并行计算的框架。

• 已经集成了现有的框架，比如：numpy，pandas，scikit-learn，xgboost，lightGBM等

• API与已有框架的API一致

• 可以扩展到上千个节点，也可以在笔记本上使用

• 有低阶API可供用户定制化

组成

• 动态任务调度（Dynamic task scheduling）优化交互计算的工作量，与Airflow，Luigi，Celery或Make类似
• “大数据”集合（Big Data Collection）扩展了NumPy，pandas，Python iterators可以处理比内存大的数据及在分布式的环境上

特性

• Familiar：数据结构一致

• Flexible：提供了一个任务调度接口，可以定制化的集成其它算法

• Native：纯python环境

• Fast：减少了工作量，增加了并行计算，速度更快

• Scales up：可以扩展到1000+cores

• Scales down：可以在laptop上使用

• Responsive：有诊断系统，反馈更及时

  有一个Task Graph，与spark的类似

Install Dask

• conda安装
  • 完全安装，包含了所有的信赖，比如numpy,pandas
    conda install dask
  • 只安装内核
    conda install dask-core
    与pip install dask一样
• pip安装
  pip install “dask[compete]” # Install everything
  pip install dask # Install only core
• cluster 部署

      # 安装dask 1.2.2
      conda install dask==1.2.2 或者 pip install dask[complete]==1.2.2
      # 启动scheduler进程，并挂后台
      nohup dask-scheduler --host 172.16.36.20 &
      # 启动worker进程，指定scheduler的地址是203，端口是8786，代码中提交的端口也是8786,并挂后台
      nohup dask-worker --name work-01 172.16.36.20:8786 &
      # 关闭防火墙就可以通过8787端口查看集群状态
      sudo systemctl status firewalld # 查看防火墙状态，加d是服务
      sudo systemctl stop firewalld # 关闭防火墙
      http://172.16.36.30:8787/status

Setup

Dask有两种task scheduler

• Single machine scheduler：

  • 默认 scheduler，不用设置
  • 调用compute()方法，使用默认scheduler
  • 示例

       import dask.dataframe as dd
       df = dd.read_csv(...)
       df.x.sum().compute()  # This uses the single-machine scheduler by default
  
  

• Distributed scheduler

  • 需要设置一个Client
  • 示例

    from dask.distributed import Client
    client = Client(...)  # Connect to distributed cluster and override default
    df.x.sum().compute()  # This now runs on the distributed system
    

use case

分为两类：

• Collection example：单机处理Large Numpy/Pandas/list，类似于spark。目前80+%的Dask用户是使用这种类型。
• Custom example：自定义任务调度（Custom task scheduler），类似于Luigi，Airflow，Celery或Makefiles