用 Mars Remote API 轻松分布式执行 Python 函数-低调大师

用 Mars Remote API 轻松分布式执行 Python 函数

2020-11-04 491

Mars 是一个并行和分布式 Python 框架，能轻松把单机大家耳熟能详的的 numpy、pandas、scikit-learn 等库，以及 Python 函数利用多核或者多机加速。这其中，并行和分布式 Python 函数主要利用 Mars Remote API。

启动 Mars 分布式环境可以参考：

命令行方式在集群中部署。
Kubernetes 中部署。
MaxCompute 开箱即用的环境，购买了 MaxCompute 服务的可以直接使用。

如何使用 Mars Remote API

使用 Mars Remote API 非常简单，只需要对原有的代码做少许改动，就可以分布式执行。

拿用蒙特卡洛方法计算 π 为例。代码如下，我们编写了两个函数，calc_chunk 用来计算每个分片内落在圆内的点的个数，calc_pi 用来把多个分片 calc_chunk 计算的结果汇总最后得出 π 值。

from typing import List
import numpy as np

def calc_chunk(n: int, i: int):
    # 计算n个随机点（x和y轴落在-1到1之间）到原点距离小于1的点的个数
    rs = np.random.RandomState(i)
    a = rs.uniform(-1, 1, size=(n, 2))
    d = np.linalg.norm(a, axis=1)
    return (d < 1).sum()

def calc_pi(fs: List[int], N: int):
    # 将若干次 calc_chunk 计算的结果汇总，计算 pi 的值
    return sum(fs) * 4 / N

N = 200_000_000
n = 10_000_000

