前面的介绍见part1,part2
计划使用spark接收消息后调用一个服务,记录访问次数,再做一个页面显示访问次数。
具体使用Python的Flask提供服务,Spark调用服务增加访问次数,使用jQuery显示访问次数,
这个只是一个demo,具体实现后还是会找专业的人去做。
6.需要提供使用一个记录访问次数的服务
具体参考了以下文档(使用python的Flask实现一个RESTful API服务器端)创建了服务
具体步骤大致如下:
先安装虚拟环境
sudo easy_install virtualenv
设置虚拟环境
mkdir pythonproject
cd pythonprojectvirtualenv venv
进入应用环境
. venv/bin/activate
安装Flask
pip install Flask
根据参考文档改了一个flaskservice_simple.py
核心代码如下:
@app.route('/todo/api/v1.0/pv', methods=['GET'])
def get_pvdata():
global pvdata
return str(pvdata)
@app.route('/todo/api/v1.0/pv/<int:pvadd>', methods=['PUT'])
def update_pv(pvadd):
global pvdata
pvdata= pvdata + pvadd
return jsonify({'pvdata': pvdata})
具体的调用命令:
#!/bin/bash
cd /pythonproject
. venv/bin/activate
python ./flaskscript/flaskwebservice_simple.py
deactivate
可以使用如下命令测试
读取次数
curl -i -H "Content-Type: application/json" -X GET http://localhost:5000/todo/api/v1.0/pv
增加访问次数
curl -i -H "Content-Type: application/json" -X PUT http://localhost:5000/todo/api/v1.0/pv/5
---
7.做一个访问页面
使用jquery-1.11.3.js,放到Nginx下html目录
注意好像不能跨服务器调用,使用Nginx可以反向代理使用,这个我原来不知道,花了好长时间测试。
代码主要是这一段
$.get('todo/api/v1.0/pv', function(data){
//alert("Data Loaded: " +data);
$("#divMsg").html("BI portal visit: " +data);
});
不过我在onload设置了30秒刷新
<body onload="setInterval('clickButton()',30)">
最后好像浏览器有泄露,内存好用越来越大。
不过我只是测试,就先不管这个了。
8.SparkStream增加页面访问计数
在5的Spark代码上增加调用service就可以了, 我这里就偷懒使用os命令了
def echo(time, rdd):
counts_num = 'curl -i -H "Content-Type: application/json" -X PUT http://localhost:5000/todo/api/v1.0/pv/' + "%s" % (rdd.count())
counts = "Counts at time %s %s" % (time, rdd.count())
print(counts_num)
os.system(counts_num)
words.foreachRDD(echo)
到目前为止,系统的逻辑结构图