MADlib开源机器学习框架

简介

MADlib 是一个基于 SQL 的数据库内置的开源机器学习库，具有良好的并行度和可扩展性，有高度的预测精准度。MADlib 1.14 可以与 PostgreSQL、Greenplum 和 HAWQ 等数据库系统无缝集成。

通常 SQL 查询能发现数据最明显的模式和趋势，但要想获取数据中最为有用的信息，需要的其实是完全不同的一套技术，一套牢固扎根于数学和应用数学的技能（机器学习）。将 SQL 的简单易用与数据挖掘的复杂算法结合起来，充分利用两者的优势和特点，对于广大传统数据库应用技术人员来说，就可将他们 长期积累的数据库操作技能复用到机器学习领域。

MADlib 提供了可在 SQL 查询语句中调用的函数，即可以用 select + function name 的方式来调用这个库。这就意味着，所有的数据调用和计算都在数据库内完成而不需要数据的导入导出。

MADlib 是 SQL 中的机器学习库，就注定它不关心数据可视化，本身不带数据的图形化表示功能。MADlib 作为工具，并不是传统意义上的机器学习系统软件，而只是一套可在 SQL 中调用的函数库，其出发点是让数据库技术人员用 SQL 快速完成简单的机器学习工作，比较适合做一些简单的、特征相对明显的机器学习。

设计思想

MADlib 架构的关键设计思想体现在以下方面：

操作数据库内的本地数据，避免在多个运行时环境之间不必要地移动数据。

充分利用数据库引擎功能，但将机器学习逻辑从数据库特定的实现细节中分离出来。

利用 MPP 无共享技术提供的并行性和可扩展性，如 Greenplum 或 HAWQ 数据库系统。

开放实施，保持与 Apache 社区的积极联系和持续的学术研究。

MADlib 库表现为数据库内置的函数。当函数在 SQL 语句中执行时，可以充分利用数据库引擎提供的功能。在客户端，可以使用Jupyter、 Zeppelin、psql 等工具连接数据库并调用 MADlib Function。MADlib 预处理后根据具体算法生成多个查询传入数据库服务器，之后数据库服务器执行查询并返回 String（一般是一个或多个存放结果的表）， MADlib 函数调用过程的执行流程如下：

MADlib架构如下所示，包含四个主要组件：

1、Python调用SQL模版实现的驱动函数：

2、Python实现的高级抽象层：负责算法流程控制

3、C++实现的核心函数：实现特定机器学习算法，注重性能

4、C++实现的低级数据库抽象层：对所有的 PostgreSQL 数据库内核实现细节进行抽象

支持的模型类型

MADlib支持以下机器学习模型类型：

回归：预测输出值

分类：分类目标

关联规则：购物篮分析

聚类：非监督分类

主题建模：文本领域发掘相似数据组

描述性统计：数据集分布情况

模型验证：测试数据评估模型

安装与卸载

在不同的数据库系统，安装过程不尽相同。以在 Greenplum 中安装 MADlib 为例：

1、下载MADlib 二进制压缩包

2、上传压缩包至Greenplum的Master主机

3、解压安装gppkg文件

4、将 MADlib 函数添加到 Greenplum 数据库

将madlib添加至testdb命令如下：

$GPHOME/madlib/bin/madpack install -s madlib -p greenplum -c gpadmin@mdw:5432/testdb

卸载：

gppkg -r madlib-1.18.0+2-gp6-rhel7-x86_64

数据类型

和其他机器学习语言或工具一样，MADlib 操作的基本对象也是向量与矩阵。在 MADlib中，对向量和矩阵的操作是通过一系列函数完成的

向量操作函数：

-- 向量加法
select id, madlib.array_add(array1,array2), madlib.array_sub(array1,array2) from array_tbl; 
-- 标量乘法
select id, madlib.array_scalar_mult(madlib.array_scalar_mult(array1,3),2),
madlib.array_scalar_mult(madlib.array_scalar_mult(array1,2),3) from array_tbl;

矩阵运算函数：

-- 稠密矩阵表生成稀疏表示的表
drop table if exists mat_a_sparse; 
select madlib.matrix_sparsify('mat_a', 'row=row_id, val=row_vec', 'mat_a_sparse', 'col=col_id, val=val');
-- 矩阵转置
drop table if exists mat_a_r; 
select madlib.matrix_trans('mat_a', 'row=row_id, val=row_vec','mat_a_r'); 
select * from mat_a_r order by row_id;

使用

逻辑回归算法使用如下：

-- 创建表格
CREATE TABLE patients( id INTEGER NOT NULL,
                        second_attack INTEGER,
                        treatment INTEGER,
                        trait_anxiety INTEGER);
-- 插入数据
INSERT INTO patients VALUES                                                     
(1,     1,      1,      70),
(3,     1,      1,      50),
(5,     1,      0,      40),

-- 创建模型
SELECT madlib.logregr_train(
    'patients',                                 -- 源数据表
    'patients_logregr',                         -- 结果数据表
    'second_attack',                            -- 标签列
    'ARRAY[1, treatment, trait_anxiety]',       -- 特征列
    NULL,                                       -- 分组列
    20,                                         -- 迭代次数
    'irls'                                      -- 优化器
    );

-- 显示训练迭代过程 
on SELECT * from patients_logregr;

-- 使用模型预测
SELECT p.id, madlib.logregr_predict(m.coef, ARRAY[1, p.treatment, p.trait_anxiety]),
       p.second_attack
FROM patients p, patients_logregr m
ORDER BY p.id;

自定义深度学习模型实现如下，可采用Jupyter，以Keras为例，数据处理是一样的：

-- 导入数据
DROP TABLE IF EXISTS iris_data;
CREATE TABLE iris_data(
    id serial,
    attributes numeric[],
    class_text varchar
);

