bloom_filter服务
bloom_filter的简介与使用请看这里。
bloom_filter的配置文件conf.ini如下
[filter]
cnt=4
num_bits=4294967296
num_hash_func=7
[server]
pid_file=/var/run/bloom_filter.pid
daemonlize=1
port=8000
cnt表示启动的bloom_filter的数量
num_bits表示位数组大小
cnt* num_bits/(8 * 1024 * 1024 * 1024)即为bloom_filter运行时所占用的内存大小(GB)。
数据插入
提供两种方式：
1、远程http请求。curl http://w21*********.8000/?ks=1,2 -X POST 。这种方式网络开销较大，数据量很大时不推荐。
2、本地文件导入。当启动bloom_filter服务时，会默认读取当前目录下名为new_users的文件（待改进），将每一行当作一条记录插入bloom_filter中。速度可达100,000条/秒。
数据查询
http请求，curl http://w21********:8000/?ks=1,2。
在w6机器上，qps可达4000/s。由于每个http请求中可以带多个参数，因此可以支持每秒上万个记录的查询任务。
storm计算新用户
任务：计算2014年1月1号的游戏级新用户。
启动bloom_filter服务。
在机器上启动bloom_filter服务，并将2014年1月1号前的所有游戏级新用户采用读取本地文件的方式导入bloom_filter中。
编写storm的bolt脚本。
程序从标准输入中，每读取30条记录，将就其拼接成一条curl请求，然后访问bloom_filter服务，根据返回值计算这里面有多少个新用户。
$filter_service = "http://w21*******:8000/?ks=";
$uv_count = 0;
$record_combine = array();
while(($line = fgets(STDIN)) !== false) {
$line = trim($line);
$record = explode("\t", $line);
$id = $record[0];
if(count($record)%3 !== 0) {
echo "$id\tack\n";
fwrite(STDERR, "wrong data".$line."\n");
flush();
continue;
}
for($i = 0; $i<count($record); $i+=3){
$qid = $record[$i+1];
$gkey = $record[$i+2];
$record_combine[] = $qid.$gkey;}
if(count($record_combine)===30){
$request = $filter_service;
$request .= implode(",", $record_combine);
$ch = curl_init();
curl_setopt($ch, CURLOPT_URL, $request);
curl_setopt($ch, CURLOPT_RETURNTRANSFER, 1);
$filter_result = curl_exec($ch);
curl_close($ch);
$uv_count +=substr_count($filter_result,"false");
fwrite(STDERR, "uv is ".$uv_count."\n");
unset($record_combine);
}
echo "$id\tack\n"; flush();
}
一次查询多条记录，能够大幅度减少网络压力与耗时。
提交storm任务到测试集群，并发送数据。
运行run.sh脚本提交任务。然后发送异步数据，
cat glogin_daily_2014_1_1.dat | async_send -batch 100 webgame_statistic_49 >/dev/null
这里采用 -batch指令，将日志每300行发送一次。
单个spout的发送数据的速度有限，为30~40万次/小时，若一次只发送一行日志，将大幅度降低storm的吞吐量
误判率
理论误判率计算公式在storm中使用bloom_filter消重服务。
m为位数组大小，k为哈希函数个数，n为记录数。
令m=235,k=7,n=240,000,000。
带入公式计算得到的理论误判率为：5.64e-10。
从程序中查询得到的程序理论误判率为：1.4e-4。
storm运行结果与线上数据比对后，得到程序实际误判率为：1.86e-3。
令m=236,k=7,n=240,000,000。
带入公式计算得到的理论误判率为：4.79e-12。
从程序中查询得到的程序理论误判率为：1.95e-6。
storm运行结果与线上数据比对后，得到程序实际误判率为：9.3e-4。
程序实际误判率比理论值大
结果比对
1、将storm任务运行时中bloom_fliter的返回结果（json）保存成日志
2、去掉json中的花括号{}。
cat log | awk '/"*"/ {print $0}' >process_log
3、找出返回结果为true的记录，并去掉引号""。
cat process_log | awk -F ":" '{gsub(/\"/, ""); if(match($2,/true/)) print $1}' >process_log_true
4、去掉记录的空格
sed 's/ //g' process_log_true > process_log_final
5、从hadoop中导出2014年1月1号的新用户的记录，并存成文本。
hive -e
"SELECT gp.qid, gp.gkey FROM
(SELECT distinct qid, gkey FROM glogin_daily WHERE year!=2014) np
RIGHT OUTER JOIN
(SELECT distinct qid, gkey FROM glogin_daily where year=2014 and month = 1 and day = 1) gp
ON np.qid = gp.qid AND np.gkey = gp.gkey
WHERE np.qid IS NOT NULL and np.gkey IS NOT NULL and gp.qid IS NOT NULL and gp.gkey IS NOT NULL;"
> exists_game_2014_01_01.dat
6、比对storm与hadoop中的记录。发现确实是bloom_filter误判了。
例如记录为"101326892zwj"，并没有在以往的日志中出现过，确实是新用户，但bloom_filter中却认为是已存在的用户，返回结果为true。
注意的问题
1、bloom_filter的误判率，与占用的总内存有关。就是说，存储2.4亿数据，使用4个512M的bloom_filter，与使用1个2G的bloom_filter的误判率是一样的。从公式也可以看出来。
2、storm集群若与bloom_filter的服务器跨机房，则会因为网络延迟，导致bolt中的数据没有被及时处理而堵塞。导致spout经常性失败 。
3、为防止iptable的设置导致bloom_filter不可用，可以开放8000端口。 sudo iptables -I INPUT -p tcp --dport 8000 -j ACCEPT
4、hadoop中的日志，不能只看分区，由于hydra任务的堆积，会使今天的部分日志记录到明天的分区中。因此需要指定year,month,day来导出日志
