继 无编码利用协同算法实现个性化推荐,我纯粹使用SQL和配置实现了一个更为复杂一些的,计算文章词汇的tf/idf值,将浏览数作为预测值,使用线性回归算法进行模型训练的示例。帮助大家更好的了解StreamingPro对算法的优秀支持。这篇文章的示例将会跑在Spark 2.0 上了。为了方便大家体验,我已经将Spark 安装包,StreamignPro,以及分词包都准备好,大家下载即可。
准备工作
- 下载Spark 2.0,基于Scala 2.10版本
- 下载StreamingPro
- 下载分词包 ansj-seg
我们假设你下载的StreamingPro,ansi-seg包在/tmp目录下。然后将Spark 2.0 解压,进入主目录。
复制如下模板
我已经发布了三个配置文件,分别计算:
- 词汇的 idf 值 ,链接
- 给每个词汇生成一个唯一的数字标示,链接
- 使用线性回归算法进行训练, 链接
PS : 有道笔记有时候第一次刷不出来,你刷新下就好。
复制保存三个文件:
- /tmp/idf.json
- /tmp/term-index.json
- /tmp/lr-train.json
本机运行
生成idf 文件:
代码语言:javascript复制cd $SPARK_HOME
./bin/spark-submit --class streaming.core.StreamingApp
--master local[2]
--name test
--jars /tmp/ansj_seg-5.0.0-all-in-one.jar
/tmp/streamingpro-0.3.3-SNAPSHOT-online-mllib-2.0.0.jar
-streaming.name test
-streaming.platform spark
-streaming.job.file.path file:///tmp/idf.json
生成内容会存储成Parquet文件。在/tmp/idf 目录下可以看到具体文件。
接着生成 term index ,文件位于 /tmp/term-with-index,最后进行模型训练,训练好的模型在/tmp/lr-model
后续只要在Spark Streaming中加载,即可完成流式计算。
配置文件简要说明
以lr-train.json为例,大体框架如下:
代码语言:javascript复制{
"lr1": {
"desc": "LR模型训练Job",
"strategy": "streaming.core.strategy.SparkStreamingStrategy",
"compositor": [ ]
},
"udf_register": {
"desc": "通过这个方式可以注册你自己开发的udf函数",
"strategy": "streaming.core.strategy.SparkStreamingRefStrategy",
"compositor": [ ]
},
"term_index_ref_table": {
"desc": "在这里申明表,可以在job中被引用",
"strategy": "streaming.core.strategy.SparkStreamingRefStrategy",
"algorithm": [],
"ref": [],
"compositor": [ ]
},
"term_idf_ref_table": {
"desc": "在这里申明表,可以在job中被引用",
"strategy": "streaming.core.strategy.SparkStreamingRefStrategy",
"algorithm": [],
"ref": [],
"compositor": [ ]
}
}
这里有一个job,两个关联表,一个UDF函数注册模块。我在配置文件的描述中已经有说明。job 是一个可执行的main函数,你可以这么理解。关联表申明后可以直接在job的sql中使用。UDF函数注册模块则可以使得你很容易扩展SQL的功能。
他们唯一的区别是,Job 的strategy 是 SparkStreamingStrategy
,而其他非Job则是SparkStreamingRefStrategy
。
因为一个配置文件里可能有多个Job,每个Job引用的关联表也是不一样,你需要显示指定引用,在Job 的ref中申明即可:
代码语言:javascript复制 "lr1": {
"strategy": "streaming.core.strategy.SparkStreamingStrategy",
"ref": [
"udf_register",
"term_index_ref_table",
"term_idf_ref_table"
],
"compositor": [
这样框架自动为你准备好关联引用,注册UDF函数,然后在lr1 这个job中就可以使用了。比如lr里的parse 函数就是通过udf_register模块提供的。
之后就是定义输入,执行的SQL,以及输出(存储或者模型引擎)。 SQL在案例中你可以看到,可以非常复杂,多个SQL模块之间可以互相作用,通过多条SQL实现一个复杂的逻辑。比如我们这里试下了tf/idf计算等功能。