fs = [calc_chunk(n, i)
      for i in range(N // n)]
pi = calc_pi(fs, N)
print(pi)

%%time 下可以看到结果：

3.1416312
CPU times: user 9.47 s, sys: 2.62 s, total: 12.1 s
Wall time: 12.3 s

在单机需要 12.3 s。

要让这个计算使用 Mars Remote API 并行起来，我们不需要对函数做任何改动，需要变动的仅仅是最后部分。

import mars.remote as mr

# 函数调用改成 mars.remote.spawn
fs = [mr.spawn(calc_chunk, args=(n, i))
      for i in range(N // n)]
# 把 spawn 的列表传入作为参数，再 spawn 新的函数
pi = mr.spawn(calc_pi, args=(fs, N))
# 通过 execute() 触发执行，fetch() 获取结果
print(pi.execute().fetch())

%%time 下看到结果：

3.1416312
CPU times: user 29.6 ms, sys: 4.23 ms, total: 33.8 ms
Wall time: 2.85 s

结果一模一样，但是却有数倍的性能提升。

可以看到，对已有的 Python 代码，Mars remote API 几乎不需要做多少改动，就能有效并行和分布式来加速执行过程。

一个例子

为了让读者理解 Mars Remote API 的作用，我们从另一个例子开始。现在我们有一个数据集，我们希望对它们做一个分类任务。要做分类，我们有很多算法和库可以选择，这里我们用 RandomForest、LogisticRegression，以及 XGBoost。

困难的地方是，除了有多个模型选择，这些模型也会包含多个超参，那哪个超参效果最好呢？对于调参不那么有经验的同学，跑过了才知道。所以，我们希望能生成一堆可选的超参，然后把他们都跑一遍，看看效果。

准备数据

这个例子里我们使用 otto 数据集。

首先，我们准备数据。读取数据后，我们按 2：1 的比例把数据分成训练集和测试集。

import pandas as pd
from sklearn.preprocessing import LabelEncoder
from sklearn.model_selection import train_test_split

def gen_data():
    df = pd.read_csv('otto/train.csv')
    
    X = df.drop(['target', 'id'], axis=1)
    y = df['target']
    
    label_encoder = LabelEncoder()
    label_encoder.fit(y)
    y = label_encoder.transform(y)
    
    return train_test_split(X, y, test_size=0.33, random_state=123)

X_train, X_test, y_train, y_test = gen_data()

模型

接着，我们使用 scikit-learn 的 RandomForest 和 LogisticRegression 来处理分类。

RandomForest：

from sklearn.ensemble import RandomForestClassifier

def random_forest(X_train: pd.DataFrame, 
                  y_train: pd.Series, 
                  verbose: bool = False,
                  **kw):
    model = RandomForestClassifier(verbose=verbose, **kw)
    model.fit(X_train, y_train)
    return model

接着，我们生成供 RandomForest 使用的超参，我们用 yield 的方式来迭代返回。

def gen_random_forest_parameters():
    for n_estimators in [50, 100, 600]:
        for max_depth in [None, 3, 15]:
            for criterion in ['gini', 'entropy']:
                yield {
                    'n_estimators': n_estimators,
                    'max_depth': max_depth,
                    'criterion': criterion
                }

LogisticRegression 也是这个过程。我们先定义模型。

from sklearn.linear_model import LogisticRegression

def logistic_regression(X_train: pd.DataFrame,
                        y_train: pd.Series,
                        verbose: bool = False,
                        **kw):
    model = LogisticRegression(verbose=verbose, **kw)
    model.fit(X_train, y_train)
    return model

接着生成供 LogisticRegression 使用的超参。

def gen_lr_parameters():
    for penalty in ['l2', 'none']:
        for tol in [0.1, 0.01, 1e-4]:
            yield {
                'penalty': penalty,
                'tol': tol
            }

XGBoost 也是一样，我们用 XGBClassifier 来执行分类任务。

from xgboost import XGBClassifier

def xgb(X_train: pd.DataFrame,
        y_train: pd.Series,
        verbose: bool = False,
        **kw):
    model = XGBClassifier(verbosity=int(verbose), **kw)
    model.fit(X_train, y_train)
    return model

生成一系列超参。

def gen_xgb_parameters():
    for n_estimators in [100, 600]:
        for criterion in ['gini', 'entropy']:
            for learning_rate in [0.001, 0.1, 0.5]:
                yield {
                    'n_estimators': n_estimators,
                    'criterion': criterion,
                    'learning_rate': learning_rate
                }

验证

接着我们编写验证逻辑，这里我们使用 log_loss 来作为评价函数。

from sklearn.metrics import log_loss

def metric_model(model, 
                 X_test: pd.DataFrame,
                 y_test: pd.Series) -> float:
    if isinstance(model, bytes):
        model = pickle.loads(model)
    y_pred = model.predict_proba(X_test)
    return log_loss(y_test, y_pred)


def train_and_metric(train_func,
                     train_params: dict,
                     X_train: pd.DataFrame, 
                     y_train: pd.Series, 
                     X_test: pd.DataFrame, 
                     y_test: pd.Series,
                     verbose: bool = False
                     ):
    # 把训练和验证封装到一起
    model = train_func(X_train, y_train, verbose=verbose, **train_params)
    metric = metric_model(model, X_test, y_test)
    return model, metric

找出最好的模型

做好准备工作后，我们就开始来跑模型了。针对每个模型，我们把每次生成的超参们送进去训练，除了这些超参，我们还把 n_jobs 设成 -1，这样能更好利用单机的多核。

results = []

# -------------
# Random Forest
# -------------

for params in gen_random_forest_parameters():
    print(f'calculating on {params}')
    # fixed random_state
    params['random_state'] = 123
    # use all CPU cores
    params['n_jobs'] = -1
    model, metric = train_and_metric(random_forest, params,
                                     X_train, y_train,
                                     X_test, y_test)
    print(f'metric: {metric}')
    results.append({'model': model, 
                    'metric': metric})
    
# -------------------
# Logistic Regression
# -------------------

for params in gen_lr_parameters():
    print(f'calculating on {params}')
    # fixed random_state
    params['random_state'] = 123
    # use all CPU cores
    params['n_jobs'] = -1
    model, metric = train_and_metric(logistic_regression, params,
                                     X_train, y_train,
                                     X_test, y_test)
    print(f'metric: {metric}')
    results.append({'model': model, 
                    'metric': metric})
    
# -------
# XGBoost
# -------
    
for params in gen_xgb_parameters():
    print(f'calculating on {params}')
    # fixed random_state
    params['random_state'] = 123
    # use all CPU cores
    params['n_jobs'] = -1
    model, metric = train_and_metric(xgb, params,
                                     X_train, y_train,
                                     X_test, y_test)
    print(f'metric: {metric}')
    results.append({'model': model, 
                    'metric': metric})

运行一下，需要相当长时间，我们省略掉一部分输出内容。

calculating on {'n_estimators': 50, 'max_depth': None, 'criterion': 'gini'}
metric: 0.6964123781828575
calculating on {'n_estimators': 50, 'max_depth': None, 'criterion': 'entropy'}
metric: 0.6912312790832288
# 省略其他模型的输出结果
CPU times: user 3h 41min 53s, sys: 2min 34s, total: 3h 44min 28s
Wall time: 31min 44s

从 CPU 时间和 Wall 时间，能看出来这些训练还是充分利用了多核的性能。但整个过程还是花费了 31 分钟。

使用 Remote API 分布式加速

现在我们尝试使用 Remote API 通过分布式方式加速整个过程。

集群方面，我们使用最开始说的第三种方式，直接在 MaxCompute 上拉起一个集群。大家可以选择其他方式，效果是一样的。

n_cores = 8
mem = 2 * n_cores  # 16G
# o 是 MaxCompute 入口，这里创建 10 个 worker 的集群，每个 worker 8核16G
cluster = o.create_mars_cluster(10, n_cores, mem, image='extended')

为了方便在分布式读取数据，我们对数据处理稍作改动，把数据上传到 MaxCompute 资源。对于其他环境，用户可以考虑 HDFS、Aliyun OSS 或者 Amazon S3 等存储。

if not o.exist_resource('otto_train.csv'):
    with open('otto/train.csv') as f:
        # 上传资源
        o.create_resource('otto_train.csv', 'file', fileobj=f)
        
def gen_data():
    # 改成从资源读取
    df = pd.read_csv(o.open_resource('otto_train.csv'))
    
    X = df.drop(['target', 'id'], axis=1)
    y = df['target']
    
    label_encoder = LabelEncoder()
    label_encoder.fit(y)
    y = label_encoder.transform(y)
    
    return train_test_split(X, y, test_size=0.33, random_state=123)

稍作改动之后，我们使用 mars.remote.spawn 方法来让 gen_data 调度到集群上运行。

import mars.remote as mr

# n_output 说明是 4 输出
# execute() 执行后，数据会读取到 Mars 集群内部
data = mr.ExecutableTuple(mr.spawn(gen_data, n_output=4)).execute()
# remote_ 开头的都是 Mars 对象，这时候数据在集群内，这些对象只是引用
remote_X_train, remote_X_test, remote_y_train, remote_y_test = data

目前 Mars 能正确序列化 numpy ndarray、pandas DataFrame 等，还不能序列化模型，所以，我们要对 train_and_metric 稍作改动，把模型 pickle 了之后再返回。

def distributed_train_and_metric(train_func,
                                 train_params: dict,
                                 X_train: pd.DataFrame, 
                                 y_train: pd.Series, 
                                 X_test: pd.DataFrame, 
                                 y_test: pd.Series,
                                 verbose: bool = False
                                 ):
    model, metric = train_and_metric(train_func, train_params,
                                     X_train, y_train, 
                                     X_test, y_test, verbose=verbose)
    return pickle.dumps(model), metric

后续 Mars 支持了序列化模型后可以直接 spawn 原本的函数。

接着我们就对前面的执行过程稍作改动，把函数调用全部都用 mars.remote.spawn 来改写。

import numpy as np

tasks = []
models = []
metrics = []

# -------------
# Random Forest
# -------------

for params in gen_random_forest_parameters():
    # fixed random_state
    params['random_state'] = 123
    task = mr.spawn(distributed_train_and_metric,
        args=(random_forest, params,
              remote_X_train, remote_y_train,
              remote_X_test, remote_y_test), 
        kwargs={'verbose': 2},
        n_output=2
    )
    tasks.extend(task)
    # 把模型和评价分别存储
    models.append(task[0])
    metrics.append(task[1])
    
    
# -------------------
# Logistic Regression
# -------------------

for params in gen_lr_parameters():
    # fixed random_state
    params['random_state'] = 123
    task = mr.spawn(distributed_train_and_metric,
        args=(logistic_regression, params,
              remote_X_train, remote_y_train,
              remote_X_test, remote_y_test), 
        kwargs={'verbose': 2},
        n_output=2
    )
    tasks.extend(task)
    # 把模型和评价分别存储
    models.append(task[0])
    metrics.append(task[1])

# -------
# XGBoost
# -------
    
for params in gen_xgb_parameters():
    # fixed random_state
    params['random_state'] = 123
    # 再指定并发为核的个数
    params['n_jobs'] = n_cores
    task = mr.spawn(distributed_train_and_metric,
        args=(xgb, params,
              remote_X_train, remote_y_train,
              remote_X_test, remote_y_test), 
        kwargs={'verbose': 2},
        n_output=2
    )
    tasks.extend(task)
    # 把模型和评价分别存储
    models.append(task[0])
    metrics.append(task[1])


# 把顺序打乱，目的是能分散到 worker 上平均一点
shuffled_tasks = np.random.permutation(tasks)
_ = mr.ExecutableTuple(shuffled_tasks).execute()

可以看到代码几乎一致。

运行查看结果：

CPU times: user 69.1 ms, sys: 10.9 ms, total: 80 ms
Wall time: 1min 59s

时间一下子从 31 分钟多来到了 2 分钟，提升 15x+。但代码修改的代价可以忽略不计。

细心的读者可能注意到了，分布式运行的代码中，我们把模型的 verbose 给打开了，在分布式环境下，因为这些函数远程执行，打印的内容只会输出到 worker 的标准输出流，我们在客户端不会看到打印的结果，但 Mars 提供了一个非常有用的接口来让我们查看每个模型运行时的输出。

以第0个模型为例，我们可以在 Mars 对象上直接调用 fetch_log 方法。

print(models[0].fetch_log())

输出我们简略一部分。

building tree 1 of 50
building tree 2 of 50
building tree 3 of 50
building tree 4 of 50
building tree 5 of 50
building tree 6 of 50
# 中间省略
building tree 49 of 50
building tree 50 of 50

要看哪个模型都可以通过这种方式。试想下，如果没有 fetch_log API，你确想看中间过程的输出有多麻烦。首先这个函数在哪个 worker 上执行，不得而知；然后，即便知道是哪个 worker，因为每个 worker 上可能有多个函数执行，这些输出就可能混杂在一起，甚至被庞大日志淹没了。fetch_log 接口让用户不需要关心在哪个 worker 上执行，也不用担心日志混合在一起。

想要了解 fetch_log 接口，可以查看文档。

还有更多

Mars Remote API 的能力其实不止这些，举个例子，在 remote 内部可以 spawn 新的函数；也可以调用 Mars tensor、DataFrame 或者 learn 的算法。这些内容，读者们可以先行探索，后续我们再写别的文章介绍。

总结

Mars Remote API 通过并行和分布式 Python 函数，用很小的修改代价，极大提升了执行效率。

原文链接
本文为阿里云原创内容，未经允许不得转载。

微信关注我们

原文链接：https://my.oschina.net/yunqi/blog/4702185

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

直接保存“DataFrame表格”到本地，这个“骚操作”90%以上的人不知道！

知识是需要积累的，有些冷知识、骚操作，你可能现在不需要，但是只有当你玩儿过，以后再碰到这个需求，你才会有印象，方便查询。引入问题其实，这个知识点也是在群里面遇到了，如果当时问我，我也会很蒙逼。因此，我做了一个简单的学习，并将其整理后，供大家学习和参考。比如说：我们得到了一个df_new表格，我们想要将其保存在本地，应该怎么办呢？保存图片，你可能用的多。但是保存这个表格，你估计就不一定知道了。为什么需要将df_new保存在本地呢？其实提问者是为了将表格保存在本地，后面需要完成自动化群发消息的操作，这样一来，你不仅仅可以发图片，现在还可以发表格了，是不是很舒服？那么，这样一个操作，应该怎么完成呢？下面我为大家慢慢讲述。方法介绍完成这个需求使用的是dataframe_image库，因此在使用他之前，需要我们先安装这个库。 pipinstalldataframe_image 然后在使用之前，还需要导入这个库。 importdataframe_imageasdfi 接着，调用dfi中的export()方法，就可以实现这个需求。但是关于这个知识点，你在百度上面其实看不到任何解答，那么...

2020-11-03

651

简介：本文将简要介绍卫星定位的原理和应用情况，方便大家对北斗、卫星定位有更多的了解。 5G和北斗，是国之重器。北斗作为卫星定位系统，目前在国际上已处于领先地位，而且已经渗透到我们工作和生活的方方面面。本文将简要介绍卫星定位的原理和应用情况，方便大家对北斗、卫星定位有更多的了解。卫星定位的原理卫星定位系统的英文是Global Navigation Satellite System(GNSS），虽然直接翻译过来是导航卫星系统，但它真正提供的能力是定位，能定位后，导航就变得相对简单了。卫星定位的原理，是利用卫星播发时间信号，当设备接收到后，可以根据信号发射时间和本地时间，计算出信号传输时间，再结合光速获得卫星-设备距离。有了多颗卫星的信号，可以列出一组方程，求解4个未知数：设备的三维坐标x/y/z，以及本地时间与GNSS系统的时间差。式中的代表卫星j的三维坐标，这个坐标可以通过卫星星历计算获得。星历是描述卫星运行轨道的一组参数，卫星轨道是一个椭圆，通过几个参数和时间，可以唯一确定卫星的准确位置。星历的获取有两种方式，一种是卫星直接播发，这种方式的好处是定位过程不依赖卫星信号以...

2020-11-04

569

资源下载

更多资源

优质分享App

近一个月的开发和优化，本站点的第一个app全新上线。该app采用极致压缩，本体才4.36MB。系统里面做了大量数据访问、缓存优化。方便用户在手机上查看文章。后续会推出HarmonyOS的适配版本。

Mario

马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长，特征是大鼻子、头戴帽子、身穿背带裤，还留着胡子。与他的双胞胎兄弟路易基一起，长年担任任天堂的招牌角色。

腾讯云软件源

为解决软件依赖安装时官方源访问速度慢的问题，腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构，目前腾讯云软件源站支持公网访问和内网访问。

Nacos

Nacos /nɑ:kəʊs/ 是 Dynamic Naming and Configuration Service 的首字母简称，一个易于构建 AI Agent 应用的动态服务发现、配置管理和AI智能体管理平台。Nacos 致力于帮助您发现、配置和管理微服务及AI智能体应用。Nacos 提供了一组简单易用的特性集，帮助您快速实现动态服务发现、服务配置、服务元数据、流量管理。Nacos 帮助您更敏捷和容易地构建、交付和管理微服务平台。