第一章.项目需求一:日活统计

1.创建子模块(gmall-realtime)

该模块为实时处理模块,主要负责对采集到的数据进行实时处理

一.pom.xml

  1. <dependencies>
  3.         <dependency>
  4.             <groupId>org.apache.spark</groupId>
  5.             <artifactId>spark-core_2.12</artifactId>
  6.         </dependency>
  8.         <dependency>
  9.             <groupId>org.apache.spark</groupId>
  10.             <artifactId>spark-streaming_2.12</artifactId>
  11.         </dependency>
  13.         <dependency>
  14.             <groupId>org.apache.spark</groupId>
  15.             <artifactId>spark-streaming-kafka-0-10_2.12</artifactId>
  16.         </dependency>
  18.         <!-- 使用java连接redis  -->
  19.         <dependency>
  20.             <groupId>redis.clients</groupId>
  21.             <artifactId>jedis</artifactId>
  22.             <version>2.9.0</version>
  23.         </dependency>
  25.         <dependency>
  26.             <artifactId>gmall-common</artifactId>
  27.             <groupId>org.example</groupId>
  28.             <version>1.0-SNAPSHOT</version>
  29.         </dependency>
  31.         <dependency>
  32.             <groupId>org.apache.phoenix</groupId>
  33.             <artifactId>phoenix-spark</artifactId>
  34.             <version>5.0.0-HBase-2.0</version>
  35.             <exclusions>
  36.                 <exclusion>
  37.                     <groupId>org.glassfish</groupId>
  38.                     <artifactId>javax.el</artifactId>
  39.                 </exclusion>
  40.             </exclusions>
  41.         </dependency>
  44.         <dependency>
  45.             <groupId>org.apache.spark</groupId>
  46.             <artifactId>spark-sql_2.12</artifactId>
  47.         </dependency>
  49.         <dependency>
  50.             <groupId>commons-httpclient</groupId>
  51.             <artifactId>commons-httpclient</artifactId>
  52.             <version>3.1</version>
  53.         </dependency>
  56.     </dependencies>
  58.     <build>
  59.         <plugins>
  60.             <!-- 该插件用于将Scala代码编译成class文件 -->
  61.             <plugin>
  62.                 <groupId>net.alchim31.maven</groupId>
  63.                 <artifactId>scala-maven-plugin</artifactId>
  64.                 <version>3.2.0</version>
  65.                 <executions>
  66.                     <execution>
  67.                         <!-- 声明绑定到maven的compile阶段 -->
  68.                         <goals>
  69.                             <goal>compile</goal>
  70.                             <goal>testCompile</goal>
  71.                         </goals>
  72.                     </execution>
  73.                 </executions>
  74.             </plugin>
  75.         </plugins>
  76.     </build>

二.配置文件

  1. config.properties
  1. # Kafka配置
  2. kafka.broker.list=hadoop102:9092,hadoop103:9092,hadoop104:9092
  3. # Redis配置
  4. redis.host=hadoop102
  5. redis.port=6379
  1. log4j.properties
  1. #定义了名为atguigu.MyConsole的一个ConsoleAppender 向控制台输出日志，红色打印，
  2. log4j.appender.atguigu.MyConsole=org.apache.log4j.ConsoleAppender
  3. log4j.appender.atguigu.MyConsole.target=System.err
  4. log4j.appender.atguigu.MyConsole.layout=org.apache.log4j.PatternLayout
  6. # 年-月-日 时:分:秒  10个占位符  日志级别  (全类名:方法名) - 消息 换行
  7. log4j.appender.atguigu.MyConsole.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss} %10p (%c:%M) - %m%n
  9. #代表指定哪个类的logger使用什么级别和appender进行日志输出  全类名可能需要修改
  10. log4j.rootLogger=error,atguigu.MyConsole

三.工具类

  • 消费kafka中的数据
  • 利用Redis过滤当日已经计入的日活设备
  • 把每批次新增的当日的日活信息保存在Hbase中
  • 从HBase中查询出数据,发布成数据接口,通过可视化工程调用
  1. PropertiesUtil.scala
  1. package com.atguigu.realtime.utils
  3. import java.io.InputStreamReader
  4. import java.util.Properties
  6. // 加载类路径的一个properties文件，只需要调用load方法，自动将properties文件读取，封装一个Properties返回
  7. object PropertiesUtil {
  9.   def load(propertieName:String): Properties ={
  10.     val prop=new Properties()
  11.     prop.load(new InputStreamReader(Thread.currentThread().getContextClassLoader.getResourceAsStream(propertieName) , "UTF-8"))
  12.     prop
  13.   }
  15. }
  1. MyKafkaUtil.scala
  1. package com.atguigu.realtime.utils
  3. import java.util.Properties
  4. import org.apache.kafka.clients.consumer.ConsumerRecord
  5. import org.apache.kafka.common.TopicPartition
  6. import org.apache.kafka.common.serialization.StringDeserializer
  7. import org.apache.spark.streaming.StreamingContext
  8. import org.apache.spark.streaming.dstream.InputDStream
  9. import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
  11. object MyKafkaUtil {
  13.   // 读取配置文件中的信息
  14.   private val properties: Properties = PropertiesUtil.load("config.properties")
  16.   // kafka集群地址
  17.   val broker_list: String = properties.getProperty("kafka.broker.list")
  19.   def getKafkaStream(topics: Array[String], ssc: StreamingContext, groupId:String, saveOffsetToMysql:Boolean = false ,
  20.                      offsetsMap: Map[TopicPartition, Long] =null , ifAutoCommit:String ="false" ): InputDStream[ConsumerRecord[String, String]] = {
  22.     //kafka消费者配置
  23.     val kafkaParam = Map(
  24.       "bootstrap.servers" -> broker_list,
  25.       "key.deserializer" -> classOf[StringDeserializer],
  26.       "value.deserializer" -> classOf[StringDeserializer],
  27.       "group.id" -> groupId,
  28.       "auto.offset.reset" -> "earliest",
  29.       "enable.auto.commit" -> ifAutoCommit
  30.     )
  32.     var ds: InputDStream[ConsumerRecord[String, String]] =null;
  34.     if (saveOffsetToMysql){
  36.       ds = KafkaUtils.createDirectStream[String, String](
  37.         ssc,
  38.         LocationStrategies.PreferConsistent,
  39.         ConsumerStrategies.Subscribe[String, String](topics, kafkaParam,offsetsMap)
  40.       )
  41.     }else{
  43.       ds = KafkaUtils.createDirectStream[String, String](
  44.         ssc,
  45.         LocationStrategies.PreferConsistent,
  46.         ConsumerStrategies.Subscribe[String, String](topics, kafkaParam))
  47.     }
  49.     ds
  50.   }
  51. }
  1. RedisUtil
  1. package com.atguigu.realtime.utils
  3. import java.util.Properties
  4. import redis.clients.jedis.{Jedis}
  6. object RedisUtil {
  8.   val config: Properties = PropertiesUtil.load("config.properties")
  9.   val host: String = config.getProperty("redis.host")
  10.   val port: String = config.getProperty("redis.port")
  12.   def getJedisClient():Jedis={
  14.     new Jedis(host,port.toInt)
  16.   }
  17. }

四.DAU实现

  1. 流程图

