大数据-100 Spark 集群 Spark Streaming DStream转换 黑名单过滤的三种实现 ...

喜大普奔！破百了！

点一下关注吧！！！非常感谢！！持续更新！！！

目前已经更新到了：

• Hadoop（已更完）
  
• HDFS（已更完）
  
• MapReduce（已更完）
  
• Hive（已更完）
  
• Flume（已更完）
  
• Sqoop（已更完）
  
• Zookeeper（已更完）
  
• HBase（已更完）
  
• Redis （已更完）
  
• Kafka（已更完）
  
• Spark（正在更新！）
  

章节内容

上节我们完成了如下的内容：

• Spark Streaming 底子数据源
  
• 文件流、Socket流、RDD队列流
  
• 引入依赖、Java编写多种流举行测试
  

DStream 转换

DStream上的操作与RDD类似，分为Transformations（转换）和 Output Operations（输出）两种，别的转换操作中还有一些比力特别的方法，如：

• updateStateByKey
  
• transform
  
• window干系操作
  

map(func)

对 DStream 中的每个元素应用 func 函数，并返回一个新的 DStream。
例如，将每个记载转换为其长度。
示例：val lengths = lines.map(line => line.length)
flatMap(func)

对 DStream 中的每个元素应用 func 函数，并将结果展平（即将集合的集合展开）。
例如，将每一行文本拆分为单词。
示例：val words = lines.flatMap(line => line.split(" "))
filter(func)

对 DStream 中的每个元素应用 func 函数，并保留返回值为 true 的元素。
例如，过滤掉长度小于 5 的单词。
示例：val filteredWords = words.filter(word => word.length > 5)
reduceByKey(func)

对键值对 DStream 举行聚合操作，对具有相同键的元素应用 func 函数。
例如，盘算每个单词的总数。
示例：val wordCounts = words.map(word => (word, 1)).reduceByKey(_ + _)
groupByKey()

对键值对 DStream 中的每个键举行分组，并将具有相同键的值聚合到一个列表中。
示例：val grouped = pairs.groupByKey()
count()

统计 DStream 中每个 RDD 的元素个数。
示例：val count = words.count()
countByValue()

统计 DStream 中每个 RDD 中每个值的出现次数。
示例：val valueCounts = words.countByValue()
union(otherDStream)

将两个 DStream 归并为一个新的 DStream，包罗两个 DStream 中的所有元素。
示例：val mergedStream = stream1.union(stream2)
join(otherDStream)

对两个键值对 DStream 举行毗连操作，类似 SQL 中的 JOIN 操作。
示例：val joinedStream = stream1.join(stream2)
备注：

• 在DStream与RDD上的转换操作非常类似（无状态操作）
  
• DStream有自己特别的操作（窗口操作、追踪状态变化操作）
  
• 在DStream上的转换操作比RDD上的转换操作少
  

DStream 的转换操作可以分为 无状态（stateless）和 有状态（stateful）两种：

• 无状态转换操作，每个批次的处理不依赖与之前批次的数据，常见的RDD转化操作，例如：map、Filter、reduceByKey等
  
• 有状态转换操作，需要使用之前批次的数据或者是中间结果来盘算当前批次的数据，有状态转换操作包括：基于滑动窗口的转换操作或追踪状态变化的转化操作
  

无状态转换

无状态转换操作就是把简单的RDD转换操作应用到每个批次上，也就是转换DStream中的每一个RDD。
常见的无状态转换包括：

• map
  
• flatMap
  
• repartition
  
• reduceByKey
  
• groupByKey
  

重要的转换操作：transform，通过对源DStream的每个RDD应用RDD-To-RDD函数，创建一个新的DStream，支持在新的DStream中任何RDD操作。
这是一个功能强大的函数，它可以允许开发者直接操作其内部的RDD，也就是说开发者，可以恣意提供一个RDDToRDD的函数，这个函数在数据流每个批次中都被调用，天生一个新的流。
案例1 黑名单过滤

1. 假设：arr1为黑名单数据(自定义)，true表示数据生效，需要被过滤掉；false表示数据
2. 未生效
3. val arr1 = Array(("spark", true), ("scala", false))
4. 假设：流式数据格式为"time word"，需要根据黑名单中的数据对流式数据执行过滤操
5. 作。如"2 spark"要被过滤掉
6. 1 hadoop
7. 2 spark
8. 3 scala
9. 4 java
10. 5 hive
11. 结果："2 spark" 被过滤

复制代码

方案1 外毗连实现

1. package icu.wzk
3. import org.apache.spark.SparkConf
4. import org.apache.spark.rdd.RDD
5. import org.apache.spark.streaming.dstream.ConstantInputDStream
6. import org.apache.spark.streaming.{Seconds, StreamingContext}
8. object BlackListFilter1 {
9.   def main(args: Array[String]): Unit = {
10.     val conf = new SparkConf()
11.       .setAppName("BlackListFilter1")
12.       .setMaster("local[*]")
13.     val ssc = new StreamingContext(conf, Seconds(10))
15.     // 黑名单
16.     val blackList = Array(("spark", true), ("scala", true))
17.     val blackListRDD = ssc.sparkContext.makeRDD(blackList)
19.     // 测试数据
20.     val strArray: Array[String] = "spark java scala hadoop kafka hive hbase zookeeper"
21.       .split("\\s+")
22.       .zipWithIndex
23.       .map {
24.         case (word, index) => s"$index $word"
25.       }
26.     val rdd = ssc.sparkContext.makeRDD(strArray)
27.     val clickStream = new ConstantInputDStream(ssc, rdd)
29.     // 流式数据的处理
30.     val clickStreamFormatted = clickStream
31.       .map(value => (value.split(" ")(1), value))
32.     clickStreamFormatted.transform(clickRDD => {
33.       val joinedBlockListRDD: RDD[(String, (String, Option[Boolean]))] = clickRDD.leftOuterJoin(blackListRDD)
34.       joinedBlockListRDD.filter {
35.         case (word, (streamingLine, flag)) => {
36.           if (flag.getOrElse(false)) {
37.             false
38.           } else {
39.             true
40.           }
41.         }
42.       }.map {
43.         case (word, (streamingLine, flag)) => streamingLine
44.       }
45.     }).print()
47.     // 启动
48.     ssc.start()
49.     ssc.awaitTermination()
50.   }
51. }

