算法训练和模型部署如何避免多次重写数据预处理代码

六月北京 | 高性能计算之GPU CUDA培训

6月22-24日三天密集式学习快速带你入门阅读全文>

正文共1927个字，预计阅读时间10分钟。

前言

前段时间，我们对接算法的工程师哭丧的和我说，模型生成后一般都要部署成API的形态对外提供服务，但是算法工程师并没有提供如何将一条数据转化特征向量的方法，他能拿到的是代码逻辑以及一些“中间元数据”。数据预处理本来就复杂，翻译也是一件极其困难的事情。我解释了这件事情难以解决的原因，但是显然他还是有些失望。

今天的目标就是谈谈如何尝试改善这件事情。

解决方案

在我看来，之前业界已经给出解决方案了，就是pipeline，pipeline不仅仅包括数据特征化，还包括模型。Spark也是这么做的，但是其实应用并不广泛。原因有如下几个：

1、不同的框架pipeline模式不一样。比如Sklearn的pipeline并没不太容易用在工程团队上,毕竟大部分研发工程师都是用java/c++系的。

2、Spark 的pipeline 不适合API服务，因为他是为了批处理设计的，而不是为了响应时间。

知道原因后，解决方案就变得相对直观了：

1、用一种统一的语言描述pipeline，横跨数据处理框架和算法框架。

2、pipeline对单条数据处理必须能够在毫秒级，同时需要保持数据预处理离线训练和online预测/流预测的一致性。

用一种统一的语言描述pipeline

能做统一描述的其实是SQL,我们会将数据特征化流程也抽象成了算法训练，通过调参的方式干预特征处理，另外，我们尽可能的根据任务进行高级抽象而不是小功能点。这点很重要。比如：

 1-- 把文本字段转化为tf/idf向量,可以自定义词典 2train orginal_text_corpus as TfIdfInPlace.`/tmp/tfidfinplace` 3where inputCol="content" 4-- 分词相关配置 5and ignoreNature="true" 6-- 停用词路径 7and stopWordPath="/tmp/tfidf/stopwords" 8-- 高权重词路径 9and priorityDicPath="/tmp/tfidf/prioritywords"10-- 高权重词加权倍数11and priority="5.0"12and nGram="2,3"13;

我们将tf/idf特征化抽象成了一个算法，通过参数配置覆盖了如下的一些诉求：

分词（可以指定自定义词典）
过滤停用词
ngram特征组合
字符转化为数字
计算idf/tf值
记在权重高的词汇，并且给对应特征加权。

如果是在训练阶段，我们直接加载模型的数据部分，从而让后续的算法可以继续进行处理。

使用方式如下：

1load parquet.`/tmp/tfidfinplace/data` 2as trainningdata;

我们也可能在其他的批处理或者流程序里去使用它预处理新的数据，这个时候我们需要
先注册一下：

1register TfIdfInPlace.`/tmp/tfidfinplace` as tfidf;

然后通过UDF函数的方式去使用即可：

1select tfidf(content) from hivetable;

如果你是部署成API服务，那么通过接口注册后，可以使用类似下面的http请求：

1curl -XPOST 'http://127.0.0.1:9003/model/predict' -d '2data=["你好，世界"] & 3dataType=string &4pipeline= tfidf,bayes

我这里额外添加了一个贝叶斯贝叶斯模型，这里pipeline的的调用相当于 bayes(tfidf("你好，世界")) 最后返回的是一个预测结果。

pipeline对单条数据处理必须能够在毫秒级

这个如何能做到呢？这就需要我们保存每个“数据处理模型”中间的元数据以及计算规则。比如以前面的TfIdfInPlace为例，他训练完成后会保存所有的训练参数，词空间，词和数字的映射等等。这样我们下次使用时就可以加载这些元数据，并且按特定的规则对新数据进行处理。

因为训练时的数据预处理和预测时的数据预处理本质是不同的，训练时的数据预处理只能针对批量数据，从中学习特征化的方式，而预测时的数据预处理更偏向于“利用训练时学到的经验仅仅进行计算”，这种天然不匹配带来的成本在于，你需要针对pipeline里的每个模型的预测部分（包括数据预处理和算法模型）进行重新的实现，而无法复用之前批训练时的逻辑。

对于MLSQL而言，它重新实现了大部分Spark mllib算法/数据处理模型的预测逻辑，增加了更多高阶的数据预处理模型，并且提供对tensorflow,sklearn,dl4j等框架的预测支持。

实际案例