Spark实时数仓(二) - 图1

  1. 设计redis的key,value | key | value | | —- | —- | | 当前批次日期(logDate) | 设备号(mid) |
  1. Phoenix建表
  1. create table gmall2021_startuplog
  2. (
  3.               ar varchar,
  4.               ba varchar,
  5.               ch varchar,
  6.               is_new Integer,
  7.               md varchar,
  8.               mid varchar,
  9.               os varchar,
  10.               uid BIGINT,
  11.               vc varchar,
  12.               entry varchar,
  13.               loading_time Integer,
  14.               open_ad_id Integer,
  15.               open_ad_ms Integer,
  16.               open_ad_skip_ms Integer,
  17.               logDate varchar,
  18.               logHour varchar,
  19.               ts BIGINT
  20.               CONSTRAINT dau_pk PRIMARY KEY (mid, logDate));
  1. 为了便于对数据进行解析,需要将数据封装为样例类(StartUpLog,StartLogoInfo)
  1. package com.atguigu.realtime.beans
  3. case class StartUpLog(
  4.                        //common
  5.                        var ar:String,
  6.                        var ba:String,
  7.                        var  ch:String,
  8.                        var is_new:Int,
  9.                        var md:String,
  10.                        var mid:String,
  11.                        var os:String,
  12.                        var  uid:Long,
  13.                        var   vc:String,
  14.                        // start
  15.                        var entry:String,
  16.                        var loading_time:Int,
  17.                        var open_ad_id:Int,
  18.                        var open_ad_ms:Int,
  19.                        var open_ad_skip_ms:Int,
  20.                        //kafka里面没有，为了统计最终结果，额外设置的字段，需要从ts转换得到
  21.                        var logDate:String,
  22.                        var logHour:String,
  24.                        var ts:Long){
  27.   def mergeStartInfo(startInfo:StartLogInfo):Unit={
  29.     if (startInfo != null){
  31.       this.entry = startInfo.entry
  32.       this.loading_time = startInfo.loading_time
  33.       this.open_ad_id = startInfo.open_ad_id
  34.       this.open_ad_ms = startInfo.open_ad_ms
  35.       this.open_ad_skip_ms = startInfo.open_ad_skip_ms
  37.     }
  38.   }
  39. }
  41. // 封装了 启动日志中 start:{}部分的属性，目的是为了方便展平
  42. case class StartLogInfo(
  43.                          entry:String,
  44.                          loading_time:Int,
  45.                          open_ad_id:Int,
  46.                          open_ad_ms:Int,
  47.                          open_ad_skip_ms:Int
  48.                        )
  1. 将每个需求实现的公共部分封装为一个抽象类(BaseApp)
  1. package com.atguigu.realtime.app
  3. import org.apache.spark.streaming.StreamingContext
  4. //该类是所有实时App程序继承的父类
  6. abstract class BaseApp {
  7.   //抽象属性(只声明未赋值),让子类实现的
  8.   var appName:String
  9.   var batchDuration:Int
  10.   var ssc:StreamingContext = null
  11.   def runApp(op: =>Unit){
  12.     try{
  13.       //一堆代码
  14.       op
  15.       ssc.start()
  16.       ssc.awaitTermination()
  17.     }catch{
  18.       case e:Exception =>
  19.         e.printStackTrace()
  20.          throw new RuntimeException("运行失败")
  21.     }
  22.   }
  24. }
  1. 日活统计程序(DAUApp)
  1. package com.atguigu.realtime.app
  3. import com.alibaba.fastjson.JSON
  4. import com.atguigu.constants.MyConstants
  5. import com.atguigu.realtime.beans.{StartLogInfo, StartUpLog}
  6. import com.atguigu.realtime.utils.{MyKafkaUtil, RedisUtil}
  7. import org.apache.hadoop.hbase.HBaseConfiguration
  8. import org.apache.kafka.clients.consumer.ConsumerRecord
  9. import org.apache.spark.SparkConf
  10. import org.apache.spark.rdd.RDD
  11. import org.apache.spark.streaming.kafka010.{CanCommitOffsets, HasOffsetRanges}
  12. import org.apache.spark.streaming.{Seconds, StreamingContext}
  14. import java.time.{Instant, LocalDateTime, ZoneId}
  15. import java.time.format.DateTimeFormatter
  16. import scala.Seq
  17. //导入Phoenix提供的静态方法
  18. import org.apache.phoenix.spark._
  21. /**
  22.  * 精确一次性:  at least once(offsets维护到kafka) + hbase(幂等性输出)
  23.  */
  25. object DAUApp extends BaseApp {
  26.   override var appName: String = "DAUApp"
  27.   override var batchDuration: Int = 10
  28.   val groupId = "gmallrealtime"
  29.   /**
  30.    * 将rdd封装为样例类
  31.    * @param rdd
  32.    * @return
  33.    */
  34.   def parseRecordToBean(rdd: RDD[ConsumerRecord[String, String]]):RDD[StartUpLog] = {
  35.     rdd.map(record => {
  36.       //将record.value()封装为json对象
  37.       val jSONObject = JSON.parseObject(record.value())
  38.       //封装common部分压平的结果
  39.       val startUpLog = JSON.parseObject(jSONObject.getString("common"), classOf[StartUpLog])
  40.       //封装start部分并合并
  41.       val startLogInfo = JSON.parseObject(jSONObject.getString("start"), classOf[StartLogInfo])
  42.       startUpLog.mergeStartInfo(startLogInfo)
  43.       //合并ts
  44.       startUpLog.ts = jSONObject.getLong("ts")
  45.       //获取当前日志的日期和小时
  46.       val formatter1 = DateTimeFormatter.ofPattern("yyyy-MM-dd")
  47.       val formatter2 = DateTimeFormatter.ofPattern("HH")
  48.       val time = LocalDateTime.ofInstant(Instant.ofEpochMilli(startUpLog.ts), ZoneId.of("Asia/Shanghai"))
  49.       startUpLog.logDate = time.format(formatter1)
  50.       startUpLog.logHour = time.format(formatter2)
  51.       startUpLog
  53.     })
  55.   }
  57.   /**
  58.    * 同批次去重(根据当前批次的日期和设备号去重)
  59.    * @param rdd
  60.    * @return
  61.    */
  62.   def removeDuplicateLogsInCommonBatch(rdd: RDD[StartUpLog]):RDD[StartUpLog] = {
  63.     val rdd1 = rdd.map(log => {
  64.       ((log.logDate, log.mid), log)
  66.     })
  67.     val rdd2 = rdd1.groupByKey()
  68.     //只对value进行运算,可以使用mapValues
  69.     val rdd3 = rdd2.mapValues({
  70.       _.toList.sortBy(_.ts).take(1)
  71.     })
  72.     val rdd4 = rdd3.map(date => {
  73.       ((date._1, date._2.head))
  74.     })
  75.     rdd4.values
  76.   }
  78.   /**
  79.    * 跨批次过滤
  80.    * @param rdd
  81.    * @return
  82.    */
  83.   def removeDuplicateLogsInDiffBatch(rdd: RDD[StartUpLog]):RDD[StartUpLog] = {
  84.     //连接数据库,都是以分区为单位获取连接
  85.     rdd.mapPartitions(partition=>{
  86.       //获取连接
  87.       val jedis = RedisUtil.getJedisClient()
  88.       //判断这条日志的mid和logdate是否在set集合中已经存在
  89.       val filterdPartition = partition.filter(log => !jedis.sismember("DAU:" + log.logDate, log.mid))
  90.       //关闭
  91.       jedis.close()
  92.       //返回处理后的分区
  93.       filterdPartition
  94.     })
  96.   }
  98.   def main(args: Array[String]): Unit = {
  99.     ssc = new StreamingContext(new SparkConf().setMaster("local[4]").setAppName(appName),Seconds(batchDuration))
  100.     runApp{
  101.       //1.获取初始ds
  102.       val ds = MyKafkaUtil.getKafkaStream(Array(MyConstants.STARTUP_LOG),ssc,
  103.         groupId)
  104.       ds.foreachRDD(rdd => {
  105.         //真正从kafka消费到数据后,再执行以下运算,防止job每次数据提交,但是并无数据计算!
  106.         if (!rdd.isEmpty()){
  107.           //2.获取偏移量
  108.           val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
  109.           //3.封装为样例类
  110.           val rdd1 = parseRecordToBean(rdd)
  111.           println("当前这批数据消费到:" + rdd1.count())
  112.           //4.同批次去重
  113.           val rdd2 = removeDuplicateLogsInCommonBatch(rdd1)
  114.           println("同批次去重后: " + rdd2.count())
  115.           //5.跨批次过滤
  116.           val rdd3 = removeDuplicateLogsInDiffBatch(rdd2)
  117.           rdd3.cache()
  118.           println("跨批次过滤后: " + rdd3.count())
  119.           //6.写入hbase
  120.           rdd3.saveToPhoenix("GMALL2021_STARTUPLOG",
  121.             Seq("AR", "BA", "CH", "IS_NEW", "MD", "MID", "OS", "UID", "VC", "ENTRY", "LOADING_TIME","OPEN_AD_ID","OPEN_AD_MS","OPEN_AD_SKIP_MS","LOGDATE","LOGHOUR","TS"),
  122.             HBaseConfiguration.create(),
  123.             Some("hadoop102:2181"))
  125.           //7.将写入hbase完成的mid,logdate 记录到redis
  126.           rdd3.foreachPartition(partition=>{
  127.             val jedis = RedisUtil.getJedisClient()
  128.             partition.foreach(log=>{
  129.               jedis.sadd("DAU:"+log.logDate,log.mid)
  130.               //设置key的过期时间,设置24小时过期
  131.               jedis.expire("DAU:"+log.logDate,60*60*24)
  132.             })
  133.             jedis.close()
  134.           })
  135.           //8.提交偏移量
  136.           ds.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)
  137.         }
  138.       })
  139.     }
  140.   }
  141. }
  1. 将数据写入数据库并测试

