本期课程内容：
在线黑名单过滤实现解析
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
本文出自 “叮咚” 博客，请务必保留此出处http://lqding.blog.51cto.com/9123978/1769290
第94课：SparkStreaming 实现广告计费系统中在线黑名单过滤实战
原文地址：http://lqding.blog.51cto.com/9123978/1769290
