12.transform以及实时黑名单过滤案例实战

transform以及实时黑名单过滤案例实战

transform操作，应用在DStream上时，可以用于执行任意的RDD到RDD的转换操作。它可以用于实现，DStream API中所没有提供的操作。比如说，DStream API中，并没有提供将一个DStream中的每个batch，与一个特定的RDD进行join的操作。但是我们自己就可以使用transform操作来实现该功能。

DStream.join()，只能join其他DStream。在DStream每个batch的RDD计算出来之后，会去跟其他DStream的RDD进行join。

案例：广告计费日志实时黑名单过滤

java版本：

  1. package cn.spark.study.streaming;
  2.  
  3. import java.util.ArrayList;
  4. import java.util.List;
  5.  
  6. import org.apache.spark.SparkConf;
  7. import org.apache.spark.api.java.JavaPairRDD;
  8. import org.apache.spark.api.java.JavaRDD;
  9. import org.apache.spark.api.java.function.Function;
  10. import org.apache.spark.api.java.function.PairFunction;
  11. import org.apache.spark.streaming.Durations;
  12. import org.apache.spark.streaming.api.java.JavaDStream;
  13. import org.apache.spark.streaming.api.java.JavaPairDStream;
  14. import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
  15. import org.apache.spark.streaming.api.java.JavaStreamingContext;
  16.  
  17. import com.google.common.base.Optional;
  18.  
  19. import scala.Tuple2;
  20.  
  21. /**
  22.  * 基于transform的实时广告计费日志黑名单过滤
  23.  * 这里案例，完全脱胎于真实的广告业务的大数据系统，业务是真实的，实用
  24.  * @author Administrator
  25.  *
  26.  */
  27. public class TransformBlacklist {
  28. 	
  29. 	@SuppressWarnings("deprecation")
  30. 	public static void main(String[] args) {
  31. 		SparkConf conf = new SparkConf()
  32. 				.setMaster("local[2]")
  33. 				.setAppName("TransformBlacklist");  
  34. 		JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(5));
  35. 		
  36. 		// 用户对我们的网站上的广告可以进行点击
  37. 		// 点击之后，是不是要进行实时计费，点一下，算一次钱
  38. 		// 但是，对于那些帮助某些无良商家刷广告的人，那么我们有一个黑名单
  39. 		// 只要是黑名单中的用户点击的广告，我们就给过滤掉
  40. 		
  41. 		// 先做一份模拟的黑名单RDD
  42. 		List> blacklist = new ArrayList>();
  43. 		blacklist.add(new Tuple2("tom", true));  
  44. 		final JavaPairRDD blacklistRDD = jssc.sc().parallelizePairs(blacklist);
  45. 		
  46. 		// 这里的日志格式，就简化一下，就是date username的方式
  47. 		JavaReceiverInputDStream adsClickLogDStream = jssc.socketTextStream("spark1", 9999);
  48. 		
  49. 		// 所以，要先对输入的数据，进行一下转换操作，变成，(username, date username)
  50. 		// 以便于，后面对每个batch RDD，与定义好的黑名单RDD进行join操作
  51. 		JavaPairDStream userAdsClickLogDStream = adsClickLogDStream.mapToPair(
  52. 				
  53. 				new PairFunction() {
  54.  
  55. 					private static final long serialVersionUID = 1L;
  56.  
  57. 					@Override
  58. 					public Tuple2 call(String adsClickLog)
  59. 							throws Exception {
  60. 						return new Tuple2(
  61. 								adsClickLog.split(" ")[1], adsClickLog);
  62. 					}
  63. 					
  64. 				});
  65. 		
  66. 		// 然后，就可以执行transform操作了，将每个batch的RDD，与黑名单RDD进行join、filter、map等操作
  67. 		// 实时进行黑名单过滤
  68. 		JavaDStream validAdsClickLogDStream = userAdsClickLogDStream.transform(
  69. 				
  70. 				new Function, JavaRDD>() {
  71.  
  72. 					private static final long serialVersionUID = 1L;
  73.  
  74. 					@Override
  75. 					public JavaRDD call(JavaPairRDD userAdsClickLogRDD)
  76. 							throws Exception {
  77. 						// 这里为什么用左外连接？
  78. 						// 因为，并不是每个用户都存在于黑名单中的
  79. 						// 所以，如果直接用join，那么没有存在于黑名单中的数据，会无法join到
  80. 						// 就给丢弃掉了
  81. 						// 所以，这里用leftOuterJoin，就是说，哪怕一个user不在黑名单RDD中，没有join到
  82. 						// 也还是会被保存下来的
  83. 						JavaPairRDD>> joinedRDD = 
  84. 								userAdsClickLogRDD.leftOuterJoin(blacklistRDD);
  85. 						
  86. 						// 连接之后，执行filter算子
  87. 						JavaPairRDD>> filteredRDD = 
  88. 								joinedRDD.filter(
  89. 										
  90. 										new Function
  91. 												Tuple2>>, Boolean>() {
  92.  
  93. 											private static final long serialVersionUID = 1L;
  94.  
  95. 											@Override
  96. 											public Boolean call(
  97. 													Tuple2
  98. 															Tuple2>> tuple)
  99. 													throws Exception {
  100. 												// 这里的tuple，就是每个用户，对应的访问日志，和在黑名单中
  101. 												// 的状态
  102. 												if(tuple._2._2().isPresent() && 
  103. 														tuple._2._2.get()) {  
  104. 													return false;
  105. 												}
  106. 												return true;
  107. 											}
  108. 											
  109. 										});
  110. 						
  111. 						// 此时，filteredRDD中，就只剩下没有被黑名单过滤的用户点击了
  112. 						// 进行map操作，转换成我们想要的格式
  113. 						JavaRDD validAdsClickLogRDD = filteredRDD.map(
  114. 								
  115. 								new Function>>, String>() {
  116.  
  117. 									private static final long serialVersionUID = 1L;
  118.  
  119. 									@Override
  120. 									public String call(
  121. 											Tuple2>> tuple)
  122. 											throws Exception {
  123. 										return tuple._2._1;
  124. 									}
  125. 									
  126. 								});
  127. 						
  128. 						return validAdsClickLogRDD;
  129. 					}
  130. 					
  131. 				});
  132. 		
  133. 		// 打印有效的广告点击日志
  134. 		// 其实在真实企业场景中，这里后面就可以走写入kafka、ActiveMQ等这种中间件消息队列
  135. 		// 然后再开发一个专门的后台服务，作为广告计费服务，执行实时的广告计费，这里就是只拿到了有效的广告点击
  136. 		validAdsClickLogDStream.print();
  137. 		
  138. 		jssc.start();
  139. 		jssc.awaitTermination();
  140. 		jssc.close();
  141. 	}
  142. }