分别启动gmall-logger,nginx,zookeeper,kafka,hadoop,hbase,redis,phoenix

生产数据到kafka()

Spark实时数仓(二) - 图2

启动APP

Spark实时数仓(二) - 图3

分别在redis和Phoenix客户端验证

Spark实时数仓(二) - 图4

Spark实时数仓(二) - 图5

2.创建子模块(gmall-publisher)

该模块为日活数据查询接口,是一个springboot工程,主要负责对实时处理后的数据进行查询

一.pom.xml

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  3.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
  4.     <modelVersion>4.0.0</modelVersion>
  5.     <parent>
  6.         <artifactId>spark-gmall</artifactId>
  7.         <groupId>org.example</groupId>
  8.         <version>1.0-SNAPSHOT</version>
  9.     </parent>
  11.     <artifactId>gmall-publisher</artifactId>
  12.     <version>0.0.1-SNAPSHOT</version>
  13.     <name>gmall-publisher</name>
  14.     <description>Demo project for Spring Boot</description>
  15.     <properties>
  16.         <java.version>1.8</java.version>
  17.     </properties>
  18.     <dependencies>
  19.         <dependency>
  20.             <groupId>org.springframework.boot</groupId>
  21.             <artifactId>spring-boot-starter-web</artifactId>
  22.         </dependency>
  23.         <dependency>
  24.             <groupId>org.mybatis.spring.boot</groupId>
  25.             <artifactId>mybatis-spring-boot-starter</artifactId>
  26.             <version>2.2.0</version>
  27.         </dependency>
  29.         <dependency>
  30.             <groupId>org.projectlombok</groupId>
  31.             <artifactId>lombok</artifactId>
  32.             <optional>true</optional>
  33.         </dependency>
  34.         <dependency>
  35.             <groupId>org.springframework.boot</groupId>
  36.             <artifactId>spring-boot-starter-test</artifactId>
  37.             <scope>test</scope>
  38.         </dependency>
  39.         <dependency>
  40.             <artifactId>gmall-common</artifactId>
  41.             <groupId>org.example</groupId>
  42.             <version>1.0-SNAPSHOT</version>
  43.         </dependency>
  44.         <dependency>
  45.             <groupId>org.apache.phoenix</groupId>
  46.             <artifactId>phoenix-core</artifactId>
  47.             <version>5.0.0-HBase-2.0</version>
  48.             <exclusions>
  49.                 <exclusion>
  50.                     <groupId>org.glassfish</groupId>
  51.                     <artifactId>javax.el</artifactId>
  52.                 </exclusion>
  53.             </exclusions>
  54.         </dependency>
  56.         <!-- google提供的java开发的工具包，在很多框架中，都引入了guava-->
  57.         <dependency>
  58.             <groupId>com.google.guava</groupId>
  59.             <artifactId>guava</artifactId>
  60.             <version>20.0</version>
  61.         </dependency>
  63.         <!--java.lang.NoSuchMethodError: org.apache.hadoop.security.authentication.util.KerberosUtil.hasKerberosKeyTab-->
  64.         <dependency>
  65.             <groupId>org.apache.hadoop</groupId>
  66.             <artifactId>hadoop-common</artifactId>
  67.             <version>2.7.2</version>
  68.         </dependency>
  69.         <!-- 异构(不同架构 mysql hbase)数据源的切换 -->
  70.         <dependency>
  71.             <groupId>com.baomidou</groupId>
  72.             <artifactId>dynamic-datasource-spring-boot-starter</artifactId>
  73.             <version>2.4.2</version>
  74.         </dependency>
  76.         <dependency>
  77.             <groupId>mysql</groupId>
  78.             <artifactId>mysql-connector-java</artifactId>
  79.         </dependency>
  80.     </dependencies>
  82.     <build>
  83.         <plugins>
  84.             <plugin>
  85.                 <groupId>org.springframework.boot</groupId>
  86.                 <artifactId>spring-boot-maven-plugin</artifactId>
  87.                 <version>1.5.10.RELEASE</version>
  88.                 <configuration>
  89.                     <excludes>
  90.                         <exclude>
  91.                             <groupId>org.projectlombok</groupId>
  92.                             <artifactId>lombok</artifactId>
  93.                         </exclude>
  94.                     </excludes>
  95.                 </configuration>
  96.             </plugin>
  97.         </plugins>
  98.     </build>
  100. </project>

