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

-- 把文本字段转化为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"
;

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

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

 select tfidf(content) from hivetable;

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

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;

./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

-- 注册特征处理模型
register TfIdfInPlace.`${traning_dir}/tfidf`  as 
tfidf_compute;
register StringIndex.`${traning_dir}/si` as label_convert;

-- 注册算法模型
register SKLearn.`${traning_dir}/model`   as predict_label;

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

label_convert(predict_label(tfidf_compute(feature)))