使用SparkStreaming实现的三个实时需求

短命女 2022-10-24 13:29 151阅读 0赞

### 目录 ###

*  三个需求
 *   *  一、构建基础类
     *  二、需求一（广告点击量实时统计）
     *  三、需求二（最近一小时广告点击量）
     *  四、需求三（广告黑名单）

# 三个需求 #

*  需求一：广告点击量实时统计

> 描述：实时统计 每天 各地区 各城市 各广告 的点击总流量，并将其存入MySQL。

*  需求二：最近一小时广告点击量

> 最近一小时广告点击量, 结果展示：List \[15:50->10,15:51->25,15:52->30\]

*  需求三：广告黑名单

> 实现实时的动态黑名单机制：将每天对某个广告点击超过 100 次的用户拉黑。(黑名单保存到MySQL中)

## 一、构建基础类 ##

在基础类中封装三个需求都会用到的方法和属性。  
使用到的技术点：

1.  抽象类
2.  SparkStreaming应用上下文StreamingContext
3.  抽象控制
4.  构建流式应用的DStream
5.  TransFormat

package com.saprkstreaming.exec.app
    
    import com.saprkstreaming.exec.bean.AdsInfo
    import org.apache.kafka.clients.consumer.ConsumerRecord
    import org.apache.spark.streaming.dstream.{ DStream, InputDStream}
    import org.apache.spark.streaming.kafka010.{ ConsumerStrategies, KafkaUtils, LocationStrategies}
    import org.apache.spark.streaming.{ Seconds, StreamingContext}
    
    /** * BaseApp: 封装三个都需要使用的公共功能 * 每个需求都需要继承BaseApp */
    abstract class BaseApp { 
      
      //提供应用上下文
      //每三秒采集一批数据
      val streamingContext: StreamingContext = new StreamingContext("local[*]","test",Seconds(3))
      
      //使用抽象控制,允许每个需求将运行逻辑传入执行
      def runApp(op : => Unit):Unit={ 
        //每个需求自定义的逻辑
        op
        
        //启动应用
        streamingContext.start()
        //阻塞线程一直运行
        streamingContext.awaitTermination()
        
      }
      
      
      //设置采集kafka数据的相关参数 （消费者的参数名都可以从ConsumerConfig中获取）
      val kafkaParams:Map[String,String]=Map[String,String](
        "group.id"->"testApp",
        "bootstrap.servers"->"hadoop102:9092,hadoop103:9092",
        "enable.auto.commit"->"true",
        "auto.commit.interval.ms"->"500",
        "auto.offset.reset"->"earliest",
        "client.id"->"client1",
        "key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
        "value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer"
      )
      
      
      //构建DStream[AdsInfo]
      def getDstream() ={ 
        
        //从kafka中消费数据，获取DStream
        val ds: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream(
          streamingContext,
          LocationStrategies.PreferConsistent,
          ConsumerStrategies.Subscribe[String, String](List("mytest"), kafkaParams)
        )
        
        //获取DStream中的数据 1611907339612,华南,深圳,100,1
         val ds2: DStream[String] = ds.map(_.value())
      
         val result: DStream[AdsInfo] = ds2.map(adsinfo => { 
            val pros: Array[String] = adsinfo.split(",")
            AdsInfo(pros(0).toLong,
              pros(1),
              pros(2),
              pros(3),
              pros(4)
            )
          })
        result
        } 
    }

## 二、需求一（广告点击量实时统计） ##

使用到的技术点：

1.  数据采集，划窗时间
2.  updateStateByKey有状态操作
3.  写入数据库主键冲突问题
4.  数据tupe格式转换，获取不同的key类型

