189 8069 5689
本期课程内容：
创新互联是一家专业提供望花企业网站建设,专注与成都做网站、成都网站设计、HTML5建站、小程序制作等业务。10年已为望花众多企业、政府机构等服务。创新互联专业网站建设公司优惠进行中。
在线黑名单过滤实现解析
SparkStreaming实现在线黑名单过滤
广告计费系统，是电商必不可少的一个功能点。为了防止恶意的广告点击(假设商户A和B同时在某电商做了广告，A和B为竞争对手，那么如果A使用点击机器人进行对B的广告的恶意点击，那么B的广告费用将很快被用完)，必须对广告点击进行黑名单过滤。
可以使用leftOuterJoin对目标数据和黑名单数据进行关联，将命中黑名单的数据过滤掉。
本文主要介绍的是DStream的transform函数的使用
SparkStreaming代码实现
package com.dt.spark.sparkapps.streaming
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
/**
* 使用Scala开发集群运行的Spark 在线黑名单过滤程序
* Created by Limaoran on 2016/5/2.
* 新浪微博：http://weibo.com/ilovepains/
*
* 背景描述：在广告点击计费系统中，我们在线过滤掉黑名单的点击，进而保护广告商的利益，
* 只进行有效的广告点击计费或者在防刷评分（或者流量）系统，过滤掉无效的投票或者评分或者流量；
* 实现技术：使用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() //创建SparkConf对象
conf.setAppName("OnlineBlackListFilter") //设置应用程序的名称，在程序运行的监控界面可以看到名称
conf.setMaster("spark://Master:7077") //此时，程序在Spark集群
val ssc = new StreamingContext(conf,Seconds(30))
/**
* 黑名单数据准备，实际上黑名单一般都是动态的，例如在redis或者数据库中，黑名单的生成往往有复杂的业务
* 逻辑，具体情况算法不同，但是在Spark Streaming进行处理的时候每次都能工访问完整的信息
*/
val blackList = Array(("hadoop",true),("mahout",true))
val blackListRDD = ssc.sparkContext.parallelize(blackList,8)
val adsClickStream = ssc.socketTextStream("Master" ,9999)
/**
* 此处模拟的广告点击的每条数据的格式为：time、name
* 此处map操作的结果是name、（time，name）的格式
*/
val adsClientStreamFormated = adsClickStream.map(ads=>(ads.split(" ")(1),ads))
adsClientStreamFormated.transform(userClickRDD => {
//通过leftOuterJoin操作既保留了左侧用户广告点击内容的RDD的所有内容，又获得了相应点击内容是否在黑名单中
val joinedBlackListRDD = userClickRDD.leftOuterJoin(blackListRDD)
/**
* 进行filter过滤的时候，其输入元素是一个Tuple：（name,((time,name), boolean)）
* 其中第一个元素是黑名单的名称，第二元素的第二个元素是进行leftOuterJoin的时候是否存在在值
* 如果存在的话，表面当前广告点击是黑名单，需要过滤掉，否则的话则是有效点击内容；
*/
val validClicked = joinedBlackListRDD.filter(joinedItem=>{
if(joinedItem._2._2.getOrElse(false)){
false
}else{
true
}
})
validClicked.map(validClick => {validClick._2._1})
}).print()
/**
* 计算后的有效数据一般都会写入Kafka中，下游的计费系统会从kafka中pull到有效数据进行计费
*/
ssc.start()
ssc.awaitTermination()
}
}将程序打包，并上传至spark集群
在spark-master节点，启动nc
root@spark-master:~# nc -lk 9999
运行OnlineBlacklistFilter程序
root@spark-master:~# /usr/local/spark-1.6.0/bin/spark-submit --class com.dt.spark.sparkapps.streaming.OnlineBlackListFilter --master spark://Master:7077 ./sparkApps.jar
在nc端输入数据
root@spark-master:~# nc -lk 9999 22555 spark 124321 hadoop 5555 Flink 6666 HDFS 2222 Kafka 572231 Java 66662 mahout
SparkStreaming运行结果：
16/05/02 08:28:00 INFO MapPartitionsRDD: Removing RDD 8 from persistence list ------------------------------------------- 5555 Flink 6666 HDFS 572231 Java 22555 spark 2222 Kafka
可见，结果已经将黑名单设置的hadoop和mathou过滤掉了。
在此程序的基础上，可以添加更复杂的业务逻辑规则，以满足企业的需求。
备注：
1、DT大数据梦工厂微信公众号DT_Spark
2、IMF晚8点大数据实战YY直播频道号：68917580
3、新浪微博: http://www.weibo.com/ilovepains