精选30+云产品，助力企业轻松上云！>>>

最近要对大量的网格数据(千亿级别)建模，尝试了各种解决方案：多线程,多进程,分布式,GPU,多GPU,分布式多GPU……

而开源世界里，又有各种号称自己能搞定问题的框架：cloudera, spark, h2o, dask, rapidsai……

相信很多人遇到这种问题，也会头晕：一旦数据量大到一定程度，各种技术都好像就没那么靠谱了

在各种框架中摸爬滚打一番后，我找到一个相对稳定的思路，一是对工作的总结，二是抛砖引玉，看看大家有没有更好的解决方案

1.大数据建模动机？

有人会说，你需要将问题分解。分而治之的思想固然是万金油，但分解后的计算规模依然相当庞大，分解后的小数据又容易过拟合，这不是我们想要的结果。所以，分解的动作是一定会有的，可是具体用什么技术才能最快的达到目标呢？还是需要深入各种技术权衡取舍

2.方案比较：

A. spark

spark有相对完整的生态圈 spark + mllib + stream

特点：java生态，需要数据存储在hadoop中，并需要安装配置java框架，考虑到目前我没有java数仓团队的支持，所以暂不考虑

B. dask

dask与spark类似，不过它提供了一套基于python体系的解决方案

特点：快速轻量，通过pip安装即可，不需要数仓团队支持

C. rapidsai

rapidsai提供了一套完整的基于GPU的解决方案，也是基于python的

特点：同样快速轻量，通过conda安装即可，同样不需要数仓团队支持

3.探索：

基于以上的调研，所以考虑对 dask 和 rapidsai的方案进行探索

一个是分布式、一个是GPU，中间就衍生了很多附带框架，奈何GPU的世界还很不成熟，有些坑总是要掉进去了才知道

A.纯分布式方案

dask: 简单了解了下dask后，就不用再细想了，它是纯分布式的首选方案，架构很简单： scheduler + worker

启动了一个scheduler作为调度，剩下的就是不断的往集群里增加worker

唯一需要注意的是，每个worker分配的内存要能足够任务执行，否则worker会不断崩溃，而dask会自动尝试重启worker

1.在中心主机(192.168.0.10)上启动一个scheduler

dask-scheduler --dashboard-address 0.0.0.0:8788# dashboard-address 服务监听地址,端口可以任意,方便监控集群

2.在任意有计算资源的从机上启动worker

dask-worker tcp://192.168.0.10:8786 --nthreads 1 --nprocs 25 --memory-limit 4GB# tcp://192.168.0.10:8786 集群scheduler服务地址
# nthreads     每个worker可以使用的线程数量
# nprocs       从机可以使用的worker数量
# memory-limit 每个worker使用内存数量

3.将python任务提交给dask集群

from dask.distributed import Client# 连接到dask集群
client = Client(address='localhost:8786')# 使用client.map分发任务 类似pool.map
# user_func 你定义的需要执行的方法
# parts 可循环的数组
process_num = client.map(user_func, parts)
# 汇总结果
future = client.submit(sum, process_num)
# 通过调用result执行任务
stats = future.result()

这样就能做大数据处理了，比起spark的配置，轻量了很多

唯一需要吐槽的就是监控界面太简单了，只能简单的观察，没法管理运行的任务：

看似很丰富的监控报表，没啥用，真正有用的只有 workers 和 profile，workers能观察当前任务是否正常分配执行，profile能发现任务的性能瓶颈

B.纯GPU解决方案

目前找了下，只有rapidsai这款框架有基于python的GPU方案，那就尝试下

1.指定GPU

import os
os.environ["CUDA_DEVICE_ORDER"]="PCI_BUS_ID"
os.environ["CUDA_VISIBLE_DEVICES"]="1"# 指定第2块GPU(序号从0开始)对任务可见,必须在加载cudf,cuml……之前调用

2.读取数据

import cudfcdf1 = cudf.read_csv("/data/test.csv")

啊哈，读取数据看起来和原生的pandas没啥区别？

别被表象迷惑了，cudf不支持批次读取，仅支持分段读取，这就意味着一次读取的数据量不能超过GPU内存，这还没有考虑数据处理所需的GPU内存

用户为了处理大数据，不得不自己实现一套批次读取的逻辑，想想都是很痛苦的过程

并且cudf目前还没能实现pandas的apply方法，取而代之的是非常丑陋的 apply_rows方法

3.数据处理

from numba import cuda@cuda.jit
def test_data(in1, out1):for i in range(in1.size):out1[i] = in[i] * 2test = cdf1.apply_rows(test_data,incols=['in1'],outcols={'out1':np.int32},kwargs={})

想想每次处理数据都要自己实现一个这么麻烦的方法，是多么痛苦的体验？

哦，对了， GPU核方法里(test_data)还不能有其他python高级对象，只支持基本的python运算

4.模型训练

import cuml
from cuml.ensemble import RandomForestRegressor as curfccu_rf = curfc(max_features=1.0, max_depth=80, min_rows_per_node=2, n_estimators=80, accuracy_metric='mae')cu_rf.fit(X, y)# 调用 cuml 包下对应模型的 fit方法即可

