第一章.项目概述

1.实时需求与离线需求的比较

  • 离线需求(T+1):一般是根据前一日的数据生成报表等数据,虽然统计指标,报表繁多,但是对时效性不敏感
  • 实时需求(T+0):主要侧重于对当日数据的实时监控,通常业务逻辑相对离线需求简单一下,统计指标也少一些,但是更注重数据的时效性,以及用户的交互性

2.需求说明

  • 日用户首次登陆（日活）分时趋势图，昨日对比
  • 交易额及分时趋势图，昨日对比
  • 购物券功能风险预警
  • 用户购买明细灵活分析功能

3.架构分析

一.离线数仓

Spark实时数仓(一) - 图1

二.Spark实时数仓

Spark实时数仓(一) - 图2

4.项目涉及技术

Spark实时数仓(一) - 图3

第二章.基础工程搭建

1.创建父工程(spark-gmall)

pom.xml

  1. <properties>
  2.         <maven.compiler.source>8</maven.compiler.source>
  3.         <maven.compiler.target>8</maven.compiler.target>
  4.         <spark.version>3.0.0</spark.version>
  5.         <scala.version>2.12.10</scala.version>
  6.         <log4j.version>1.2.17</log4j.version>
  7.         <fastjson.version>1.2.62</fastjson.version>
  8.         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  9.         <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
  10.         <java.version>1.8</java.version>
  11.         <kafka.client.verson>2.4.1</kafka.client.verson>
  12.     </properties>
  15.     <dependencies>
  16.         <!-- 具体的日志实现 -->
  17.         <dependency>
  18.             <groupId>log4j</groupId>
  19.             <artifactId>log4j</artifactId>
  20.             <version>${log4j.version}</version>
  21.         </dependency>
  22.     </dependencies>
  24.     <dependencyManagement>
  25.         <dependencies>
  27.             <dependency>
  28.                 <groupId>com.alibaba</groupId>
  29.                 <artifactId>fastjson</artifactId>
  30.                 <version>${fastjson.version}</version>
  31.             </dependency>
  33.             <dependency>
  34.                 <groupId>org.apache.spark</groupId>
  35.                 <artifactId>spark-core_2.12</artifactId>
  36.                 <version>${spark.version}</version>
  37.             </dependency>
  39.             <dependency>
  40.                 <groupId>org.apache.spark</groupId>
  41.                 <artifactId>spark-hive_2.12</artifactId>
  42.                 <version>${spark.version}</version>
  43.             </dependency>
  45.             <dependency>
  46.                 <groupId>org.apache.spark</groupId>
  47.                 <artifactId>spark-sql_2.12</artifactId>
  48.                 <version>${spark.version}</version>
  49.             </dependency>
  51.             <dependency>
  52.                 <groupId>org.apache.spark</groupId>
  53.                 <artifactId>spark-streaming_2.12</artifactId>
  54.                 <version>${spark.version}</version>
  55.             </dependency>
  57.             <dependency>
  58.                 <groupId>org.apache.spark</groupId>
  59.                 <artifactId>spark-streaming-kafka-0-10_2.12</artifactId>
  60.                 <version>${spark.version}</version>
  61.             </dependency>
  63.             <dependency>
  64.                 <groupId>org.apache.kafka</groupId>
  65.                 <artifactId>kafka-clients</artifactId>
  66.                 <version>${kafka.client.verson}</version>
  67.             </dependency>
  68.         </dependencies>
  69.     </dependencyManagement>

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

gmall-common模块主要提供整个项目各模块共用的公共工具类,常量和依赖

一.pom.xml

  1. <dependencies>
  2.  <dependency>
  3.       <groupId>com.alibaba</groupId>
  4.       <artifactId>fastjson</artifactId>
  5.       </dependency>
  6. </dependencies>

二.MyConstants.java

  1. package com.atguigu.constants;
  3. public interface MyConstants {
  4.     //保存kafka的五种用户行为
  5.     String STARTUP_LOG = "STARTUP_LOG";
  6.     String ERROR_LOG = "ERROR_LOG";
  7.     String DISPLAY_LOG = "DISPLAY_LOG";
  8.     String PAGE_LOG = "PAGE_LOG";
  9.     String ACTIONS_LOG = "ACTIONS_LOG";
  10.     //交易明细需要使用的三种不同的业务数据
  11.     String GMALL_ORDER_INFO = "GMALL_ORDER_INFO";
  12.     String GMALL_ORDER_DETAIL = "GMALL_ORDER_DETAIL";
  13.     String GMALL_USER_INFO = "GMALL_USER_INFO";
  14. }

