在Klustron中进行机器学习

在Klustron中运行机器学习算法有一系列优势。首先，相比于在PostgreSQL中运行机器学习算法，在Klustron中运行机器学习算法不会受限于单台服务器的硬件资源限制，可以使用大量服务器的计算资源，来使用更大规模的数据集更快速递完成算法执行。这依赖于Klustron的多层级并行执行机制，该机制让Klustron可以用多台服务器的CPU和内存来执行同一个机器学习任务。

另外，相比于在数据库之外运行机器学习算法，在数据库系统中有完整的基础设施应对数据规模较大的问题，包括临时表，buffer pool，索引等。如果在数据库之外运行机器学习算法，要么需要手动实现这些基础设施（复杂度很高），要么就只能在内存中运行并且受限于数据量大于内存导致操作系统频繁做换页带来的性能严重下降等问题。

并且在数据库中运行机器学习算法，避免了跨网络节点传输大量数据的问题，也会有利于提升机器学习算法的性能。

Klustron不仅支持通过python等存储过程调用pytorch等流行的机器学习算法库来运行机器学习算法，还将在1.3版本中支持常用的机器学习算法扩展PostgresML和Apache MadLib以便实现更加强大和高性能的机器学习算法。

PostgreSQL生态的各种机器学习算法库和扩展原本在PostgreSQL中受限于单机硬件资源限制，在Klustron中就不再受限，可以处理更大量的数据，并且性能更高，耗时更短。并且这些机器学习算法都会受益于Klustron的多层级并行查询处理能力，实现更高的性能。

因此Klustron充分放大了这些in-database机器学习组件的能力，给用户带来强劲和高性能的机器学习处理能力。

下面让我们看一下如何使用PLPython直接在Klustron中执行Kmeans（最流行的无监督学习算法之一）。

psql -h 192.168.40.152 -p 47001 postgres
create user kunlun_test with password 'kunlun';
create database testdb owner kunlun_test;
grant all privileges on database testdb to kunlun_test;
alter user kunlun_test with superuser;
\q

psql -h 192.168.40.152 -p 47001 -U kunlun_test testdb
create table iris (
sepal_length REAL,
sepal_width REAL,
petal_length REAL,
petal_width REAL,
species varchar(20)
);
\copy iris from '/kunlun/iris.data' delimiter ',';

yum install python3
yum install python3-devel

python3 -V
pip3 -V

ln -s etc/alternatives/python3 usr/bin/python
ln -s /etc/alternatives/pip3 /usr/bin/pip

su - kunlun
pip install --user scikit-learn
pip install --user pandas

psql -h 192.168.40.152 -p 47001 -U kunlun_test testdb
CREATE EXTENSION plpython3u;

CREATE OR replace FUNCTION kmeans(input_table text, columns text[], clus_num int) RETURNS bytea AS
$$
from pandas import DataFrame
from sklearn.cluster import KMeans
from pickle import dumps
from pandas import pandas as pd


all_columns = ",".join(columns)
if all_columns == "":
    all_columns = "*"
 
rv = plpy.execute('SELECT %s FROM %s;' % (all_columns, plpy.quote_ident(input_table)))
 
frame = []
 
for i in rv:
    frame.append(i)
df = DataFrame(frame).apply(pd.to_numeric, errors="ignore")
kmeans = KMeans(n_clusters=clus_num, random_state=0).fit(df._get_numeric_data())
return dumps(kmeans)
 
$$ LANGUAGE plpython3u;

CREATE TABLE models (
id SERIAL PRIMARY KEY,
model BYTEA NOT NULL
);

INSERT INTO models(model) SELECT kmeans('iris', array[]::text[], 3);

CREATE OR replace FUNCTION get_kmeans_centroids(model_table text, model_column text, model_id int)
RETURNS SETOF real[] AS
$$
from pandas import DataFrame
from pickle import loads
rv = plpy.execute('SELECT %s FROM %s WHERE id = %s;' % (plpy.quote_ident(model_column), plpy.quote_ident(model_table), model_id))
model = loads(rv[0][model_column])
ret = map(list, model.cluster_centers_)
return ret
$$ LANGUAGE plpython3u;

select get_kmeans_centroids('models','model',1);

CREATE OR replace FUNCTION predict_kmeans(model_table text, model_column text, model_id int,
input_values real[]) RETURNS int[] AS
$$
from pickle import loads
 
rv = plpy.execute('SELECT %s FROM %s WHERE id = %s;' % (plpy.quote_ident(model_column), plpy.quote_ident(model_table), model_id))
model = loads(rv[0][model_column])
ret = model.predict(input_values)
return ret
 
$$ LANGUAGE plpython3u;

select predict_kmeans('models','model',1,array[[0.5,0.5,0.5,0.5]]);