复制代码

方案1 运行结果

1. -------------------------------------------
2. Time: 1721618670000 ms
3. -------------------------------------------
4. 5 hive
5. 6 hbase
6. 1 java
7. 7 zookeeper
8. 3 hadoop
9. 4 kafka
11. ... 下一批

复制代码

对应的结果如下图所示：

方案2 SQL实现

1. package icu.wzk
3. import org.apache.spark.SparkConf
4. import org.apache.spark.sql.{DataFrame, SparkSession}
5. import org.apache.spark.streaming.dstream.ConstantInputDStream
6. import org.apache.spark.streaming.{Seconds, StreamingContext}
8. object BlackListFilter2 {
9.   def main(args: Array[String]): Unit = {
10.     val conf = new SparkConf()
11.       .setAppName("BlackListFilter2")
12.       .setMaster("local[*]")
13.     val ssc = new StreamingContext(conf, Seconds(10))
14.     ssc.sparkContext.setLogLevel("WARN")
16.     // 黑名单
17.     val blackList = Array(("spark", true), ("scala", true))
18.     val blackListRDD = ssc.sparkContext.makeRDD(blackList)
20.     // 生成测试 DStream
21.     val strArray: Array[String] = "spark java scala hadoop kafka hive hbase zookeeper"
22.       .split("\\s+")
23.       .zipWithIndex
24.       .map {
25.         case (word, index) => s"$index $word"
26.       }
27.     val rdd = ssc.sparkContext.makeRDD(strArray)
28.     val clickStream = new ConstantInputDStream(ssc, rdd)
30.     // 流式数据的处理
31.     val clickStreamFormatted = clickStream
32.       .map(value => (value.split(" ")(1), value))
33.     clickStreamFormatted.transform {
34.       clickRDD =>
35.         val spark = SparkSession
36.           .builder()
37.           .config(rdd.sparkContext.getConf)
38.           .getOrCreate()
40.         import spark.implicits._
41.         val clickDF: DataFrame = clickRDD.toDF("word", "line")
42.         val blackDF: DataFrame = blackListRDD.toDF("word", "flag")
43.         clickDF.join(blackDF, Seq("word"), "left")
44.           .filter("flag is null or flag == false")
45.           .select("line")
46.           .rdd
47.     }.print()
49.     ssc.start()
50.     ssc.awaitTermination()
51.   }
52. }

复制代码

方案2 SQL运行结果

1. -------------------------------------------
2. Time: 1721619900000 ms
3. -------------------------------------------
4. [6 hbase]
5. [4 kafka]
6. [7 zookeeper]
7. [1 java]
8. [3 hadoop]
9. [5 hive]

复制代码

运行结果截图如下图所示：

方案3 直接过滤

1. package icu.wzk
3. import org.apache.spark.SparkConf
4. import org.apache.spark.broadcast.Broadcast
5. import org.apache.spark.streaming.dstream.ConstantInputDStream
6. import org.apache.spark.streaming.{Seconds, StreamingContext}
8. object BlackListFilter3 {
10.   def main(args: Array[String]): Unit = {
11.     val conf = new SparkConf()
12.       .setAppName("BlackListFilter3")
13.       .setMaster("local[*]")
14.     val ssc = new StreamingContext(conf, Seconds(10))
15.     ssc.sparkContext.setLogLevel("WARN")
17.     // 黑名单
18.     val blackList = Array(("spark", true), ("scala", true))
19.     val blackListBC: Broadcast[Array[String]] = ssc
20.       .sparkContext
21.       .broadcast(blackList.filter(_._2).map(_._1))
23.     // 生成测试DStream
24.     val strArray: Array[String] = "spark java scala hadoop kafka hive hbase zookeeper"
25.       .split("\\s+")
26.       .zipWithIndex
27.       .map {
28.         case (word, index) => s"$index $word"
29.       }
31.     val rdd = ssc.sparkContext.makeRDD(strArray)
32.     val clickStream = new ConstantInputDStream(ssc, rdd)
34.     // 流式数据的处理
35.     clickStream.map(value => (value.split(" ")(1), value))
36.       .filter {
37.         case (word, _) => !blackListBC.value.contains(word)
38.       }
39.       .map(_._2)
40.       .print()
42.     // 启动
43.     ssc.start()
44.     ssc.awaitTermination()
45.    
46.   }
47. }

复制代码

方案3 直接过滤运行结果

1. -------------------------------------------
2. Time: 1721627600000 ms
3. -------------------------------------------
4. 1 java
5. 3 hadoop
6. 4 kafka
7. 5 hive
8. 6 hbase
9. 7 zookeeper
11. ... 下一批

复制代码

运行结果如下图所示：

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。