二.配置文件

  1. application.properties
  1. #绑定8070端口，没写项目名  ，使用http://主机名:8070/
  2. server.port=8070
  4. #默认使用springboot提供的logger，记录级别为error
  5. logging.level.root=error
  7. #mybatis
  8. mybatis.mapperLocations=classpath:mybatis/*.xml
  9. mybatis.configuration.map-underscore-to-camel-case=true
  1. application.yml
  1. spring:
  2.   datasource:
  3.     dynamic:
  4.       primary: hbase #设置默认的数据源或者数据源组,默认值即为master
  5.       strict: false #严格匹配数据源,默认false. true未匹配到指定数据源时抛异常,false使用默认数据源
  6.       datasource:
  7.         hbase:
  8.           url: jdbc:phoenix:hadoop102,hadoop103,hadoop104:2181
  9.           username:
  10.           password:
  11.           driver-class-name: org.apache.phoenix.jdbc.PhoenixDriver
  12.         mysql:
  13.           url: jdbc:mysql://hadoop103:3306/sparkproduct
  14.           username: root
  15.           password: "321074"
  16.           driver-class-name: com.mysql.jdbc.Driver

三.业务代码实现

  1. pojo层(DAUPerHour)
  1. package com.example.gmallpublisher.pojo;
  3. import lombok.AllArgsConstructor;
  4. import lombok.Data;
  5. import lombok.NoArgsConstructor;
  7. @NoArgsConstructor
  8. @AllArgsConstructor
  9. @Data
  10. public class DAUPerHour {
  11.     private String hour;
  12.     private Integer dau;
  13. }
  1. mapper层(DAUMapper)
  1. package com.example.gmallpublisher.mapper;
  3. import com.baomidou.dynamic.datasource.annotation.DS;
  4. import com.example.gmallpublisher.pojo.DAUPerHour;
  5. import org.springframework.stereotype.Repository;
  7. import java.util.List;
  8. @Repository
  9. @DS("hbase")
  10. public interface DAUMapper {
  11.     //查询单日日活
  12.     Integer getDAUByDate(String date);
  13.     //查询当日新增的设备数
  14.     Integer getNewMidCountByDate(String date);
  15.     //查询当日分时的设备数
  16.     List<DAUPerHour> getDAUPerHourData(String date);
  17. }
  1. DAUMapper.xml
  1. <?xml version="1.0" encoding="UTF-8" ?>
  2. <!DOCTYPE mapper
  3.         PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
  4.         "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
  5. <!--namespace=绑定一个对应的Dao/Mapper接口-->
  6. <mapper namespace="com.example.gmallpublisher.mapper.DAUMapper">
  7. <!--查询当日日活-->
  8.     <select id="getDAUByDate" resultType="int">
  9.         select
  10.             count(*)
  11.            from GMALL2021_STARTUPLOG
  12.         where logdate = #{logdate}
  13.     </select>
  14. <!-- 查询当日新增的设备数-->
  15.     <select id="getNewMidCountByDate" resultType="int">
  16.         select count(*)
  17.         from
  18.         (select mid
  19.         from GMALL2021_STARTUPLOG
  20.         where logdate = #{logdate}) t1
  21.         left join
  22.         (select mid
  23.         from GMALL2021_STARTUPLOG
  24.         where logdate &lt; #{logdate}) t2
  25.         on t1.mid = t2.mid
  26.         where t2.mid is null
  27.     </select>
  28. <!-- 查询当日分时的设备数   -->
  29.     <select id="getDAUPerHourData" resultType="com.example.gmallpublisher.pojo.DAUPerHour">
  30.         select
  31.           loghour hour,
  32.           count(*) dau
  33.         from GMALL2021_STARTUPLOG
  34.         where logdate = #{logdate}
  35.         group by loghour
  36.     </select>
  39. </mapper>
  1. service层(PublisherService,PublisherServiceImpl)
  1. package com.example.gmallpublisher.service;
  3. import com.example.gmallpublisher.pojo.DAUPerHour;
  4. import com.example.gmallpublisher.pojo.GMVPerHour;
  6. import java.util.List;
  8. public interface PublisherService {
  9.     //查询单日日活
  10.     Integer getDAUByDate(String date);
  11.     //查询当日新增的设备数
  12.     Integer getNewMidCountByDate(String date);
  13.     //查询当日分时的设备数
  14.     List<DAUPerHour> getDAUPerHourData(String date);
  16. }
  1. package com.example.gmallpublisher.service;
  3. import com.example.gmallpublisher.mapper.DAUMapper;
  4. import com.example.gmallpublisher.mapper.GMVMapper;
  5. import com.example.gmallpublisher.pojo.DAUPerHour;
  8. import com.example.gmallpublisher.pojo.GMVPerHour;
  9. import org.springframework.beans.factory.annotation.Autowired;
  10. import org.springframework.stereotype.Service;
  13. import java.util.List;
  15. @Service
  16. public class PublisherServiceImpl implements PublisherService {
  17.     @Autowired
  18.     private DAUMapper dauMapper;
  20.     @Override
  21.     public Integer getDAUByDate(String date) {
  22.         System.out.println("遵循的业务步骤");
  23.         return dauMapper.getDAUByDate(date);
  24.     }
  26.     @Override
  27.     public Integer getNewMidCountByDate(String date) {
  28.         System.out.println("遵循的业务步骤");
  29.         return dauMapper.getNewMidCountByDate(date);
  30.     }
  32.     @Override
  33.     public List<DAUPerHour> getDAUPerHourData(String date) {
  34.         System.out.println("遵循业务步骤");
  35.         return dauMapper.getDAUPerHourData(date);
  36.     }
  39. }
  1. controller层
  1. package com.example.gmallpublisher.controller;
  3. import com.alibaba.fastjson.JSONObject;
  4. import com.example.gmallpublisher.pojo.DAUPerHour;
  5. import com.example.gmallpublisher.pojo.GMVPerHour;
  6. import com.example.gmallpublisher.service.PublisherService;
  7. import org.springframework.beans.factory.annotation.Autowired;
  8. import org.springframework.web.bind.annotation.RequestMapping;
  9. import org.springframework.web.bind.annotation.RestController;
  11. import java.time.LocalDate;
  12. import java.util.ArrayList;
  13. import java.util.List;
  15. @RestController
  16. public class PublisherController {
  17.     @Autowired
  18.     private PublisherService publisherService;
  20.     /**
  21.      * [
  22.      *    {"id":"dau","name":"当日日活数","value":1200},
  23.      *    {"id":"new_mid","name":"新增设备数","value":233}
  24.      *    
  25.      *  ]
  26.      * @param date
  27.      * @return
  28.      */
  29.     @RequestMapping(value="/realtime-total")
  30.     public Object handle1(String date){
  31.         //获取当日日活
  32.         Integer dauByDate = publisherService.getDAUByDate(date);
  33.         //获取当日新增设备数
  34.         Integer newMidCountByDate = publisherService.getNewMidCountByDate(date);
  36.         ArrayList<JSONObject> result = new ArrayList<>();
  37.         JSONObject jsonObject1 = new JSONObject();
  38.         JSONObject jsonObject2 = new JSONObject();
  39.         /*JSONObject jsonObject3 = new JSONObject();*/
  40.         jsonObject1.put("id","dau");
  41.         jsonObject1.put("name","当日日活数");
  42.         jsonObject1.put("value",dauByDate);
  43.         jsonObject2.put("id","new_mid");
  44.         jsonObject2.put("name","新增设备数");
  45.         jsonObject2.put("value",newMidCountByDate);
  46.         result.add(jsonObject1);
  47.         result.add(jsonObject2);
  48.         /*result.add(jsonObject3);*/
  51.         return result;
  54.     }
  56.     /**
  57.      *
  58.      *   {"yesterday":
  59.            {
  60.               "11":383,
  61.               "12":123,
  62.               "17":88,
  63.               "19":200
  64.          },
  65.      "today":
  66.             {"12":38,"13":1233,"17":123,"19":688 }
  67.      }
  68.      *
  69.      * @param date
  70.      * @return
  71.      */
  73.     @RequestMapping(value="realtime-hours")
  74.     public Object handle2(String id,String date){
  75.         JSONObject result = new JSONObject();
  76.         //根据今天的日期计算昨天的日期
  77.         String yestodayDate = LocalDate.parse(date).minusDays(1).toString();
  79.         if("dau".equals(id)){
  80.             List<DAUPerHour> todayData = publisherService.getDAUPerHourData(date);
  81.             List<DAUPerHour> yestodayData = publisherService.getDAUPerHourData(yestodayDate);
  82.             result.put("yesterday",parseDAUDate(yestodayData));
  83.             result.put("today",parseDAUDate(todayData));
  84.         }
  87.         return result;
  89.     }
  94.     public JSONObject parseDAUDate(List<DAUPerHour> data){
  95.         JSONObject result = new JSONObject();
  96.         for (DAUPerHour dauPerHour : data) {
  97.             result.put(dauPerHour.getHour(),dauPerHour.getDau());
  98.         }
  99.         return result;
  100.     }
  101. }
  1. 主程序
  1. package com.example.gmallpublisher;
  3. import org.mybatis.spring.annotation.MapperScan;
  4. import org.springframework.boot.SpringApplication;
  5. import org.springframework.boot.autoconfigure.SpringBootApplication;
  7. @SpringBootApplication
  8. @MapperScan(basePackages = "com.example.gmallpublisher.mapper")
  9. public class GmallPublisherApplication {
  11.     public static void main(String[] args) {
  12.         SpringApplication.run(GmallPublisherApplication.class, args);
  13.     }
  15. }
  1. index.html
  1. <!DOCTYPE html>
  2. <html lang="en">
  3. <head>
  4.     <meta charset="UTF-8">
  5.     <title>首页</title>
  6. </head>
  7. <body>
  8. <a href=" http://localhost:8070/realtime-total?date=2021-10-07">请求今天DAU</a> <br>
  9. <a href=" http://localhost:8070/realtime-hours?id=dau&date=2021-10-07">请求今天和昨天分时DAU</a>
  10. <br/>
  14. </body>
  15. </html>

