在电商等线上业务场景中，需要对新生成的数据进行实时统计，尽快体现在当天的数据统计中，以便反应出当前时间点的数据变动情况。
1、添加依赖package com.demo.daystatisimport com.demo.PropertiesUtilimport com.demo.utils.{JdbcUtil, MyKafkaUtil}import org.apache.kafka.clients.consumer.ConsumerRecordimport org.apache.spark.SparkConfimport org.apache.spark.streaming.dstream.{DStream, InputDStream}import org.apache.spark.streaming.{Seconds, StreamingContext}object RealTimeApp { def main(args: Array[String]): Unit = { //1.创建 SparkConf val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RealTimeApp") //2.创建 StreamingContext val ssc = new StreamingContext(sparkConf, Seconds(3)) //3.读取 Kafka 数据 1583288137305 华南 深圳 4 3 val topic: String = PropertiesUtil.load("config.properties").getProperty("kafka.topic") val kafkaDStream: InputDStream[ConsumerRecord[String, String]] = MyKafkaUtil.getKafkaStream(topic, ssc) //4.将每一行数据转换为样例类对象 val adsLogDStream: DStream[Ads_log] = kafkaDStream.map(record => { //a.取出 value 并按照" "切分 val arr: Array[String] = record.value().split(" ") //b.封装为样例类对象 Ads_log(arr(0).toLong, arr(1), arr(2), arr(3), arr(4)) }) adsLogDStream.cache() //统计每天各大区各个城市广告点击总数并保存至 MySQL 中 DayStatisHandler.saveDateAreaCityAdCountToMysql(adsLogDStream) //开启任务 ssc.start() ssc.awaitTermination() }}
3、逻辑处理代码package com.demo.daystatisimport com.demo.utils.JdbcUtilimport java.sql.Connectionimport java.text.SimpleDateFormatimport java.util.Dateimport org.apache.spark.streaming.dstream.DStreamcase class Ads_log(timestamp: Long, area: String, city: String, userid: String, adid: String)object DayStatisHandler { //时间格式化对象 private val sdf: SimpleDateFormat = new SimpleDateFormat("yyyy-MM-dd") def saveDateAreaCityAdCountToMysql(filterAdsLogDStream: DStream[Ads_log]): Unit = { //1.统计每天各大区各个城市广告点击总数 val dateAreaCityAdToCount: DStream[((String, String, String, String), Long)] = filterAdsLogDStream.map(ads_log => { //a.取出时间戳 val timestamp: Long = ads_log.timestamp //b.格式化为日期字符串 val dt: String = sdf.format(new Date(timestamp)) //c.组合,返回 ((dt, ads_log.area, ads_log.city, ads_log.adid), 1L) }).reduceByKey(_ + _) //2.将单个批次统计之后的数据集合 MySQL 数据对原有的数据更新 dateAreaCityAdToCount.foreachRDD(rdd => { //对每个分区单独处理 rdd.foreachPartition(iter => { //a.获取连接 val connection: Connection = JdbcUtil.getConnection //b.写库 iter.foreach { case ((dt, area, city, adid), count) => JdbcUtil.executeUpdate(connection, """ |INSERT INTO area_city_ad_count (dt,area,city,adid,count) |VALUES(?,?,?,?,?) |ON DUPLICATE KEY |UPDATE count=count+?; """.stripMargin, Array(dt, area, city, adid, count, count)) } //c.释放连接 connection.close() }) }) }}
将rdd中的数据进行汇总后，在与mysql数据库中的数据进行比对操作，如果不存在，则进行添加，如果存在，则累计统计数据。 为了提高更新效率，这里使用了批量更新操作。
3、kafka消费者工具类package com.demo.utilsimport com.demo.PropertiesUtilimport org.apache.kafka.clients.consumer.ConsumerRecordimport org.apache.kafka.common.serialization.StringDeserializerimport org.apache.spark.streaming.StreamingContextimport org.apache.spark.streaming.dstream.InputDStreamimport org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}import java.util.Propertiesobject MyKafkaUtil { //1.创建配置信息对象 private val properties: Properties = PropertiesUtil.load("config.properties") //2.用于初始化链接到集群的地址 val broker_list: String = properties.getProperty("kafka.broker.list") //3.kafka 消费者配置 val kafkaParam = Map( "bootstrap.servers" -> broker_list, "key.deserializer" -> classOf[StringDeserializer], "value.deserializer" -> classOf[StringDeserializer], //消费者组 "group.id" -> "commerce-consumer-group", //如果没有初始化偏移量或者当前的偏移量不存在任何服务器上，可以使用这个配置属性 //可以使用这个配置，latest 自动重置偏移量为最新的偏移量 "auto.offset.reset" -> "latest", //如果是 true，则这个消费者的偏移量会在后台自动提交,但是 kafka 宕机容易丢失数据 //如果是 false，会需要手动维护 kafka 偏移量 "enable.auto.commit" -> (true: java.lang.Boolean) ) // 创建 DStream，返回接收到的输入数据 // LocationStrategies：根据给定的主题和集群地址创建 consumer // LocationStrategies.PreferConsistent：持续的在所有 Executor 之间分配分区 // ConsumerStrategies：选择如何在 Driver 和 Executor 上创建和配置 Kafka Consumer // ConsumerStrategies.Subscribe：订阅一系列主题 def getKafkaStream(topic: String, ssc: StreamingContext): InputDStream[ConsumerRecord[String, String]] = { val dStream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](ssc, LocationStrategies.PreferConsistent, ConsumerStrategies.Subscribe[String, String](Array(topic), kafkaParam)) dStream }}
4、mysql读写工具类package com.demo.utilsimport java.sql.{Connection, PreparedStatement, ResultSet}import java.util.Propertiesimport javax.sql.DataSourceimport com.alibaba.druid.pool.DruidDataSourceFactoryimport com.demo.PropertiesUtilobject JdbcUtil { //初始化连接池 var dataSource: DataSource = init() //初始化连接池方法 def init(): DataSource = { val properties = new Properties() val config: Properties = PropertiesUtil.load("config.properties") properties.setProperty("driverClassName", "com.mysql.jdbc.Driver") properties.setProperty("url", config.getProperty("jdbc.url")) properties.setProperty("username", config.getProperty("jdbc.user")) properties.setProperty("password", config.getProperty("jdbc.password")) properties.setProperty("maxActive", config.getProperty("jdbc.datasource.size")) DruidDataSourceFactory.createDataSource(properties) } //获取 MySQL 连接 def getConnection: Connection = { dataSource.getConnection } //执行 SQL 语句,单条数据插入 def executeUpdate(connection: Connection, sql: String, params: Array[Any]): Int = { var rtn = 0 var pstmt: PreparedStatement = null try { connection.setAutoCommit(false) pstmt = connection.prepareStatement(sql) if (params != null && params.length > 0) { for (i <- params.indices) { pstmt.setObject(i + 1, params(i)) } } rtn = pstmt.executeUpdate() connection.commit() pstmt.close() } catch { case e: Exception => e.printStackTrace() } rtn } //执行 SQL 语句,批量数据插入 def executeBatchUpdate(connection: Connection, sql: String, paramsList: Iterable[Array[Any]]): Array[Int] = { var rtn: Array[Int] = null var pstmt: PreparedStatement = null try { connection.setAutoCommit(false) pstmt = connection.prepareStatement(sql) for (params <- paramsList) { if (params != null && params.length > 0) { for (i <- params.indices) { pstmt.setObject(i + 1, params(i)) } pstmt.addBatch() } } rtn = pstmt.executeBatch() connection.commit() pstmt.close() } catch { case e: Exception => e.printStackTrace() } rtn } //判断一条数据是否存在 def isExist(connection: Connection, sql: String, params: Array[Any]): Boolean = { var flag: Boolean = false var pstmt: PreparedStatement = null try { pstmt = connection.prepareStatement(sql) for (i <- params.indices) { pstmt.setObject(i + 1, params(i)) } flag = pstmt.executeQuery().next() pstmt.close() } catch { case e: Exception => e.printStackTrace() } flag } //获取 MySQL 的一条数据 def getDataFromMysql(connection: Connection, sql: String, params: Array[Any]): Long = { var result: Long = 0L var pstmt: PreparedStatement = null try { pstmt = connection.prepareStatement(sql) for (i <- params.indices) { pstmt.setObject(i + 1, params(i)) } val resultSet: ResultSet = pstmt.executeQuery() while (resultSet.next()) { result = resultSet.getLong(1) } resultSet.close() pstmt.close() } catch { case e: Exception => e.printStackTrace() } result } //主方法,用于测试上述方法 def main(args: Array[String]): Unit = { }}
6、配置信息（config.properties）#jdbc 配置jdbc.datasource.size=10jdbc.url=jdbc:mysql://localhost:3306/spark2020?useUnicode=true&characterEncoding=utf8&rewriteBatchedStatements=truejdbc.user=usernamejdbc.password=password# Kafka 配置kafka.broker.list=localhost:9092kafka.topic=test
7、mysql建表脚本CREATE TABLE area_city_ad_count (dt VARCHAr(255),area VARCHAr(255),city VARCHAr(255),adid VARCHAr(255), count BIGINT,PRIMARY KEY (dt,area,city,adid));
8、通用工具类（PropertiesUtil.scala）package com.demoimport java.io.InputStreamReaderimport java.util.Propertiesobject PropertiesUtil { def load(propertiesName:String): Properties ={ val prop=new Properties() prop.load(new InputStreamReader(Thread.currentThread().getContextClassLoader.getResourceAsStream(propertiesName) , "UTF-8")) prop }}
9、执行测试在kafka队列中持续输入类似下面数据的情况下。
1645164248400 华东 上海 1 11645164248400 华东 上海 5 41645164248400 华北 北京 3 11645164248400 华北 天津 6 11645164248400 华南 深圳 4 31645164248400 华南 深圳 5 11645164248400 华北 北京 3 41645164248400 华北 北京 6 51645164248400 华北 北京 4 31645164248400 华东 上海 4 21645164248400 华北 北京 2 31645164248400 华南 深圳 3 5
可以在数据库表中，看到统计的数据。