前段时间,我们对接算法的工程师哭丧的和我说,模型生成后一般都要部署成API的形态对外提供服务,但是算法工程师并没有提供如何将一条数据转化特征向量的方法,他能拿到的是代码逻辑以及一些“中间元数据”。数据预处理本来就复杂,翻译也是一件极其困难的事情。我解释了这件事情难以解决的原因,但是显然他还是有些失望。
今天的目标就是谈谈如何尝试改善这件事情。
在我看来,之前业界已经给出解决方案了,就是pipeline,pipeline不仅仅包括数据特征化,还包括模型。Spark也是这么做的,但是其实应用并不广泛。原因有如下几个:
知道原因后,解决方案就变得相对直观了:
能做统一描述的其实是SQL,我们会将数据特征化流程也抽象成了算法训练,通过调参的方式干预特征处理,另外,我们尽可能的根据任务进行高级抽象而不是小功能点。这点很重要。比如:
-- 把文本字段转化为tf/idf向量,可以自定义词典
train orginal_text_corpus as TfIdfInPlace.`/tmp/tfidfinplace`
where inputCol="content"
-- 分词相关配置
and ignoreNature="true"
-- 停用词路径
and stopWordPath="/tmp/tfidf/stopwords"
-- 高权重词路径
and priorityDicPath="/tmp/tfidf/prioritywords"
-- 高权重词加权倍数
and priority="5.0"
and nGram="2,3"
;
我们将tf/idf特征化抽象成了一个算法,通过参数配置覆盖了如下的一些诉求:
如果是在训练阶段,我们直接加载模型的数据部分,从而让后续的算法可以继续进行处理。 使用方式如下:
load parquet.`/tmp/tfidfinplace/data`
as trainningdata;
我们也可能在其他的批处理或者流程序里去使用它预处理新的数据,这个时候我们需要 先注册一下:
register TfIdfInPlace.`/tmp/tfidfinplace` as tfidf;
然后通过UDF函数的方式去使用即可:
select tfidf(content) from hivetable;
如果你是部署成API服务,那么通过接口注册后,可以使用类似下面的http请求:
curl -XPOST 'http://127.0.0.1:9003/model/predict' -d '
data=["你好,世界"] &
dataType=string &
pipeline= tfidf,bayes
'
我这里额外添加了一个贝叶斯贝叶斯模型,这里pipeline的的调用相当于 bayes(tfidf("你好,世界"))
最后返回的是一个预测结果。
这个如何能做到呢?这就需要我们保存每个“数据处理模型”中间的元数据以及计算规则。比如以前面的TfIdfInPlace为例,他训练完成后会保存所有的训练参数,词空间,词和数字的映射等等。这样我们下次使用时就可以加载这些元数据,并且按特定的规则对新数据进行处理。
因为训练时的数据预处理和预测时的数据预处理本质是不同的,训练时的数据预处理只能针对批量数据,从中学习特征化的方式,而预测时的数据预处理更偏向于“利用训练时学到的经验仅仅进行计算”,这种天然不匹配带来的成本在于,你需要针对pipeline里的每个模型的预测部分(包括数据预处理和算法模型)进行重新的实现,而无法复用之前批训练时的逻辑。
对于MLSQL而言,它重新实现了大部分Spark mllib算法/数据处理模型的预测逻辑,增加了更多高阶的数据预处理模型,并且提供对tensorflow,sklearn,dl4j等框架的预测支持。
目前StreamingPro已经实现了一个案例,比如下面的代码通过SVM和随机深林实现了一个文本分类,特征工程用的是TfIdfInPlace算法:
set traning_dir = "/tmp/lwys_corpus";
-- 加载数据
load csv.`/Users/allwefantasy/Downloads/lwys_corpus` options header="true" and delimiter="\t" and quote="'"
as lwys_corpus;
select cut as features,cast(sid as int) as label from lwys_corpus
as orginal_text_corpus;
-- 把文本字段转化为tf/idf向量,可以自定义词典
train orginal_text_corpus as TfIdfInPlace.`${traning_dir}/tfidf`
where inputCol="features"
-- 分词的字典路径,支持多个
and `dic.paths`="....."
-- 停用词路径
and stopWordPath="..."
-- 高权重词路径
and priorityDicPath="...."
-- 高权重词加权倍数
and priority="5.0"
;
load parquet.`${traning_dir}/tfidf/data`
as lwys_corpus_with_featurize;
-- 把label转化为递增数字
train lwys_corpus_with_featurize StringIndex.`${traning_dir}/si`
where inputCol="label";
register StringIndex.`${traning_dir}/si` as predict;
select predict(label) as label,features as features from lwys_corpus_with_featurize
as lwys_corpus_final_format;
-- 切分训练集、验证集,该算法会保证每个分类都是按比例切分。
train lwys_corpus_final_format as RateSampler.`${traning_dir}/ratesampler`
where labelCol="label"
and sampleRate="0.9,0.1";
load parquet.`${traning_dir}/ratesampler` as data2;
select * from data2 where __split__=1
as validateTable;
select * from data2 where __split__=0
as trainingTable;
-- 训练,可以配置多个模型同时进行训练
train trainingTable as SKLearn.`${traning_dir}/model`
where `kafkaParam.bootstrap.servers`="127.0.0.1:9092"
and `kafkaParam.topic`="test"
and `kafkaParam.group_id`="g_test-1"
and `fitParam.0.batchSize`="300"
and `fitParam.0.labelSize`="41"
and `fitParam.0.alg`="RandomForestClassifier"
and `fitParam.1.batchSize`="300"
and `fitParam.1.labelSize`="41"
and `fitParam.1.alg`="SVM"
and validateTable="validateTable"
and `systemParam.pythonPath`="python"
and `systemParam.pythonVer`="2.7"
;
训练完成后,如果我想在流式/批处理里用,那么应该是这样的:
-- 注册特征处理模型
register TfIdfInPlace.`${traning_dir}/tfidf` as
tfidf_compute;
register StringIndex.`${traning_dir}/si` as label_convert;
-- 注册算法模型
register SKLearn.`${traning_dir}/model` as predict_label;
-- 对数据进行特征处理
select tfidf_compute(content) as feature from some-hive-table
as newdata;
--对数据进行预测:
select label_convert_reverse(predict_label(feature)) as predict_catogory from newdata;
如果我们希望部署成一个API服务,首先启动一个MLSQL服务:
./bin/spark-submit --class streaming.core.StreamingApp \
--master local[2] \
--name predict_service \
streamingpro-spark-2.0-1.0.0.jar \
-streaming.name predict_service \
-streaming.job.file.path file:///tmp/query.json \
-streaming.platform spark \
-streaming.rest true \
-streaming.driver.port 9003 \
-streaming.spark.service true \
-streaming.thrift false \
-streaming.enableHiveSupport true
访问 http://127.0.0.1:9003/run/script 接口动态注册已经生成的模型:
-- 注册特征处理模型
register TfIdfInPlace.`${traning_dir}/tfidf` as
tfidf_compute;
register StringIndex.`${traning_dir}/si` as label_convert;
-- 注册算法模型
register SKLearn.`${traning_dir}/model` as predict_label;
MLSQL可以注册MLSQL自身的一些数据处理模型,对于算法模型,则包含了spark mllib, tensorflow, sklearn,dl4j等流行框架。
访问http://127.0.0.1:9003/model/predict进行预测请求:
curl -XPOST 'http://127.0.0.1:9003/model/predict' -d '
data=["你好,世界"] &
dataType=string &
pipeline= tfidf_compute, predict_label, label_convert
'
这个时候,会分别调用tfidf_compute, predict_label, label_convert 三个模型处理传递过来的数据,完成最后的预测。可以简单理解为一个嵌套函数调用(实际上就是,?):
label_convert(predict_label(tfidf_compute(feature)))
通过上面描述的方式,我们可以有效的通过pipeline机制,在保证Online predict的响应时间的情况下,复用在训练时的数据预处理和算法模型,避免了重复开发,减少了研发的负担,并且基于统一的pipline dsl成功的无缝融合了数据处理框架和多个流行的机器学习框架。