四.运行程序

浏览器访问: http://localhost:8070/index.html

Spark实时数仓(二) - 图6

  1. 请求今天DAU

Spark实时数仓(二) - 图7

  1. 请求今天和昨天分时DAU

Spark实时数仓(二) - 图8

第二章.项目需求二:交易额统计

数据处理流程图

Spark实时数仓(二) - 图9

1.业务采集数据

交易额统计需要采集业务数据,业务数据一般存储在关系型数据库mysql中,此时可以使用Canal来实时采集业务数据到kafka

Canal的工作原理:将自己伪装成Slave(从数据库),假装从Master(主数据库)复制数据

一.MySql的准备

  1. 使用存储过程模拟数据
  1. call `init_data`('2021-10-06',10,3,FALSE);
  2. call `init_data`('2021-10-07',10,3,FALSE);
  1. 开启Binlog

MYSQL需要让主机把写操作记录到日志中,故需要开启binlog

在安装mysql的主机(hadoop103)上编辑配置文件(/etc/my.cnf),在[mysqld]下面添加

  1. log-bin=mysql-bin
  2. server-id=3
  3. binlog_format=row
  4. binlog-do-db=sparkproduct

修改完之后重启mysql

  1. sudo systemctl restart mysqld

二.安装Canal

  1. 下载Canal

下载地址: https://github.com/alibaba/canal/releases

  1. 将压缩包上传到Linux上并解压

注意: 解压前自己要先创建一个目录来存放解压后的Canal数据

  1. 修改canal.properties(全局配置文件)
  1. vim conf/canal.properties
  2. canal.ip = hadoop103
  1. 修改instance.properties (实例配置文件)
  1. vim conf/example/instance.properties
  2. canal.instance.mysql.slaveId=0
  3. # position info
  4. canal.instance.master.address=hadoop103:3306
  5. canal.instance.master.journal.name=mysql-bin.000001
  6. canal.instance.master.position=154
  1. 在mysql中执行以下语句(赋权限)
  1. GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%' IDENTIFIED BY 'canal';
  2. FLUSH PRIVILEGES;
  1. 启动canal
  1. [atguigu@hadoop103 canal]$ bin/startup.sh

2.创建子模块(gmall-canalclient)

该模块为canal客户端,订阅canalServer拉取到的数据

一.pom.xml

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <project xmlns="http://maven.apache.org/POM/4.0.0"
  3.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  4.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  5.     <parent>
  6.         <artifactId>spark-gmall</artifactId>
  7.         <groupId>org.example</groupId>
  8.         <version>1.0-SNAPSHOT</version>
  9.     </parent>
  10.     <modelVersion>4.0.0</modelVersion>
  12.     <artifactId>gmall-canalclient</artifactId>
  14.     <properties>
  15.         <maven.compiler.source>8</maven.compiler.source>
  16.         <maven.compiler.target>8</maven.compiler.target>
  17.     </properties>
  18.     <dependencies>
  19.         <dependency>
  20.             <artifactId>gmall-common</artifactId>
  21.             <groupId>org.example</groupId>
  22.             <version>1.0-SNAPSHOT</version>
  23.         </dependency>
  24.         <!--canal client-->
  25.         <dependency>
  26.             <groupId>com.alibaba.otter</groupId>
  27.             <artifactId>canal.client</artifactId>
  28.             <version>1.1.2</version>
  29.         </dependency>
  31.         <!--kafka : 需要使用生产者写入kafka-->
  32.         <dependency>
  33.             <groupId>org.apache.kafka</groupId>
  34.             <artifactId>kafka-clients</artifactId>
  35.         </dependency>
  36.     </dependencies>
  38. </project>

二.生产者(MyProducer.java)

  1. package com.atguigu.client;
  3. import org.apache.kafka.clients.producer.KafkaProducer;
  4. import org.apache.kafka.clients.producer.Producer;
  5. import org.apache.kafka.clients.producer.ProducerRecord;
  7. import java.util.Properties;
  9. public class MyProducer {
  12.     private static Producer producer;
  13.     static{
  14.          producer= getProducer();
  15.     }
  18.     //获取一个生产者
  20.     public static Producer getProducer(){
  21.         Properties properties = new Properties();
  22.         //集群地址 key,value的序列化器
  23.         properties.put("bootstrap.servers","hadoop102:9092,hadoop103:9092,hadoop104:9092");
  24.         properties.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
  25.         properties.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");
  26.         return new KafkaProducer<String,String>(properties);
  27.     }
  29.     public static void sendData(String topic,String value){
  30.         producer.send(new ProducerRecord(topic,value));
  31.     }
  32. }

三.canal客户端(CanalClient.java)

  1. package com.atguigu.client;
  3. import com.alibaba.fastjson.JSONObject;
  4. import com.alibaba.otter.canal.client.CanalConnector;
  5. import com.alibaba.otter.canal.client.CanalConnectors;
  6. import com.alibaba.otter.canal.protocol.CanalEntry;
  7. import com.alibaba.otter.canal.protocol.Message;
  8. import com.atguigu.constants.MyConstants;
  9. import com.google.protobuf.ByteString;
  10. import com.google.protobuf.InvalidProtocolBufferException;
  12. import java.net.InetSocketAddress;
  13. import java.util.List;
  15. //canalclient连接服务端
  16. public class CanalClient {
  17.     public static void main(String[] args) throws InterruptedException, InvalidProtocolBufferException {
  18.         //1.创建一个客户端对象
  19.         CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("hadoop103", 11111), "example", null, null);
  20.         //2.连接
  21.         connector.connect();
  22.         //3.发请求
  23.         connector.subscribe("sparkproduct.order_info");
  24.         //4.拉数据
  25.         while(true){
  26.             Message message = connector.get(100);
  27.             //判断是否拉取到了数据
  28.             if(message.getId() == -1){
  29.                 System.out.println("歇会再去");
  30.                 Thread.sleep(5000);
  31.                 continue;
  32.             }
  33.             //解析数据(message->拉取一次,多条SQL,造成的写操作变化)
  34.             //List<CanalEntry.Entry> entrys:->保存的就是多条SQL引起的变化
  35.             List<CanalEntry.Entry> entries = message.getEntries();
  36.             for (CanalEntry.Entry entry : entries) {
  37.                 //Entry:一个SQL引起的变化
  38.                 //只要insert语句
  39.                 if(entry.getEntryType().equals(CanalEntry.EntryType.ROWDATA)){
  40.                     //ByteStrinng storeValue_存的是一条SQL引起的数据变化,,不能直接使用,必须反序列化
  41.                     ByteString storeValue = entry.getStoreValue();
  42.                     parseDate(storeValue);
  43.                 }
  47.             }
  50.         }
  51.         //
  52.     }
  54.     private static void parseDate(ByteString storeValue) throws InvalidProtocolBufferException {
  55.         //反序列化
  56.         CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(storeValue);
  57.         //只要insert
  58.         if (rowChange.getEventType().equals(CanalEntry.EventType.INSERT)){
  59.             List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList();
  60.             //将每一行的所有数据封装为json字符串
  61.             for (CanalEntry.RowData rowData : rowDatasList) {
  62.                 JSONObject jsonObject = new JSONObject();
  63.                 //获取insert之后的列的信息
  64.                 //Column:可以获取列之前的值以及之后的值,以及列名
  65.                 List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
  66.                 for (CanalEntry.Column column : afterColumnsList) {
  67.                     jsonObject.put(column.getName(),column.getValue());
  68.                 }
  69.                 //System.out.println(jsonObject.toJSONString());
  70.                 //生产到kafka
  71.                 MyProducer.sendData(MyConstants.GMALL_ORDER_INFO,jsonObject.toJSONString());
  72.             }
  73.         }
  74.     }
  76. }