3.创建子模块(gmall-logger)

gmall-logger主要用于采集日志数据到kafka

该模块是一个webapp ,创建时选SpringBoot

Spark实时数仓(一) - 图4

一.项目搭建

  1. 将gmall-logger模块的pom.xml中的以下配置剪切到spark-gmall的pom.xml中
  1. <parent>
  2.         <groupId>org.springframework.boot</groupId>
  3.         <artifactId>spring-boot-starter-parent</artifactId>
  4.         <version>1.5.10.RELEASE</version>
  5.         <relativePath/> <!-- lookup parent from repository -->
  6.  </parent>
  1. 将gmall-common模块pom.xml中的以下配置复制到gmall-logger模块的pom.xml
  1. <parent>
  2.         <artifactId>spark-gmall</artifactId>
  3.         <groupId>org.example</groupId>
  4.         <version>1.0-SNAPSHOT</version>
  5. </parent>
  1. spark-gmall的pom.xml添加以下内容
  1. <modules>
  2.         <module>gmall-common</module>
  3.         <module>gmall-logger</module>
  4.         <module>gmall-realtime</module>
  5.         <module>gmall-canalclient</module>
  6. </modules>
  1. gmall-logger.xml
  1. <dependencies>
  2.         <dependency>
  3.             <groupId>org.springframework.boot</groupId>
  4.             <artifactId>spring-boot-starter-web</artifactId>
  5.             <exclusions>
  6.                 <!--排除自带的 logger -->
  7.                 <exclusion>
  8.                     <groupId>org.springframework.boot</groupId>
  9.                     <artifactId>spring-boot-starter-logging</artifactId>
  10.                 </exclusion>
  11.             </exclusions>
  13.         </dependency>
  14.         <!-- java.lang.ClassNotFoundException: org.apache.commons.logging.LogFactory-->
  15.         <dependency>
  16.             <groupId>commons-logging</groupId>
  17.             <artifactId>commons-logging</artifactId>
  18.             <version>1.1.1</version>
  19.         </dependency>
  20.         <dependency>
  21.             <groupId>org.springframework.kafka</groupId>
  22.             <artifactId>spring-kafka</artifactId>
  23.         </dependency>
  25.         <dependency>
  26.             <groupId>org.projectlombok</groupId>
  27.             <artifactId>lombok</artifactId>
  28.             <optional>true</optional>
  29.         </dependency>
  30.         <dependency>
  31.             <artifactId>gmall-common</artifactId>
  32.             <groupId>org.example</groupId>
  33.             <version>1.0-SNAPSHOT</version>
  34.         </dependency>
  35.         <dependency>
  36.             <groupId>org.springframework.boot</groupId>
  37.             <artifactId>spring-boot-starter-test</artifactId>
  38.             <scope>test</scope>
  39.         </dependency>
  40.         <dependency>
  41.             <groupId>org.springframework.kafka</groupId>
  42.             <artifactId>spring-kafka-test</artifactId>
  43.             <scope>test</scope>
  44.         </dependency>
  45.         <dependency>
  46.             <groupId>org.apache.kafka</groupId>
  47.             <artifactId>kafka-clients</artifactId>
  48.         </dependency>
  49.     </dependencies>
  51.     <build>
  52.         <plugins>
  53.             <plugin>
  54.                 <groupId>org.springframework.boot</groupId>
  55.                 <artifactId>spring-boot-maven-plugin</artifactId>
  56.                 <version>1.5.10.RELEASE</version>
  57.                 <configuration>
  58.                     <excludes>
  59.                         <exclude>
  60.                             <groupId>org.projectlombok</groupId>
  61.                             <artifactId>lombok</artifactId>
  62.                         </exclude>
  63.                     </excludes>
  64.                 </configuration>
  65.             </plugin>
  66.         </plugins>
  67.     </build>
  1. application.properties
  1. server.context-path=/gmall_logger
  2. server.port=8888
  4. #============== kafka ===================
  5. # 指定kafka 代理地址，可以多个
  6. spring.kafka.bootstrap-servers=hadoop102:9092,hadoop103:9092,hadoop104:9092
  8. # 指定消息key和消息体的编解码方式
  9. spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
  10. spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
  1. 整合log4j(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
  10. #定义了一个DailyRollingFileAppender 将日志写入到一个按照日期滚动的文件中。 当天的数据，会写入到app.log中，
  11. # 过一天后，app.log会滚动，滚动的旧文件 以 app.log.昨天的日期 ，滚动的新文件还叫app.log
  12. log4j.appender.atguigu.File=org.apache.log4j.DailyRollingFileAppender
  13. log4j.appender.atguigu.File.file=/opt/module/logs/app.log
  14. log4j.appender.atguigu.File.DatePattern='.'yyyy-MM-dd
  15. log4j.appender.atguigu.File.layout=org.apache.log4j.PatternLayout
  16. log4j.appender.atguigu.File.layout.ConversionPattern=%m%n
  19. #代表指定哪个类的logger使用什么级别和appender进行日志输出  全类名可能需要修改
  20. log4j.logger.com.atguigu.gmalllogger.controller.LogController=info,atguigu.File,atguigu.MyConsole
  1. 编写controller
  1. package com.atguigu.gmalllogger.controller;
  2. import com.alibaba.fastjson.JSON;
  3. import com.alibaba.fastjson.JSONObject;
  4. import com.atguigu.constants.MyConstants;
  5. import lombok.extern.log4j.Log4j;
  6. import org.springframework.beans.factory.annotation.Autowired;
  7. import org.springframework.web.bind.annotation.RequestMapping;
  8. import org.springframework.web.bind.annotation.RestController;
  9. import org.springframework.kafka.core.KafkaTemplate;
  10. /**
  11.  * 客户端在发送日志的时候，参数名是固定的
  12.  */
  13. @RestController
  14. @Log4j //Logger log = Logger.getLogger(LogController.class)
  15. public class LogController {
  16.     //从容器中找一个KafkaTemplate类型的对象，赋值给producer
  17.     @Autowired
  18.     private KafkaTemplate producer;
  19.     @RequestMapping(value="/applog")
  20.     public Object handle1(String param){
  21.         //System.out.println(param);
  22.         log.info(param);
  23.         //先将param由JsonStr转化为java对象（Map）
  24.         JSONObject jsonObject = JSON.parseObject(param);
  25.         if (jsonObject.get("start") != null){
  27.             producer.send(MyConstants.STARTUP_LOG, param );
  29.         }
  31.         if (jsonObject.get("actions") != null){
  33.             producer.send(MyConstants.ACTIONS_LOG, param );
  35.         }
  37.         if (jsonObject.get("page") != null){
  39.             producer.send(MyConstants.PAGE_LOG, param );
  41.         }
  44.         if (jsonObject.get("displays") != null){
  46.             producer.send(MyConstants.DISPLAY_LOG, param );
  48.         }
  50.         if (jsonObject.get("err") != null){
  52.             producer.send(MyConstants.ERROR_LOG, param );
  54.         }
  55.         return "success";
  56.     }
  57.     @RequestMapping(value = "/hello")
  58.     public  Object handle2(){
  59.         System.out.println("hello!");
  60.         return "success";
  62.     }
  63. }
  1. 将程序打成jar包,上传到虚拟机,并分发

二.配置nginx(反向代理)

  1. 什么是正向代理和反向代理?
  1. # 正向代理(代理服务器代理的是客户端的ip)
  2. 正向代理: 代理的是客户端的ip
  3. 例如 客户端Aip无法访问B域名
  4. 可以让客户端Aip先访问一下代理服务器,将Aip代理为Cip,使用Cip访问B域名
  6. #反向代理(代理服务器代理的是服务端的ip)
  7. 反向代理: 代理的是服务端的ip
  8. 例如 A用户访问 B域名,C域名,D域名
  9. 让A用户访问一个代理服务器,由代理服务器将请求转到指定的A,B,C域名
  1. 安装
  1. 1.yum安装依赖包
  2. sudo yum -y install    openssl openssl-devel pcre pcre-devel    zlib zlib-devel gcc gcc-c++
  3. 2.上传压缩包到linux上并解压
  4. tar -zxvf nginx-1.12.2.tar.gz -C /opt/module
  5. 3. 进入解压缩目录,执行
  6. ./configure --prefix=/opt/module/nginx
  7. make && make install
  1. 启动,关闭命令
  1. # 启动
  2. 在/opt/module/nginx目录下执行  sudo sbin/nginx
  3. # 关闭
  4. 在/opt/module/nginx目录下执行  sudo sbin/nginx -s stop
  5. # 重新加载
  6. 在/opt/module/nginx目录下执行  sudo sbin/nginx -s reload
  1. 修改配置文件(/opt/module/nginx/conf/nginx.conf)
  1. http{
  2.    ..........
  3.     upstream logserver{
  4.       server    hadoop102:8888 weight=1;
  5.       server    hadoop103:8888 weight=1;
  6.       server    hadoop104:8888 weight=1;
  7.     }
  8.     server {
  9.         listen       81;
  10.         server_name  logserver;
  12.         location / {
  13.             root   html;
  14.             index  index.html index.htm;
  15.             proxy_pass http://logserver;
  16.             proxy_connect_timeout 10;
  17.          }
  18.    ..........
  19. }

三.gmall-logger集群脚本

  1. #!/bin/bash
  2. #第一层采集通道的一键启动脚本，只接收单个start或stop参数
  3. if(($#!=1))
  4. then
  5.         echo 请输入单个start或stop参数！
  6.         exit
  7. fi
  10. #对传入的单个参数进行校验，在hadoop102和hadoop103且执行第一层采集通道的启动和停止命令
  11. if [ $1 = start ]
  12. then
  13.         cmd="nohup java -jar /opt/module/gmall-logger.jar > /dev/null  2>&1 &"
  14.         elif [ $1 = stop ]
  15.         then
  16.                 cmd="ps -ef | grep gmall-logger.jar | grep -v grep | awk  '{print \$2}' | xargs kill "
  17. else
  18.         echo 请输入单个start或stop参数！
  19. fi
  21. #在hadoop102和hadoop103且执行第一层采集通道的启动和停止命令
  22. for i in hadoop102 hadoop103 hadoop104
  23. do
  24.         echo "--------------$i-----------------"
  25.         ssh $i $cmd
  26. done

4.创建子模块(sparkstreaming)

sparkstreaming模块主要用于学习sparkstreaming精确一次性消费

SparkStreaming精确一次性消费.pdf

一.配置文件和工具类

pom.xml

  1. <dependencies>
  3.         <dependency>
  4.             <groupId>org.apache.spark</groupId>
  5.             <artifactId>spark-core_2.12</artifactId>
  6.             <version>3.0.0</version>
  7.         </dependency>
  9.         <dependency>
  10.             <groupId>org.apache.spark</groupId>
  11.             <artifactId>spark-streaming_2.12</artifactId>
  12.             <version>3.0.0</version>
  13.         </dependency>
  15.         <dependency>
  16.             <groupId>org.apache.spark</groupId>
  17.             <artifactId>spark-streaming-kafka-0-10_2.12</artifactId>
  18.             <version>3.0.0</version>
  19.         </dependency>
  21.         <dependency>
  22.             <groupId>mysql</groupId>
  23.             <artifactId>mysql-connector-java</artifactId>
  24.             <version>8.0.16</version>
  25.         </dependency>
  26.         <dependency>
  27.             <groupId>com.alibaba</groupId>
  28.             <artifactId>druid</artifactId>
  29.             <version>1.0.5</version>
  30.         </dependency>
  32.     </dependencies>
  35.     <build>
  36.         <finalName>SparkCoreTest</finalName>
  37.         <plugins>
  38.             <plugin>
  39.                 <groupId>net.alchim31.maven</groupId>
  40.                 <artifactId>scala-maven-plugin</artifactId>
  41.                 <version>3.2.0</version>
  42.                 <executions>
  43.                     <execution>
  44.                         <goals>
  45.                             <goal>compile</goal>
  46.                             <goal>testCompile</goal>
  47.                         </goals>
  48.                     </execution>
  49.                 </executions>
  50.             </plugin>
  51.         </plugins>
  52.     </build>

db.properties

  1. #jdbc配置
  2. jdbc.datasource.size=10
  3. jdbc.url=jdbc:mysql://hadoop102:3306/sparkproduct?useSSL=false&useUnicode=true&characterEncoding=utf8&rewriteBatchedStatements=true
  4. jdbc.user=root
  5. jdbc.password=321074
  6. jdbc.driver.name=com.mysql.cj.jdbc.Driver

JDBCUtil

  1. package com.atguigu.utils
  3. import java.sql.Connection
  5. import com.alibaba.druid.pool.DruidDataSourceFactory
  6. import javax.sql.DataSource
  8. /**
  9.  * Created by VULCAN on 2020/11/3
  10.  */
  11. object JDBCUtil {
  13.     // 创建连接池对象
  14.     var dataSource:DataSource = init()
  16.     // 连接池的初始化
  17.     def init():DataSource = {
  19.         val paramMap = new java.util.HashMap[String, String]()
  20.         paramMap.put("driverClassName", PropertiesUtil.getValue("jdbc.driver.name"))
  21.         paramMap.put("url", PropertiesUtil.getValue("jdbc.url"))
  22.         paramMap.put("username", PropertiesUtil.getValue("jdbc.user"))
  23.         paramMap.put("password", PropertiesUtil.getValue("jdbc.password"))
  24.         paramMap.put("maxActive", PropertiesUtil.getValue("jdbc.datasource.size"))
  26.         // 使用Druid连接池对象
  27.         DruidDataSourceFactory.createDataSource(paramMap)
  28.     }
  30.     // 从连接池中获取连接对象
  31.     def getConnection(): Connection = {
  32.         dataSource.getConnection
  33.     }
  35. }

PropertiesUtil

  1. package com.atguigu.utils
  3. import java.util.ResourceBundle
  5. /**
  6.  * Created by VULCAN on 2020/11/3
  7.  */
  8. object PropertiesUtil {
  10.     // 绑定配置文件
  11.     // ResourceBundle专门用于读取配置文件，所以读取时，不需要增加扩展名
  12.     val resourceFile: ResourceBundle = ResourceBundle.getBundle("db")
  14.     def getValue( key : String ): String = {
  15.         resourceFile.getString(key)
  16.     }
  19. }

二.获取偏移量

  1. package com.atguigu.kafka
  2. import org.apache.kafka.clients.consumer.ConsumerRecord
  3. import org.apache.kafka.common.serialization.StringDeserializer
  4. import org.apache.spark.streaming.kafka010._
  5. import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
  6. import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
  8. /**
  9.  * 获取偏移量
  10.  * 如果想要获取当前批次数据的偏移量,只能从初始DS(DirectKafkaInputDStream)中获取
  11.  */
  13. object GetOffsetsFromKafka {
  14.   def main(args: Array[String]): Unit = {
  15.     import org.apache.spark.SparkConf
  16.     import org.apache.spark.streaming.{Seconds, StreamingContext}
  17.     //创建StreamingContext,设置一个批次的时间
  18.     val conf = new SparkConf().setMaster("local[4]").setAppName("test")
  19.     val ssc = new StreamingContext(conf, Seconds(5))
  20.     //设置日志级别
  21.     ssc.sparkContext.setLogLevel("error")
  22.     //配置kafka
  23.     val kafkaParams = Map[String, Object](
  24.       "bootstrap.servers" -> "hadoop102:9092,hadoop103:9092,hadoop104:9092",
  25.       "key.deserializer" -> classOf[StringDeserializer],
  26.       "value.deserializer" -> classOf[StringDeserializer],
  27.       "group.id" -> "app1",
  28.       "auto.offset.reset" -> "latest",
  29.       "enable.auto.commit" -> "false"
  30.     )
  32.     val topics = Array("topicA")
  33.     val stream = KafkaUtils.createDirectStream[String, String](
  34.       ssc,
  35.       LocationStrategies.PreferConsistent,
  36.       ConsumerStrategies.Subscribe[String, String](topics, kafkaParams)
  37.     )
  39.     /**
  40.      * DS.foreachRDD{
  41.      *       只有RDD算子中的代码是在Executor端运行
  42.      *       其他的都在Driver端运行
  43.      * }
  44.      */
  46.     stream.foreachRDD { rdd =>
  47.       //获取当前批次的offsets
  48.       //OffsetRange:代表kafka一个主题的一个分区,从中拉取到的数据的offset的范围
  49.       //offsetRange的个数=消费的分区数
  50.       val offsetRanges: Array[OffsetRange] = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
  51.       stream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)
  52.       for (range <- offsetRanges) {
  53.         println(Thread.currentThread().getName + ":" + range)
  54.       }
  55.       val rdd1 = rdd.map(record => {
  56.         println(Thread.currentThread().getName + ":" + record.value())
  57.         record.value
  58.       })
  59.       rdd1.collect()
  60.     }
  61.     stream.map(record => (record.key, record.value))
  63.     ssc.start()
  64.     ssc.awaitTermination()
  65.   }
  67. }

Spark实时数仓(一) - 图5

三.提交偏移量

  1. package com.atguigu.kafka
  3. import org.apache.kafka.common.serialization.StringDeserializer
  4. import org.apache.spark.streaming.kafka010._
  6. /**
  7.  * 在SparkStreaming,只有两个算子可以将DS的运算,转换为RDD的运算
  8.  *
  9.  *     1. DS.foreachRDD(rdd=>{
  10.  *         rdd.算子
  11.  *     })
  12.  *     2. DS1 = DS.transform(rdd=>{
  13.  *         rdd.算子
  14.  *     })
  15.  *     foreachRDD:输出操作,没有返回值!
  16.  *     transform:转换操作,有返回值!
  17.  *
  18.  * 只有从kafka获取到的那个DS<DirectKafkaInputDStream>,才有提交偏移量的功能!
  19.  */
  21. object GetOffsetsFromKafka2 {
  22.   def main(args: Array[String]): Unit = {
  23.     import org.apache.spark.SparkConf
  24.     import org.apache.spark.streaming.{Seconds, StreamingContext}
  25.     //创建StreamingContext,设置一个批次的时间
  26.     val conf = new SparkConf().setMaster("local[4]").setAppName("test")
  27.     val ssc = new StreamingContext(conf, Seconds(5))
  28.     //设置日志级别
  29.     ssc.sparkContext.setLogLevel("error")
  30.     //配置kafka
  31.     val kafkaParams = Map[String, Object](
  32.       "bootstrap.servers" -> "hadoop102:9092,hadoop103:9092,hadoop104:9092",
  33.       "key.deserializer" -> classOf[StringDeserializer],
  34.       "value.deserializer" -> classOf[StringDeserializer],
  35.       "group.id" -> "app1",
  36.       "auto.offset.reset" -> "latest",
  37.       "enable.auto.commit" -> "false"
  38.     )
  40.     val topics = Array("topicA")
  41.     val stream = KafkaUtils.createDirectStream[String, String](
  42.       ssc,
  43.       LocationStrategies.PreferConsistent,
  44.       ConsumerStrategies.Subscribe[String, String](topics, kafkaParams)
  45.     )
  46.     //声明一个变量来接收偏移量
  47.     var offsetRanges: Array[OffsetRange] = null
  49.     //使用transform获取偏移量
  50.     val ds1 = stream.transform(rdd => {
  51.        offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
  52.       for (elem <- offsetRanges) {
  53.         println(elem)
  54.       }
  55.       rdd
  56.     })
  57.     val ds2 = ds1.map(record => record.value())
  58.     //输出后提交
  59.     ds2.foreachRDD(rdd=>{
  60.       val results = rdd.collect()
  61.       //输出
  62.       results.foreach(println(_))
  63.       //提交偏移量
  64.       stream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)
  65.     })
  68.     stream.map(record => (record.key, record.value))
  70.     ssc.start()
  71.     ssc.awaitTermination()
  72.   }
  74. }

Spark实时数仓(一) - 图6

四.精确一次性消费案例一

  1. package com.atguigu.kafka
  3. import org.apache.kafka.common.serialization.StringDeserializer
  4. import org.apache.spark.streaming.kafka010.{CanCommitOffsets, ConsumerStrategies, HasOffsetRanges, KafkaUtils, LocationStrategies, OffsetRange}
  6. /**
  7.  * at least once (将offset维护在kafka中| 自定义存储位置) + 幂等性输出
  8.  */
  10. object ConsumeExactlyOnceDemo1 {
  11.   def main(args: Array[String]): Unit = {
  12.     import org.apache.spark.SparkConf
  13.     import org.apache.spark.streaming.{Seconds, StreamingContext}
  14.     //创建StreamingContext,设置一个批次的时间
  15.     val conf = new SparkConf().setMaster("local[4]").setAppName("test")
  16.     val ssc = new StreamingContext(conf, Seconds(5))
  17.     //设置日志级别
  18.     ssc.sparkContext.setLogLevel("error")
  19.     //配置kafka
  20.     val kafkaParams = Map[String, Object](
  21.       "bootstrap.servers" -> "hadoop102:9092,hadoop103:9092,hadoop104:9092",
  22.       "key.deserializer" -> classOf[StringDeserializer],
  23.       "value.deserializer" -> classOf[StringDeserializer],
  24.       "group.id" -> "app1",
  25.       "auto.offset.reset" -> "latest",
  26.       //1.手动提交offset
  27.       "enable.auto.commit" -> "false"
  28.     )
  30.     val topics = Array("topicA")
  31.     //2.当决定将offsets信息维护在kafka中时,默认自动会从kafka集群中获取获取当前组上次消费的位置信息
  32.     val stream = KafkaUtils.createDirectStream[String, String](
  33.       ssc,
  34.       LocationStrategies.PreferConsistent,
  35.       ConsumerStrategies.Subscribe[String, String](topics, kafkaParams)
  36.     )
  37.     //声明一个变量来接收偏移量
  38.     var offsetRanges:Array[OffsetRange] = null
  39.     //获取当前批次数据的偏移量
  40.     val ds1 = stream.transform(rdd => {
  41.       offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
  42.       for (elem <- offsetRanges) {
  43.         println(elem)
  44.       }
  45.       //3.各种转换运算
  46.       rdd
  47.     })
  48.     //3.各种转换运算
  49.     val ds2 = ds1.map(_.value())
  50.     //4.输出
  51.     ds2.foreachRDD(rdd=>{
  52.       //5.幂等性输出,保证写出的数据库是支持幂等操作
  53.       //此处代码省略.....
  54.       //6.决定提交偏移量到kafka的consumeroffsets
  55.       //手动提交offsets
  56.       stream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)
  57.     })
  58.     ssc.start()
  59.     ssc.awaitTermination()
  60.   }
  62. }

五.精确一次性消费案例二

  1. package com.atguigu.kafka
  3. import com.atguigu.utils.JDBCUtil
  4. import org.apache.kafka.common.TopicPartition
  5. import org.apache.kafka.common.serialization.StringDeserializer
  6. import org.apache.spark.streaming.kafka010._
  8. import java.sql.{Connection, PreparedStatement}
  9. import scala.collection.mutable
  11. /**
  12.  * at least once (将offset维护在kafka中| 自定义存储位置) + 事务(结果和offsets一起写出到数据库)
  13.  * 自己维护offsets存储到外部数据库(非kafka),以mysql为例
  14.  * 适用场景:聚合类运算
  15.  * 以wordcount为例
  16.  */
  18. object ConsumeExactlyOnceDemo2 {
  19.   val groupId = "test02"
  20.   val topicName = "topicA"
  22.   /**
  23.    * 查询上一次消费的offset
  24.    * @param groupId
  25.    * @param topicName
  26.    * @return
  27.    */
  28.   def selectLatestOffsetsFromDB(groupId: String, topicName: String) = {
  29.     val offsets = mutable.Map[TopicPartition, Long]()
  30.     var connection:Connection = null
  31.     var ps:PreparedStatement = null
  32.     try {
  33.       connection = JDBCUtil.getConnection()
  34.       val sql =
  35.         """
  36.           |select * from `offsets` where `groupid` = ? and `topic` = ?
  37.           |""".stripMargin
  39.       ps = connection.prepareStatement(sql)
  40.       ps.setString(1,groupId)
  41.       ps.setString(2,topicName)
  42.       val resultSet = ps.executeQuery()
  43.       while(resultSet.next()){
  44.         val topicPartition = new TopicPartition(resultSet.getString("topic"), resultSet.getInt("partitionid"))
  45.         val offset = resultSet.getLong("offset")
  46.         offsets.put(topicPartition,offset)
  47.       }
  48.     } catch {
  49.       case e:Exception =>{
  50.         e.printStackTrace()
  51.         throw new RuntimeException("查询偏移量失败")
  52.       }
  53.     } finally {
  54.       if(ps != null){
  55.         ps.close()
  56.       }
  57.       if (connection != null){
  58.         connection.close()
  59.       }
  60.     }
  61.         offsets.toMap
  62.   }
  64.   /**
  65.    * 在一个事务中,将结果和偏移量写出到mysql数据库中
  66.    * @param results
  67.    * @param offsetRanges
  68.    */
  69.   def writeDataAndOffsetInCommonBatch(results: Array[(String, Int)], offsetRanges: Array[OffsetRange]): Unit = {
  70.     var connection:Connection = null
  71.     var ps1:PreparedStatement = null
  72.     var ps2:PreparedStatement = null
  73.     try {
  74.       connection = JDBCUtil.getConnection()
  75.       //关闭事务的自动提交
  76.       connection.setAutoCommit(false)
  77.       //先判断当前的是有状态计算,还是无状态计算
  78.       //有状态计算
  79.       val sql1 =
  80.        """
  81.          |insert into wordcount values(?,?)
  82.          |on duplicate key update count = count + values(count)
  83.          |""".stripMargin
  85.       val sql2 =
  86.         """
  87.           |insert into offsets values(?,?,?,?)
  88.           |on duplicate key update offset = values(offset)
  89.           |""".stripMargin
  91.       ps1 = connection.prepareStatement(sql1)
  92.       ps2 = connection.prepareStatement(sql2)
  93.       for ((word,count)<- results) {
  94.         ps1.setString(1,word)
  95.         ps1.setInt(2,count)
  96.         //攒起来
  97.         ps1.addBatch
  98.       }
  99.       //批量提交
  100.       val dataResults = ps1.executeBatch()
  102.       for (offsetRange <- offsetRanges) {
  103.         ps2.setString(1,groupId)
  104.         ps2.setString(2,offsetRange.topic)
  105.         ps2.setInt(3,offsetRange.partition)
  106.         ps2.setLong(4,offsetRange.untilOffset)
  107.         //遍历一次,写一个分区的信息,一个offsetRange代表一个主题的一个分区
  108.         ps2.addBatch()
  109.       }
  110.       val offsetResult = ps2.executeBatch()
  111.       //提交事务
  112.       connection.commit()
  113.       println("写入数据" + dataResults.size)
  114.       println("写入偏移量" + offsetResult.size)
  115.     } catch {
  116.       case e:Exception =>{
  117.         connection.rollback()
  118.         e.printStackTrace()
  119.         throw new RuntimeException("写入结果和偏移量失败")
  120.       }
  121.     } finally {
  122.       if(ps1 != null){
  123.         ps1.close()
  124.       }
  125.       if(ps2 != null){
  126.         ps2.close()
  127.       }
  129.       if (connection != null){
  130.         connection.close()
  131.       }
  132.     }
  133.   }
  135.   def main(args: Array[String]): Unit = {
  136.     import org.apache.spark.SparkConf
  137.     import org.apache.spark.streaming.{Seconds, StreamingContext}
  138.     //创建StreamingContext,设置一个批次的时间
  139.     val conf = new SparkConf().setMaster("local[4]").setAppName("test")
  140.     val ssc = new StreamingContext(conf, Seconds(5))
  141.     //设置日志级别
  142.     ssc.sparkContext.setLogLevel("error")
  143.     //配置kafka
  144.     val kafkaParams = Map[String, Object](
  145.       "bootstrap.servers" -> "hadoop102:9092,hadoop103:9092,hadoop104:9092",
  146.       "key.deserializer" -> classOf[StringDeserializer],
  147.       "value.deserializer" -> classOf[StringDeserializer],
  148.       "group.id" -> groupId,
  149.       "auto.offset.reset" -> "latest",
  150.       //1.手动提交offset
  151.       "enable.auto.commit" -> "false"
  152.     )
  154.     val topics = Array(topicName)
  156.     /*
  157.     2.每批数据,在运算前,要先去数据库查询上一次消费的offsets
  158.     查询的条件是,哪个组,消费到了哪个topic,到哪个位置
  159.     */
  160.     val offsets = selectLatestOffsetsFromDB(groupId, topicName)
  161.     println("-------------------------")
  162.     println(offsets)
  163.     //3.基于上一次消费的最新位置,获取DS之后,进行各种转换
  164.     val stream = KafkaUtils.createDirectStream[String, String](
  165.       ssc,
  166.       LocationStrategies.PreferConsistent,
  167.       ConsumerStrategies.Subscribe[String, String](topics, kafkaParams,offsets)
  168.     )
  169.     //声明一个变量来接收偏移量
  170.     var offsetRanges:Array[OffsetRange] = null
  171.     //获取当前批次数据的偏移量
  172.     val ds1 = stream.transform(rdd => {
  173.       offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
  174.       //4.各种转换运算
  175.       val rdd1 = rdd.map(record => record.value())
  176.       rdd1
  177.     })
  178.     //4.各种转换运算
  179.     val ds2 = ds1.flatMap(_.split(" ")).map((_, 1)).reduceByKey(_ + _)
  180.     //4.输出
  181.     ds2.foreachRDD(rdd=>{
  182.       //收集聚合后的结果到Driver端
  183.       val results = rdd.collect()
  184.       println("聚合的结果"+results.mkString(","))
  185.       //在一个事务中,将结果和offsets信息写出到mysql
  186.       writeDataAndOffsetInCommonBatch(results,offsetRanges)
  187.     })
  191.     ssc.start()
  192.     ssc.awaitTermination()
  193.   }
  195. }

在虚拟机上打开kafka生产者进行测试

  1. kafka-console-producer.sh --broker-list hadoop102:9092 --topic topicA

Spark实时数仓(一) - 图7

运行程序

Spark实时数仓(一) - 图8

打开mysql客户端查看

Spark实时数仓(一) - 图9

Spark实时数仓(一) - 图10