第94课：SparkStreaming 实现广告计费系统中在线黑名单过滤实战

本期课程内容：
在线黑名单过滤实现解析

SparkStreaming实现在线黑名单过滤

广告计费系统，是电商必不可少的一个功能点。为了防止恶意的广告点击(假设商户A和B同时在某电商做了广告，A和B为竞争对手，那么如果A使用点击机器人进行对B的广告的恶意点击，那么B的广告费用将很快被用完)，必须对广告点击进行黑名单过滤。
可以使用leftouter join 对目标数据和黑名单数据进行关联，将命中黑名单的数据过滤掉。
本文主要介绍的是DStream的transform函数的使用
SparkStreaming代码实现

package com.dt.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
* 使用Scala开发集群运行的Spark在线黑名单过滤
* @author dinglq
*         背景描述：在广告点击计费系统中，我们在线过滤掉黑名单的点击，进而保护广告商的利益，只进行有效的广告点击计费
*         或者在防刷评分（或者流量）系统，过滤掉无效的投票或者评分或者流量；
*         实现技术：使用transform Api直接基于RDD编程，进行join操作
*/
object OnlineBlackListFilter {
def main(args: Array[String]) {
/**
* 第1步：创建Spark的配置对象SparkConf，设置Spark程序的运行时的配置信息，
* 例如说通过setMaster来设置程序要链接的Spark集群的Master的URL,如果设置
* 为local，则代表Spark程序在本地运行，特别适合于机器配置条件非常差（例如
* 只有1G的内存）的初学者
*/
val conf = new SparkConf().setAppName("OnlineBlackListFilter") //设置应用名称
// 设置batch interval为30秒
val ssc = new StreamingContext(conf, Seconds(30))
/**
* 黑名单数据准备，实际上黑名单一般都是动态的，例如在Redis或者数据库中，黑名单的生成往往有复杂的业务
* 逻辑，具体情况算法不同，但是在Spark Streaming进行处理的时候每次都能工访问完整的信息
*/
// true表示是黑名单，如果需要暂时关闭，可以将值设置成false
val blackList = Array(("Hadoop", true), ("Mathou", true))
//把Array变成RDD
val blackListRDD = ssc.sparkContext.parallelize(blackList)

val adsClickStream = ssc.socketTextStream("spark-master", 9999)

/**
* 此处模拟的广告点击的每条数据的格式为：time、name
* 此处map操作的结果是name、（time，name）的格式
*/
val FormattedAdsClickStream = adsClickStream.map(item => (item.split(" ")(1), item))

val validateAds = FormattedAdsClickStream.transform(userClickRDD => {
/**
* 通过leftOuterJoin操作既保留了左侧用户广告点击内容的RDD的所有内容，
* 又获得了相应点击内容是否在黑名单中
*/

val joinedBlackListRDD = userClickRDD.leftOuterJoin(blackListRDD)

/**
* 进行filter过滤的时候，其输入元素是一个Tuple：（name,((time,name), boolean)）
* 其中第一个元素是黑名单的名称，第二元素的第二个元素是进行leftOuterJoin的时候是否存在在值
* 如果存在的话，表面当前广告点击是黑名单，需要过滤掉，否则的话则是有效点击内容；
*/
joinedBlackListRDD.filter(joinedItem =>
if (joinedItem._2._2.getOrElse(false)) {
true
} else {
false
}
)
})

validateAds.print()
ssc.start()
ssc.awaitTermination()
ssc.stop()
}
}

将程序打包，并上传至spark集群

在spark-master节点，启动nc

root@spark-master:~# nc -lk 9999

运行OnlineBlacklistFilter程序

root@spark-master:~# /usr/local/spark/spark-1.6.0-bin-hadoop2.6/bin/spark-submit --class com.dt.spark.streaming.OnlineBlackListFilter --master spark://spark-master:7077 ./spark.jar

在nc端输入数据

root@spark-master:~# nc -lk 9999
134343 Hadoop
343434 spark
3432777 Java
0983743 Hbase
893434 Mathou

SparkStreaming运行结果：

16/05/01 09:42:30 INFO scheduler.DAGScheduler: ResultStage 8 (print at OnlineBlackListFilter.scala:63) finished in 0.048 s
16/05/01 09:42:30 INFO scheduler.DAGScheduler: Job 3 finished: print at OnlineBlackListFilter.scala:63, took 0.111805 s
-------------------------------------------
Time: 1462066950000 ms
-------------------------------------------
3432777 Java
343434 spark
0983743 Hbase

可见，结果已经将黑名单设置的Hadoop和Mathou过滤掉了。

在此程序的基础上，可以添加更复杂的业务逻辑规则，以满足企业的需求。

备注：
1、DT大数据梦工厂微信公众号DT_Spark
2、IMF晚8点大数据实战YY直播频道号：68917580
3、新浪微博: http://www.weibo.com/ilovepains