同样，看似简单的API，局限性也很大，不能超过GPU内存大小，不能使用多GPU

所以rapidsai的局限性也很大，GPU上训练的优势并不能体现出来

5.推理加速

import sklearn, sklearn.datasets, numpy as np
from numba import cuda
from cuml import ForestInferencemodel_path = 'xgb.model'
X_test, y_test = sklearn.datasets.make_classification()
X_gpu = cuda.to_device(np.ascontiguousarray(X_test.astype(np.float32)))fm = ForestInference.load(model_path, output_class=True)
# fm 即为加速后的模型fil_preds_gpu = fm.predict(X_gpu)
accuracy_score = sklearn.metrics.accuracy_score(y_test, np.asarray(fil_preds_gpu))

推理加速还有点用，能将CPU上训练后的模型，使用GPU加速预测，支持sklearn，xgboost等，不过同样只支持单卡

综上，Rapidsai提供的GPU仅仅只是可用，相比于分布式方案，优势并不明显，数据处理得小心翼翼，不然就又打回原形只能在CPU上处理

GPU上训练也得小心，不能超过GPU内存，这就给它的能力大打折扣，毕竟我们需要一次处理大量的数据，很多机器学习方法是不支持batch train的

C.分布式+GPU方案

rapidsai同样还提供了dask-cuda包，支持将多个GPU作为分布式资源供dask集群管理，这样就实现了多GPU的分布式调度

调度多GPU类似于dask的分布式任务，只不过换下名称：

from dask_cuda import LocalCUDACluster
from dask.distributed import Clientcluster = LocalCUDACluster(n_workers=4, threads_per_worker=1)
# n_workers 只能等于GPU的数量，不匹配就会启动失败
# threads_per_worker 每个worker的进程数量(根据任务所需进程决定)client = Client(cluster)

其他都与dask相同，这样看来，dask-cuda极大的方便了多GPU的模型训练和推理，还是有用的

D.dask的XGBoost分布式方案

纳尼？dask也搞了一套xgboost分布式方案？这么不务正业的框架？我的心太乱

是的，dask下面也有一个 dask_ml.xgboost包，支持xgboost在dask集群上的分布式训练

不过看官方文档里的例子，目前还很简单，既不需要连接到client，又没有推理加速的示例

我简单入坑体验了一把，又是各种bug，所以不建议大家使用了

E.XGBoost的分布式+GPU方案

所以，如果要用xgboost，还是得用正中的，千万别搞错了，官方的叫xgboost.dask包，嘿嘿

1.数据准备：

import xgboost as xgb
from dask import dataframe as dd
from dask_ml.model_selection import train_test_splitdf_samples = dd.read_csv("/data/*.csv")
# 这里需要注意用dask.dataframe格式df_samples = df_samples.drop('Unnamed: 0', axis=1)X = df_samples.drop(['y'], axis=1)
Y = df_samples['y']X_train, X_test, y_train, y_test = train_test_split(X, Y, test_size=0.33, random_state=42)dtrain = xgb.dask.DaskDMatrix(client, X_train, y_train)dtest = xgb.dask.DaskDMatrix(client, X_test, y_test)

目前只支持dask.dataframe格式，不过dask.dataframe比起cudf.dataframe支持的原生pandas操作要多很多，也稳定不少，更安心更放心

2.模型训练：

params = {'colsample_bynode': 0.8,'learning_rate': 1,'max_depth': 16,'num_parallel_tree': 40,'objective': 'reg:squarederror','subsample': 0.8,'tree_method': 'gpu_hist','nthread': 12,'silent': False
}
watchlist = [(dtrain, 'train'), (dtest, 'test')]bst = xgb.dask.train(client, params, dtrain, evals=watchlist, early_stopping_rounds=3, num_boost_round=30, verbose_eval=True)

可以看到，比起传统的单机xgboost方法，分布式xgboost有以下特点：

支持更大的层深，max_depth 可以由原先5-10层增加到10-20层；

支持更多的booster，num_boost_round又原先的5-10个增加到20-30个；

支持决策树分裂时的GPU的并行计算，tree_method的直方图方法可以选用 gpu_hist

xgboost本身也支持随机森林，随机森林的规模也能增大不少，num_parallel_tree由原先10个增加到20-100个；

3.模型预测：

preds = xgb.dask.predict(client, bst, dtest)
# 注意这里 preds 是 dask.Array对象predictions = np.array(preds)

这样就通过dask预测了模型结果

4.总结

综上，我们通过各种分布式并行计算包，实现了传统模型在大数据上的快捷实现

目前我个人推荐 dask + xgboost.dask + (dask-cuda) 的方案，dask-cuda可选(需要GPU加速时再使用)

在各种不成熟的框架功能中跳坑了一遍后，顺利的走完了建模的一个生命周期

使得模型处理大数据的能力极大增强（目前数据处理能力可以超越CNN等深度学习框架）

这里记录下关键知识点，以备不时之需。写的不对的地方，也请大家不吝赐教！