目前StreamingPro已经实现了一个案例，比如下面的代码通过SVM和随机深林实现了一个文本分类,特征工程用的是TfIdfInPlace算法：

 1set traning_dir = "/tmp/lwys_corpus"; 2-- 加载数据 3load csv.`/Users/allwefantasy/Downloads/lwys_corpus` options header="true" and delimiter="\t" and quote="'" 4as lwys_corpus; 5select cut as features,cast(sid as int) as label from lwys_corpus 6as orginal_text_corpus; 7-- 把文本字段转化为tf/idf向量,可以自定义词典 8train orginal_text_corpus as TfIdfInPlace.`${traning_dir}/tfidf`  9where inputCol="features" 10-- 分词的字典路径，支持多个11and `dic.paths`="....."12-- 停用词路径13and stopWordPath="..."14-- 高权重词路径15and priorityDicPath="...."16-- 高权重词加权倍数17and priority="5.0"18;19load parquet.`${traning_dir}/tfidf/data` 20as lwys_corpus_with_featurize;21-- 把label转化为递增数字22train lwys_corpus_with_featurize StringIndex.`${traning_dir}/si` 23where inputCol="label";24register StringIndex.`${traning_dir}/si` as predict;25select predict(label) as label,features as features from lwys_corpus_with_featurize 26as lwys_corpus_final_format;27-- 切分训练集、验证集，该算法会保证每个分类都是按比例切分。28train lwys_corpus_final_format as RateSampler.`${traning_dir}/ratesampler` 29where labelCol="label"30and sampleRate="0.9,0.1";31load parquet.`${traning_dir}/ratesampler` as data2;32select * from data2 where __split__=133as validateTable;34select * from data2 where __split__=035as trainingTable;36-- 训练，可以配置多个模型同时进行训练37train trainingTable as SKLearn.`${traning_dir}/model`  38where `kafkaParam.bootstrap.servers`="127.0.0.1:9092"39and `kafkaParam.topic`="test"40and `kafkaParam.group_id`="g_test-1"41and  `fitParam.0.batchSize`="300"42and  `fitParam.0.labelSize`="41"43and  `fitParam.0.alg`="RandomForestClassifier"44and  `fitParam.1.batchSize`="300"45and  `fitParam.1.labelSize`="41"46and  `fitParam.1.alg`="SVM"47and validateTable="validateTable"48and `systemParam.pythonPath`="python"49and `systemParam.pythonVer`="2.7"50;

训练完成后，如果我想在流式/批处理里用，那么应该是这样的：

 1-- 注册特征处理模型 2register TfIdfInPlace.`${traning_dir}/tfidf`  as  3tfidf_compute; 4register StringIndex.`${traning_dir}/si` as label_convert; 5-- 注册算法模型 6register SKLearn.`${traning_dir}/model`   as predict_label; 7-- 对数据进行特征处理 8select  tfidf_compute(content) as feature  from  some-hive-table 9as newdata;10--对数据进行预测：11select label_convert_reverse(predict_label(feature)) as predict_catogory   from  newdata;

如果我们希望部署成一个API服务，首先启动一个MLSQL服务：

 1./bin/spark-submit   --class streaming.core.StreamingApp \ 2--master local[2] \ 3--name predict_service \ 4streamingpro-spark-2.0-1.0.0.jar    \ 5-streaming.name predict_service    \ 6-streaming.job.file.path file:///tmp/query.json \ 7-streaming.platform spark   \ 8-streaming.rest true   \ 9-streaming.driver.port 9003   \10-streaming.spark.service true \11-streaming.thrift false \12-streaming.enableHiveSupport true

访问 http://127.0.0.1:9003/run/script 接口动态注册已经生成的模型：

1-- 注册特征处理模型2register TfIdfInPlace.`${traning_dir}/tfidf`  as 3tfidf_compute;4register StringIndex.`${traning_dir}/si` as label_convert;5-- 注册算法模型6register SKLearn.`${traning_dir}/model`   as predict_label;

MLSQL可以注册MLSQL自身的一些数据处理模型，对于算法模型，则包含了spark mllib, tensorflow, sklearn,dl4j等流行框架。

访问http://127.0.0.1:9003/model/predict进行预测请求：

1curl -XPOST 'http://127.0.0.1:9003/model/predict' -d '2data=["你好，世界"] & 3dataType=string &4pipeline= tfidf_compute, predict_label, label_convert5'

这个时候，会分别调用tfidf_compute, predict_label, label_convert 三个模型处理传递过来的数据，完成最后的预测。可以简单理解为一个嵌套函数调用(实际上就是，