scala版本：

 

package cn.spark.study.streaming
 
import org.apache.spark.SparkConf
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.Seconds
 
/**
 * @author Administrator
 */
object TransformBlacklist {
  
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
        .setMaster("local[2]")  
        .setAppName("TransformBlacklist")
    val ssc = new StreamingContext(conf, Seconds(5))
    
    val blacklist = Array(("tom", true))  
    val blacklistRDD = ssc.sparkContext.parallelize(blacklist, 5)  
    
    val adsClickLogDStream = ssc.socketTextStream("spark1", 9999)   
    val userAdsClickLogDStream = adsClickLogDStream
        .map { adsClickLog => (adsClickLog.split(" ")(1), adsClickLog) }
    
    val validAdsClickLogDStream = userAdsClickLogDStream.transform(userAdsClickLogRDD => {
      val joinedRDD = userAdsClickLogRDD.leftOuterJoin(blacklistRDD)
      val filteredRDD = joinedRDD.filter(tuple => {
        if(tuple._2._2.getOrElse(false)) {  
          false
        } else {
          true
        }
      })
      val validAdsClickLogRDD = filteredRDD.map(tuple => tuple._2._1)
      validAdsClickLogRDD
    })
    
    validAdsClickLogDStream.print()
    
    ssc.start()
    ssc.awaitTermination()
  }
}

运行步骤：

 

1.启动nc

nc -lk 9999

2.本地运行，直接在IDE中运行程序

运行结果：

 

 

文章最后，给大家推荐一些受欢迎的技术博客链接：

  1. Hadoop相关技术博客链接
  2. Spark 核心技术链接
  3. JAVA相关的深度技术博客链接
  4. 超全干货--Flink思维导图，花了3周左右编写、校对
  5. 深入JAVA 的JVM核心原理解决线上各种故障【附案例】
  6. 请谈谈你对volatile的理解？--最近小李子与面试官的一场“硬核较量”
  7. 聊聊RPC通信，经常被问到的一道面试题。源码+笔记，包懂

 

欢迎扫描下方的二维码或 搜索 公众号“10点进修”，我们会有更多、且及时的资料推送给您，欢迎多多交流！

                                           

       

注：本文转载自blog.csdn.net的不埋雷的探长的文章"https://blog.csdn.net/weixin_32265569/article/details/78571497"。版权归原作者所有，此博客不拥有其著作权，亦不承担相应法律责任。如有侵权，请联系我们删除。
复制链接
相关推荐
发表评论

评论记录：

未查询到任何数据！