INSERT INTO iris_data(id, attributes, class_text) VALUES
(1,ARRAY[5.1,3.5,1.4,0.2],'Iris-setosa'),
(2,ARRAY[4.9,3.0,1.4,0.2],'Iris-setosa'),

DROP TABLE IF EXISTS iris_train_packed, iris_train_packed_summary;

-- 数据预处理
SELECT madlib.training_preprocessor_dl('iris_train',         -- Source table
                                       'iris_train_packed',  -- Output table
                                       'class_text',         -- Dependent variable
                                       'attributes'          -- Independent variable
                                        );

模型定义与实现：

from tensorflow import keras
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Dense

model_simple = Sequential()
model_simple.add(Dense(10, activation='relu', input_shape=(4,)))
model_simple.add(Dense(10, activation='relu'))
model_simple.add(Dense(3, activation='softmax'))
    
model_simple.summary();
# 模型定义与keras一样
model_simple.to_json()

# 调用MADlib接口加载keras模型
%%sql
DROP TABLE IF EXISTS model_arch_library;
SELECT madlib.load_keras_model('model_arch_library',                            
$$
{"class_name": "Sequential", "keras_version": "2.1.6", "config": [{"class_name": "Dense", "config": {"kernel_initializer": {"class_name": "VarianceScaling", "config": {"distribution": "uniform", "scale": 1.0, "seed": null, "mode": "fan_avg"}}, "name": "dense_1", "kernel_constraint": null, "bias_regularizer": null, "bias_constraint": null, "dtype": "float32", "activation": "relu", "trainable": true, "kernel_regularizer": null, "bias_initializer": {"class_name": "Zeros", "config": {}}, "units": 10, "batch_input_shape": [null, 4], "use_bias": true, "activity_regularizer": null}}, {"class_name": "Dense", "config": {"kernel_initializer": {"class_name": "VarianceScaling", "config": {"distribution": "uniform", "scale": 1.0, "seed": null, "mode": "fan_avg"}}, "name": "dense_2", "kernel_constraint": null, "bias_regularizer": null, "bias_constraint": null, "activation": "relu", "trainable": true, "kernel_regularizer": null, "bias_initializer": {"class_name": "Zeros", "config": {}}, "units": 10, "use_bias": true, "activity_regularizer": null}}, {"class_name": "Dense", "config": {"kernel_initializer": {"class_name": "VarianceScaling", "config": {"distribution": "uniform", "scale": 1.0, "seed": null, "mode": "fan_avg"}}, "name": "dense_3", "kernel_constraint": null, "bias_regularizer": null, "bias_constraint": null, "activation": "softmax", "trainable": true, "kernel_regularizer": null, "bias_initializer": {"class_name": "Zeros", "config": {}}, "units": 3, "use_bias": true, "activity_regularizer": null}}], "backend": "tensorflow"}
$$
::json, NULL,'Sophie', 'A simple model');

SELECT model_id, model_arch, name, description FROM model_arch_library ORDER BY model_id;

# madlib.load_keras_model 通过json字符串导入keras模型配置

# 开始训练 指定loss函数 迭代次数等keras参数
SELECT madlib.madlib_keras_fit('iris_train_packed',   # 数据源
                               'iris_model',          # 输出模型名
                               'model_arch_library',  # 模型配置
                                1,                  
                                $$ loss='categorical_crossentropy', optimizer='adam', 
                                metrics=['accuracy'] $$,  
                                $$ batch_size=5, epochs=3 $$,  
                                10);
# 模型评估
SELECT madlib.madlib_keras_evaluate('iris_model',       # 输出模型名
                                   'iris_test_packed',  # 测试数据
                                   'iris_validate'      # 结果数据
                                   );

# 模型预测

SELECT madlib.madlib_keras_predict('iris_model',  # 输出模型名
                                   'iris_test',   # 测试数据
                                   'id',          # id列
                                   'attributes',  # 特征属性列
                                   'iris_predict' # 结果数据
                                   );

SELECT * FROM iris_predict ORDER BY id, rank;

# 通过sql语句可以将数据导出至numpy，用于模型可视化分析等
# 下面就是通用的数据分析
import pandas as pd
import numpy as np
import sys
import os
from matplotlib import pyplot as plt

iters_proxy = %sql SELECT metrics_iters FROM iris_model_summary;
train_accuracy_proxy = %sql SELECT training_metrics FROM iris_model_summary;
test_accuracy_proxy = %sql SELECT validation_metrics FROM iris_model_summary;

# 获取数据关键点
num_points_proxy = %sql SELECT array_length(metrics_iters,1) FROM iris_model_summary;
num_points = num_points_proxy[0]

# 转化数据
iters = np.array(iters_proxy).reshape(num_points)
train_accuracy = np.array(train_accuracy_proxy).reshape(num_points)
test_accuracy = np.array(test_accuracy_proxy).reshape(num_points)

# 绘图
plt.title('Iris validation accuracy by iteration')
plt.xlabel('Iteration number')
plt.ylabel('Accuracy')
plt.grid(True)
plt.plot(iters, train_accuracy, 'g.-', label='Train')
plt.plot(iters, test_accuracy, 'r.-', label='Test')
plt.legend();

总结

MADlib的优点和缺点一样明显，优势是存储和计算是一体，通过简单的SQL语句就可以实现机器学习模型训练，而且还兼容自定义的深度模型，DBA也可以通过简单的语句实现机器学习算法，简单优雅

缺点也一样突出，存储和计算一体，复杂的深度模型必然会占用计算资源，这可能直接影响线上其他服务。如果涉及NLP任务，需要GPU资源支持，显然也不是MADlib能做到的了，这种情况还是需要动态申请计算资源，运算完及时释放，同时保存中间数据