四.运行程序,去kafka客户端查看

Spark实时数仓(二) - 图10

3.编辑子模块(gmall-realtime)

实时处理从canal客户端发送到kafka的数据

一.pom.xml

  1. <dependency>
  2.     <groupId>mysql</groupId>
  3.      <artifactId>mysql-connector-java</artifactId>
  4.       <version>5.1.30</version>
  5.  </dependency>
  7.  <dependency>
  8.       <groupId>org.elasticsearch</groupId>
  9.       <artifactId>elasticsearch-hadoop</artifactId>
  10.       <version>6.6.0</version>
  11.  </dependency>
  13.  <dependency>
  14.      <groupId>com.alibaba</groupId>
  15.       <artifactId>druid</artifactId>
  16.      <version>1.0.5</version>
  17.  </dependency>

二.配置文件(db.properties)

  1. #Mysql
  2. jdbc.datasource.size=20
  3. jdbc.url=jdbc:mysql://hadoop103:3306/sparkproduct?useUnicode=true&characterEncoding=utf8&rewriteBatchedStatements=true
  4. jdbc.user=root
  5. jdbc.password=321074
  6. jdbc.driver.name=com.mysql.jdbc.Driver

三.工具类(JDBCUtil.java)

  1. package com.atguigu.realtime.utils
  3. import java.sql.{Connection, PreparedStatement, ResultSet}
  4. import java.util.Properties
  5. import com.alibaba.druid.pool.DruidDataSourceFactory
  6. import javax.sql.DataSource
  9. object JDBCUtil {
  11.   private val properties: Properties = PropertiesUtil.load("db.properties")
  13.   // 创建连接池对象
  14.   var dataSource:DataSource = init()
  16.   // 连接池的初始化
  17.   def init():DataSource = {
  19.     val paramMap = new java.util.HashMap[String, String]()
  20.     paramMap.put("driverClassName", properties.getProperty("jdbc.driver.name"))
  21.     paramMap.put("url", properties.getProperty("jdbc.url"))
  22.     paramMap.put("username", properties.getProperty("jdbc.user"))
  23.     paramMap.put("password", properties.getProperty("jdbc.password"))
  24.     paramMap.put("maxActive", properties.getProperty("jdbc.datasource.size"))
  26.     // 使用Druid连接池对象
  27.     DruidDataSourceFactory.createDataSource(paramMap)
  28.   }
  30.   // 从连接池中获取连接对象
  31.   def getConnection(): Connection = {
  32.     dataSource.getConnection
  33.   }
  35.   def main(args: Array[String]): Unit = {
  37.     println(getConnection())
  39.   }
  42. }