package com.saprkstreaming.exec.app
    
    import java.sql.{ Connection, PreparedStatement}
    
    import com.saprkstreaming.exec.bean.AdsInfo
    import com.saprkstreaming.exec.util.JDBCUtil
    import org.apache.spark.streaming.dstream.DStream
    
    /** * 需求二：广告点击量实时统计 * 描述：实时统计 每天 各地区 各城市 各广告 的点击总流量，并将其存入MySQL。 * * 分析： 统计的数据的时间范围是1天 * 统计的结果形式： ( ( 日期 , 地区, 城市 , 广告 ), 点击总流量 ) * 结果写入Mysql * * 数据是每3s采集一次， 3s计算一次 ， 总流量，累积操作，需要使用有状态的计算！ * updateStateByKey * * 如何将数据写入Mysql? * insert : 必须用insert,但是需要解决主键冲突！ * INSERT INTO area_city_ad_count VALUES('2021-01-22','a','b','1',310) * ON DUPLICATE KEY UPDATE COUNT= VALUES(COUNT) | 常量 * update * * * 计算逻辑： * AdsInfo => ( ( 日期 , 地区, 城市 , 广告 ), 1 ) * => ( ( 日期 , 地区, 城市 , 广告 ), N ) * */
    object Need2Demo extends BaseApp { 
      
      def main(args: Array[String]): Unit = { 
       
        runApp{ 
          //传入逻辑
          val ds: DStream[AdsInfo] = getDstream()
      
          //设置检查点目录
          streamingContext.checkpoint("exec")
      
          val result: DStream[((String, String, String, String), Int)] = ds.map(adsinfo => ((adsinfo.dayString, adsinfo.area, adsinfo.city, adsinfo.adsId), 1))
            .updateStateByKey((x: Seq[Int], y: Option[Int]) => Some(x.sum + y.getOrElse(0)))
          
          
          result.foreachRDD(rdd=>{ 
            rdd.foreachPartition(iter=>{ 
              //以分区为单位，每个分区共用一个connection对象
              val connection: Connection = JDBCUtil.getConnection()
              
              //准备SQL
              val sql =
                """ |insert into area_city_ad_count values(?,?,?,?,?) |on duplicate key update count=? |""".stripMargin
              
              //预编译：设置占位符
              val ps: PreparedStatement = connection.prepareStatement(sql)
              
              iter.foreach{ 
                case ((data,area,city,asid),count)=>{ 
                  ps.setString(1,data)
                  ps.setString(2,area)
                  ps.setString(3,city)
                  ps.setString(4,asid)
                  ps.setInt(5,count)
                  ps.setInt(6,count)
                  
                  //执行写入操作
                  ps.executeUpdate()
                }
                
              }
              //关闭资源
              ps.close()
              connection.close()
            })
            
            
          })
          
        }
      }
    }

## 三、需求二（最近一小时广告点击量） ##

使用到的技术点：

1.  步长、窗口
2.  开窗
3.  数据格式转换

package com.saprkstreaming.exec.app
    
    import com.saprkstreaming.exec.bean.AdsInfo
    import org.apache.spark.streaming.{ Duration, Minutes}
    import org.apache.spark.streaming.dstream.DStream
    
    /** * 需求三：最近一小时广告点击量 * 结果展示： * 1：List [15:50->10,15:51->25,15:52->30] * 2：List [15:50->10,15:51->25,15:52->30] * 3：List [15:50->10,15:51->25,15:52->30] * * 运算逻辑： AdsInfo => ((adsId,hmString) , 1) * => ((adsId,hmString) , N) * => （adsId ,(hmString,Count)） // 一个广告在一个分钟的点击数据 * => (adsId , { (hmString1,Count) , (hmString2,Count) .... } ) * */
    object Need3Demo extends BaseApp { 
      
      def main(args: Array[String]): Unit = { 
      
       
        runApp{ 
          val ds: DStream[AdsInfo] = getDstream()
      
          val result = ds.window(Minutes(60))
      
          val result2=     result.map(asinfo => ((asinfo.adsId, asinfo.hmString), 1))
            .reduceByKey(_ + _)
            .map { 
              case ((adsid, ht), count) => (adsid, (ht, count))
            }.groupByKey()
          
          
          result2.print(1000)
        
        }
        
      }
      
    }

## 四、需求三（广告黑名单） ##

使用到的技术点：

1.  数据保存到mysql
2.  主键冲突问题的解决
3.  有状态运算
4.  保存状态到mysql