四.GMV实现

  1. 创建两张表分别存储GMV和偏移量
  1. CREATE TABLE `gmvstats` (
  2.   `date` VARCHAR(10) NOT NULL,
  3.   `hour` VARCHAR(10) NOT NULL,
  4.   `gmv` DECIMAL(20,2) DEFAULT NULL,
  5.   PRIMARY KEY (`date`,`hour`)
  6. ) ENGINE=INNODB DEFAULT CHARSET=utf8;
  1. CREATE TABLE `offsets` (
  2.   `group_id` VARCHAR(200) NOT NULL,
  3.   `topic` VARCHAR(200) NOT NULL,
  4.   `partition` INT(5) NOT NULL,
  5.   `offset` BIGINT(10) ,
  6.   PRIMARY KEY (`group_id`,`topic`,`partition`)
  7. ) ENGINE=INNODB DEFAULT CHARSET=utf8;
  1. 样例类
  1. package com.atguigu.realtime.beans
  3. case class OrderInfo(
  4.                       id: String,
  5.                       province_id: String,
  6.                       consignee: String,
  7.                       order_comment: String,
  8.                       var consignee_tel: String,
  9.                       order_status: String,
  10.                       payment_way: String,
  11.                       user_id: String,
  12.                       img_url: String,
  13.                       total_amount: Double,
  14.                       expire_time: String,
  15.                       delivery_address: String,
  16.                       create_time: String,
  17.                       operate_time: String,
  18.                       tracking_no: String,
  19.                       parent_order_id: String,
  20.                       out_trade_no: String,
  21.                       trade_body: String,
  22.                       // 方便分时和每日统计，额外添加的字段
  23.                       var create_date: String,
  24.                       var create_hour: String)
  1. 交易额统计程序(GMVApp)
  1. package com.atguigu.realtime.app
  3. import com.alibaba.fastjson.JSON
  4. import com.atguigu.constants.MyConstants
  5. import com.atguigu.realtime.app.DAUApp.{appName, batchDuration, groupId, ssc}
  6. import com.atguigu.realtime.beans.OrderInfo
  7. import com.atguigu.realtime.utils.{JDBCUtil, MyKafkaUtil}
  8. import org.apache.kafka.clients.consumer.ConsumerRecord
  9. import org.apache.kafka.common.TopicPartition
  10. import org.apache.spark.SparkConf
  11. import org.apache.spark.rdd.RDD
  12. import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}
  13. import org.apache.spark.streaming.{Seconds, StreamingContext}
  15. import java.sql.{Connection, PreparedStatement}
  16. import java.time.LocalDateTime
  17. import java.time.format.DateTimeFormatter
  18. import scala.collection.mutable
  19. /**
  20.  * 
  21.  *
  22.  *    计算每天，所有订单的GMV数据。
  23.  *         每天的总GMV
  24.  *         每天每个时段的GMV
  25.  *
  26.  *         结果： K(时间的说明)-V(GMV)
  27.  *
  28.  *         聚合类的运算：     采取  at least once +  事务(聚合的结果  +  偏移量) 维护到mysql!
  29.  *
  30.  *    
  31.  */
  33. object GMVApp extends BaseApp {
  34.   override var appName: String = "GMVApp"
  35.   override var batchDuration: Int = 10
  36.   val groupId="gmallrealtime"
  37.   /**
  38.    * 查询上一次消费的offset
  39.    * @param groupId
  40.    * @param topicName
  41.    * @return
  42.    */
  43.   def selectLatestOffsetsFromDB(groupId: String, topicName: String) = {
  44.     val offsets = mutable.Map[TopicPartition, Long]()
  45.     var connection:Connection = null
  46.     var ps:PreparedStatement = null
  47.     try {
  48.       connection = JDBCUtil.getConnection()
  49.       val sql =
  50.         """
  51.           |select * from `offsets` where `groupid` = ? and `topic` = ?
  52.           |""".stripMargin
  54.       ps = connection.prepareStatement(sql)
  55.       ps.setString(1,groupId)
  56.       ps.setString(2,topicName)
  57.       val resultSet = ps.executeQuery()
  58.       while(resultSet.next()){
  59.         val topicPartition = new TopicPartition(resultSet.getString("topic"), resultSet.getInt("partitionid"))
  60.         val offset = resultSet.getLong("offset")
  61.         offsets.put(topicPartition,offset)
  62.       }
  63.     } catch {
  64.       case e:Exception =>{
  65.         e.printStackTrace()
  66.         throw new RuntimeException("查询偏移量失败")
  67.       }
  68.     } finally {
  69.       if(ps != null){
  70.         ps.close()
  71.       }
  72.       if (connection != null){
  73.         connection.close()
  74.       }
  75.     }
  76.     offsets.toMap
  77.   }
  79.   /**
  80.    * 将Rdd封装为样例类
  81.    * @param rdd
  82.    * @return
  83.    */
  84.   def parseRecordToOrderInfo(rdd: RDD[ConsumerRecord[String, String]]):RDD[OrderInfo] = {
  85.     rdd.map(record=>{
  86.       val orderInfo = JSON.parseObject(record.value(), classOf[OrderInfo])
  87.       //"create_time": "2021-09-28 21:12:40",
  88.       val formatter1 = DateTimeFormatter.ofPattern("yyyy-MM-dd")
  89.       val formatter2 = DateTimeFormatter.ofPattern("HH")
  90.       val formatter3 = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")
  91.       // 先将 "2021-09-28 21:12:40" 转换为  LocalDateTime对象
  92.       val localDateTime = LocalDateTime.parse(orderInfo.create_time, formatter3)
  93.       orderInfo.create_date=localDateTime.format(formatter1)
  94.       orderInfo.create_hour=localDateTime.format(formatter2)
  95.       orderInfo
  96.     })
  97.   }
  98.   /**
  99.    * 在一个事务中,将结果和偏移量写出到mysql数据库中
  100.    * @param results
  101.    * @param offsetRanges
  102.    */
  103.   def writeDataAndOffsetInCommonBatch(results: Array[((String, String), Double)], offsetRanges: Array[OffsetRange]): Unit = {
  104.     var connection:Connection = null
  105.     var ps1:PreparedStatement = null
  106.     var ps2:PreparedStatement = null
  107.     try {
  108.       connection = JDBCUtil.getConnection()
  109.       //关闭事务的自动提交
  110.       connection.setAutoCommit(false)
  111.       //先判断当前的是有状态计算,还是无状态计算
  112.       //有状态计算
  113.       val sql1 =
  114.       """
  115.         |insert into gmvstats values(?,?,?)
  116.         |on duplicate key update gmv = gmv + values(gmv)
  117.         |""".stripMargin
  119.       val sql2 =
  120.         """
  121.           |insert into offsets values(?,?,?,?)
  122.           |on duplicate key update offset = values(offset)
  123.           |""".stripMargin
  125.       ps1 = connection.prepareStatement(sql1)
  126.       ps2 = connection.prepareStatement(sql2)
  127.       for (((date, hour), gmv)<- results) {
  128.         ps1.setString(1,date)
  129.         ps1.setString(2,hour)
  130.         ps1.setDouble(3,gmv)
  131.         //攒起来
  132.         ps1.addBatch
  133.       }
  134.       //批量提交
  135.       val dataResults = ps1.executeBatch()
  137.       for (offsetRange <- offsetRanges) {
  138.         ps2.setString(1,groupId)
  139.         ps2.setString(2,offsetRange.topic)
  140.         ps2.setInt(3,offsetRange.partition)
  141.         ps2.setLong(4,offsetRange.untilOffset)
  142.         //遍历一次,写一个分区的信息,一个offsetRange代表一个主题的一个分区
  143.         ps2.addBatch()
  144.       }
  145.       val offsetResult = ps2.executeBatch()
  146.       //提交事务
  147.       connection.commit()
  148.       println("写入数据" + dataResults.size)
  149.       println("写入偏移量" + offsetResult.size)
  150.     } catch {
  151.       case e:Exception =>{
  152.         connection.rollback()
  153.         e.printStackTrace()
  154.         throw new RuntimeException("写入结果和偏移量失败")
  155.       }
  156.     } finally {
  157.       if(ps1 != null){
  158.         ps1.close()
  159.       }
  160.       if(ps2 != null){
  161.         ps2.close()
  162.       }
  164.       if (connection != null){
  165.         connection.close()
  166.       }
  167.     }
  168.   }
  170.   def main(args: Array[String]): Unit = {
  171.     ssc = new StreamingContext(new SparkConf().setMaster("local[4]").setAppName(appName),Seconds(batchDuration))
  172.     runApp{
  173.       //1.查询上次消费的offsets的位置
  174.       val offsets = selectLatestOffsetsFromDB(groupId, MyConstants.GMALL_ORDER_INFO)
  175.       //2,根据上次消费的位置,获取ds
  176.       val ds = MyKafkaUtil.getKafkaStream(Array(MyConstants.GMALL_ORDER_INFO),ssc,
  177.         groupId,true,offsets)
  178.       //3.获取初始ds的偏移量信息
  179.       ds.foreachRDD(rdd=>{
  180.         if(!rdd.isEmpty()){
  181.           val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
  182.           //4.封装为样例类
  183.           val rdd1 = parseRecordToOrderInfo(rdd)
  184.           //5.计算GMV
  185.           val rdd2 = rdd1.map(orderInfo => ((orderInfo.create_date, orderInfo.create_hour), orderInfo.total_amount))
  186.             .reduceByKey(_ + _)
  187.           //6.将结果拉取到driver端和偏移量在一个事务中写入mysql
  188.           val result = rdd2.collect()
  189.           writeDataAndOffsetInCommonBatch(result,offsetRanges)
  190.         }
  192.       })
  193.     }
  194.   }
  195. }
  1. 启动APP,并去mysql客户端查看

Spark实时数仓(二) - 图11

Spark实时数仓(二) - 图12

4.编辑子模块(gmall-publisher)

一.业务代码实现

  1. pojo层(GMVPerHour.java)
  1. package com.example.gmallpublisher.pojo;
  3. import lombok.AllArgsConstructor;
  4. import lombok.Data;
  5. import lombok.NoArgsConstructor;
  7. @NoArgsConstructor
  8. @AllArgsConstructor
  9. @Data
  10. public class GMVPerHour {
  11.     private String item;
  12.     private Integer gmv;
  13. }
  1. mapper层(GMVMapper)
  1. package com.example.gmallpublisher.mapper;
  3. import com.baomidou.dynamic.datasource.annotation.DS;
  4. import com.example.gmallpublisher.pojo.GMVPerHour;
  5. import org.springframework.stereotype.Repository;
  7. import java.util.List;
  8. @Repository
  9. @DS("mysql")
  10. public interface GMVMapper {
  11.     //查询当天累计的GMV
  12.     Double getGMVByDate(String date);
  13.     //查询当天分时GMV
  14.     List<GMVPerHour> getGMVPerHourDate(String date);
  15. }
  1. GMVMapper.xml
  1. <?xml version="1.0" encoding="UTF-8" ?>
  2. <!DOCTYPE mapper
  3.         PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
  4.         "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
  5. <!--namespace=绑定一个对应的Dao/Mapper接口-->
  6. <mapper namespace="com.example.gmallpublisher.mapper.GMVMapper">
  7. <!--查询当日累计的GMV-->
  8.     <select id="getGMVByDate" resultType="double">
  9.         select
  10.             sum(gmv)
  11.            from gmvstats
  12.         where `date` = #{logdate}
  13.     </select>
  14. <!-- 查询分时GMV-->
  15.     <select id="getGMVPerHourDate" resultType="com.example.gmallpublisher.pojo.GMVPerHour">
  16.         select
  17.            `hour` item,
  18.             sum(gmv) gmv
  19.         from gmvstats
  20.       where `date`= #{logdate}
  21.      group by `hour`
  22.     </select>
  23. </mapper>
  1. service层(PublisherService,PublisherServiceImpl)
  1. package com.example.gmallpublisher.service;
  3. import com.example.gmallpublisher.pojo.DAUPerHour;
  4. import com.example.gmallpublisher.pojo.GMVPerHour;
  6. import java.util.List;
  8. public interface PublisherService {
  9.     //查询单日日活
  10.     Integer getDAUByDate(String date);
  11.     //查询当日新增的设备数
  12.     Integer getNewMidCountByDate(String date);
  13.     //查询当日分时的设备数
  14.     List<DAUPerHour> getDAUPerHourData(String date);
  15.     //查询当天累计的GMV
  16.     Double getGMVByDate(String date);
  17.     //查询当天分时GMV
  18.     List<GMVPerHour> getGMVPerHourDate(String date);
  19. }
  1. package com.example.gmallpublisher.service;
  3. import com.example.gmallpublisher.mapper.DAUMapper;
  4. import com.example.gmallpublisher.mapper.GMVMapper;
  5. import com.example.gmallpublisher.pojo.DAUPerHour;
  8. import com.example.gmallpublisher.pojo.GMVPerHour;
  9. import org.springframework.beans.factory.annotation.Autowired;
  10. import org.springframework.stereotype.Service;
  13. import java.util.List;
  15. @Service
  16. public class PublisherServiceImpl implements PublisherService {
  17.     @Autowired
  18.     private DAUMapper dauMapper;
  19.     @Autowired
  20.     private GMVMapper gmvMapper;
  21.     @Override
  22.     public Integer getDAUByDate(String date) {
  23.         System.out.println("遵循的业务步骤");
  24.         return dauMapper.getDAUByDate(date);
  25.     }
  27.     @Override
  28.     public Integer getNewMidCountByDate(String date) {
  29.         System.out.println("遵循的业务步骤");
  30.         return dauMapper.getNewMidCountByDate(date);
  31.     }
  33.     @Override
  34.     public List<DAUPerHour> getDAUPerHourData(String date) {
  35.         System.out.println("遵循业务步骤");
  36.         return dauMapper.getDAUPerHourData(date);
  37.     }
  39.     @Override
  40.     public Double getGMVByDate(String date) {
  41.         System.out.println("遵循业务步骤");
  42.         return gmvMapper.getGMVByDate(date);
  43.     }
  45.     @Override
  46.     public List<GMVPerHour> getGMVPerHourDate(String date) {
  47.         System.out.println("遵循业务步骤");
  48.         return gmvMapper.getGMVPerHourDate(date);
  49.     }
  50. }
  1. controller层
  1. package com.example.gmallpublisher.controller;
  3. import com.alibaba.fastjson.JSONObject;
  4. import com.example.gmallpublisher.pojo.DAUPerHour;
  5. import com.example.gmallpublisher.pojo.GMVPerHour;
  6. import com.example.gmallpublisher.service.PublisherService;
  7. import org.springframework.beans.factory.annotation.Autowired;
  8. import org.springframework.web.bind.annotation.RequestMapping;
  9. import org.springframework.web.bind.annotation.RestController;
  11. import java.time.LocalDate;
  12. import java.util.ArrayList;
  13. import java.util.List;
  15. @RestController
  16. public class PublisherController {
  17.     @Autowired
  18.     private PublisherService publisherService;
  20.     /**
  21.      * [
  22.      *    {"id":"dau","name":"当日日活数","value":1200},
  23.      *    {"id":"new_mid","name":"新增设备数","value":233}
  24.      *    {"id":"order_amount","name":"当日交易额","value":1000.2 }
  25.      *  ]
  26.      * @param date
  27.      * @return
  28.      */
  29.     @RequestMapping(value="/realtime-total")
  30.     public Object handle1(String date){
  31.         //获取当日日活
  32.         Integer dauByDate = publisherService.getDAUByDate(date);
  33.         //获取当日新增设备数
  34.         Integer newMidCountByDate = publisherService.getNewMidCountByDate(date);
  35.         //获取当天累计的交易额
  36.         Double gmvByDate = publisherService.getGMVByDate(date);
  37.         ArrayList<JSONObject> result = new ArrayList<>();
  38.         JSONObject jsonObject1 = new JSONObject();
  39.         JSONObject jsonObject2 = new JSONObject();
  40.         JSONObject jsonObject3 = new JSONObject();
  41.         jsonObject1.put("id","dau");
  42.         jsonObject1.put("name","当日日活数");
  43.         jsonObject1.put("value",dauByDate);
  44.         jsonObject2.put("id","new_mid");
  45.         jsonObject2.put("name","新增设备数");
  46.         jsonObject2.put("value",newMidCountByDate);
  47.         jsonObject3.put("id","order_amount");
  48.         jsonObject3.put("name","当日交易额");
  49.         jsonObject3.put("value",gmvByDate);
  50.         result.add(jsonObject1);
  51.         result.add(jsonObject2);
  52.         result.add(jsonObject3);
  55.         return result;
  58.     }
  60.     /**
  61.      *
  62.      *   {"yesterday":
  63.            {
  64.               "11":383,
  65.               "12":123,
  66.               "17":88,
  67.               "19":200
  68.          },
  69.      "today":
  70.             {"12":38,"13":1233,"17":123,"19":688 }
  71.      }
  72.      *
  73.      * @param date
  74.      * @return
  75.      */
  77.     @RequestMapping(value="realtime-hours")
  78.     public Object handle2(String id,String date){
  79.         JSONObject result = new JSONObject();
  80.         //根据今天的日期计算昨天的日期
  81.         String yestodayDate = LocalDate.parse(date).minusDays(1).toString();
  83.         if("dau".equals(id)){
  84.             List<DAUPerHour> todayData = publisherService.getDAUPerHourData(date);
  85.             List<DAUPerHour> yestodayData = publisherService.getDAUPerHourData(yestodayDate);
  86.             result.put("yesterday",parseDAUDate(yestodayData));
  87.             result.put("today",parseDAUDate(todayData));
  88.         }
  89.         if ("order_amount".equals(id)){
  90.             List<GMVPerHour> todayData = publisherService.getGMVPerHourDate(date);
  91.             List<GMVPerHour> yestodayData = publisherService.getGMVPerHourDate(yestodayDate);
  92.             result.put("yesterday",parseGMVData(yestodayData));
  93.             result.put("today",parseGMVData(todayData));
  95.         }
  97.         return result;
  99.     }
  101.     private Object parseGMVData(List<GMVPerHour> data) {
  102.         JSONObject result = new JSONObject();
  103.         for (GMVPerHour gmvPerHour : data) {
  104.             result.put(gmvPerHour.getItem(),gmvPerHour.getGmv());
  105.         }
  106.         return result;
  107.     }
  110.     public JSONObject parseDAUDate(List<DAUPerHour> data){
  111.         JSONObject result = new JSONObject();
  112.         for (DAUPerHour dauPerHour : data) {
  113.             result.put(dauPerHour.getHour(),dauPerHour.getDau());
  114.         }
  115.         return result;
  116.     }
  117. }
  1. 主程序
  1. package com.example.gmallpublisher;
  3. import org.mybatis.spring.annotation.MapperScan;
  4. import org.springframework.boot.SpringApplication;
  5. import org.springframework.boot.autoconfigure.SpringBootApplication;
  7. @SpringBootApplication
  8. @MapperScan(basePackages = "com.example.gmallpublisher.mapper")
  9. public class GmallPublisherApplication {
  11.     public static void main(String[] args) {
  12.         SpringApplication.run(GmallPublisherApplication.class, args);
  13.     }
  15. }
  1. index.html
  1. <!DOCTYPE html>
  2. <html lang="en">
  3. <head>
  4.     <meta charset="UTF-8">
  5.     <title>首页</title>
  6. </head>
  7. <body>
  8. <a href=" http://localhost:8070/realtime-total?date=2021-10-07">请求今天DAU和GMV</a> <br>
  9. <a href=" http://localhost:8070/realtime-hours?id=dau&date=2021-10-07">请求今天和昨天分时DAU</a>
  10. <br/>
  11. <a href=" http://localhost:8070/realtime-hours?id=order_amount&date=2021-10-07">请求今天和昨天分时GMV</a>
  14. </body>
  15. </html>

二.运行程序

浏览器访问: http://localhost:8070/index.html

Spark实时数仓(二) - 图13

  1. 请求今天DAU和GMV

Spark实时数仓(二) - 图14

  1. 请求今天和昨天分时DAU

Spark实时数仓(二) - 图15

  1. 请求今天和昨天分时GMV

Spark实时数仓(二) - 图16