package com.saprkstreaming.exec.app
    
    import java.sql.{ Connection, PreparedStatement, ResultSet}
    
    import com.saprkstreaming.exec.bean.AdsInfo
    import com.saprkstreaming.exec.util.JDBCUtil
    import org.apache.spark.streaming.dstream.DStream
    
    import scala.collection.mutable.ListBuffer
    
    /** 实现实时的动态黑名单机制： * 将每天对某个广告点击超过 100 次的用户拉黑,黑名单保存到MySQL中 * * 保存每天用户对每个广告的累积点击次数(有状态)： ((date,userid, adsid,) count) => user_ad_count * 保存拉黑的人： userid => black_list * * * 如果是有状态的计算，默认使用ck保存状态，会产生大量的小文件，不利于维护！ * 解决： 不使用默认的状态保存机制，自己保存状态(保存在mysql的 user_ad_count)！ * 在每次当前批次数据计算时，从 user_ad_count 读取历史状态，再和当前批次的数据进行累加，再将新的state写入mysql * * * * 业务流程： ①统计当前批次 用户对广告点击的总次数 * 统计当前批次的点击数据 ， 在写入mysql时，将当前数据和历史状态累加后再写入！ * * * AdsInfo => ((date,userid, adsid,) 1) * => ((date,userid, adsid,) N) * * 当前计算的结果： ((2021-1-4,105, 1,) 10) * mysql有的数据： ((2021-1-4,105, 1,) 20) * 写入时： ((2021-1-4,105, 1,) 10 + 20) * INSERT INTO user_ad_count VALUES(?,?,?,? ) * ON DUPLICATE KEY UPDATE COUNT=COUNT + ? * * ②从 user_ad_count 查询 所有 count > 100的用户 * * ③ 将这些用户，写入到 black_list * INSERT INTO black_list VALUES(? ) * ON DUPLICATE KEY UPDATE userid=? * * */
    object Need1Demo extends BaseApp { 
      
      def main(args: Array[String]): Unit = { 
        
        runApp{ 
          
          val ds: DStream[AdsInfo] = getDstream()
      
          //统计当前批次，用户的点击次数
          val ds1: DStream[((String, String, String), Int)] =
            ds.map(asdinfo => ((asdinfo.userId, asdinfo.adsId, asdinfo.dayString), 1))
            .reduceByKey(_ + _)
          
          //写入数据库
          ds1.foreachRDD(rdd=>{ 
            //按分区写入数据，减少connection连接数量
            rdd.foreachPartition(iter=>{ 
              val connection: Connection = JDBCUtil.getConnection()
              //准备SQL
              val sql=
                """ |insert into user_ad_count values(?,?,?,?) |on duplicate key update count=count+? |""".stripMargin
              
              //预编译sql，设置占位符
              val st: PreparedStatement = connection.prepareStatement(sql)
              
              iter.foreach{ 
                case((userid,adsid,day),count)=>{ 
                  st.setString(1,day)
                  st.setString(2,userid)
                  st.setString(3,adsid)
                  st.setInt(4,count)
                  st.setInt(5,count)
                  
                  st.executeUpdate()
                }
              }
              st.close()
              
              //从user_ad_count查询所有count>100的用户
              val sql2=
                """ |select userid from user_ad_count where count>100 |""".stripMargin
              val ps2: PreparedStatement = connection.prepareStatement(sql2)
              val resultSet: ResultSet = ps2.executeQuery()
              
              //获取所有点击数大于100的用户
              val needUser:ListBuffer[String]=ListBuffer()
              while (resultSet.next()){ 
                needUser.append(resultSet.getString("userid"))
                //println(resultSet.getString("userid"))
              }
              resultSet.close()
              ps2.close()
              
              
              //将这些用户写入black_list
              val sql3=
                """ |insert into black_list values(?) |on duplicate key update userid = ? |""".stripMargin
              val ps3: PreparedStatement = connection.prepareStatement(sql3)
              
             needUser.foreach(userid=>{ 
                ps3.setString(1,userid)
                ps3.setString(2,userid)
               
               ps3.executeUpdate()
              })
              ps3.close()
              
              connection.close()
          })})
          
        }
        
      }
    }