大数据实时项目(ads层)

<dependency>    <groupId>org.scalikejdbcgroupId>    <artifactId>scalikejdbc_2.11artifactId>    <version>2.5.0version>dependency><dependency>    <groupId>org.scalikejdbcgroupId>    <artifactId>scalikejdbc-config_2.11artifactId>    <version>2.5.0version>dependency><dependency><groupId>mysqlgroupId><artifactId>mysql-connector-javaartifactId><version>5.1.47version>dependency>

import java.sql.{Connection, DriverManager, ResultSet, ResultSetMetaData, Statement}import com.alibaba.fastjson.JSONObjectimport scala.collection.mutable.ListBufferobject  MysqlUtil {  def main(args: Array[String]): Unit = {      val list:  List[ JSONObject] = queryList("select * from offset_2020")      println(list)  }  def   queryList(sql:String):List[JSONObject]={         Class.forName("com.mysql.jdbc.Driver")        val resultList: ListBuffer[JSONObject] = new  ListBuffer[ JSONObject]()        val conn: Connection = DriverManager.getConnection("jdbc:mysql://hadoop2:3306/gmall1122_rs?characterEncoding=utf-8&useSSL=false","root","123123")        val stat: Statement = conn.createStatement        println(sql)        val rs: ResultSet = stat.executeQuery(sql )        val md: ResultSetMetaData = rs.getMetaData        while (  rs.next ) {           val rowData = new JSONObject();          for (i  <-1 to md.getColumnCount  ) {              rowData.put(md.getColumnName(i), rs.getObject(i))          }          resultList+=rowData        }        stat.close()        conn.close()        resultList.toList  }}

import java.utilimport com.alibaba.fastjson.JSONObjectimport org.apache.kafka.common.TopicPartitionimport redis.clients.jedis.Jedisobject OffsetManagerM {  /**    * 从Mysql中读取偏移量    * @param groupId    * @param topic    * @return    */  def getOffset(groupId:String,topic:String):Map[TopicPartition,Long]={      var offsetMap=Map[TopicPartition,Long]()      val jedisClient: Jedis = RedisUtil.getJedisClient      val redisOffsetMap: util.Map[String, String] = jedisClient.hgetAll("offset:"+groupId+":"+topic)      val offsetJsonObjList: List[JSONObject] = MysqlUtil.queryList("SELECT  group_id ,topic,partition_id  , topic_offset  FROM offset_2020 where group_id='"+groupId+"' and topic='"+topic+"'")      jedisClient.close()      if(offsetJsonObjList!=null&&offsetJsonObjList.size==0){            null      }else {            val kafkaOffsetList: List[(TopicPartition, Long)] = offsetJsonObjList.map { offsetJsonObj  =>             (new TopicPartition(offsetJsonObj.getString("topic"),offsetJsonObj.getIntValue("partition_id")), offsetJsonObj.getLongValue("topic_offset"))           }           kafkaOffsetList.toMap      }   } }

创建专用保存数据结果的数据库

create database gmall1122_rs

用于保存偏移量

CREATE TABLE `offset_1122` (

  `group_id` varchar(200) NOT NULL,

  `topic` varchar(200) NOT NULL,

  `partition_id` int(11) NOT NULL,

  `topic_offset` bigint(20) DEFAULT NULL,

  PRIMARY KEY (`group_id`,`topic`,`partition_id`)

) ENGINE=InnoDB DEFAULT CHARSET=utf8

用户保存商品聚合结果

CREATE TABLE `spu_order_final_detail_amount_stat` ( stat_time datetime  ,spu_id varchar(20) ,spu_name  varchar(200),amount decimal(16,2) ,

  PRIMARY KEY (`stat_time`,`spu_id`,`spu_name`)

  )ENGINE=InnoDB  DEFAULT CHARSET=utf8

import java.text.SimpleDateFormatimport java.util.Dateimport com.alibaba.fastjson.JSONimport com.atguigu.gmall1122.realtime.bean.OrderDetailWideimport com.atguigu.gmall1122.realtime.util.{MyKafkaUtil, OffsetManager, OffsetManagerM}import org.apache.kafka.clients.consumer.ConsumerRecordimport org.apache.kafka.common.TopicPartitionimport org.apache.spark.{SparkConf, rdd}import org.apache.spark.streaming.{Seconds, StreamingContext}import org.apache.spark.streaming.dstream.{DStream, InputDStream}import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}import scalikejdbc.{DB, SQL}import scalikejdbc.config.DBsobject SpuAmountSumApp {  def main(args: Array[String]): Unit = {    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("ads_spu_amount_sum_app")    val ssc = new StreamingContext(sparkConf, Seconds(5))    val topic = "DWS_ORDER_DETAIL_WIDE";    val groupId = "ads_spu_amount_sum_group"    /  偏移量处理///    val offset: Map[TopicPartition, Long] = OffsetManagerM.getOffset(groupId, topic)    var inputDstream: InputDStream[ConsumerRecord[String, String]] = null    // 判断如果从redis中读取当前最新偏移量 则用该偏移量加载kafka中的数据  否则直接用kafka读出默认最新的数据    if (offset != null && offset.size > 0) {      inputDstream = MyKafkaUtil.getKafkaStream(topic, ssc, offset, groupId)    } else {      inputDstream = MyKafkaUtil.getKafkaStream(topic, ssc, groupId)    }    //取得偏移量步长    var offsetRanges: Array[OffsetRange] = null    val inputGetOffsetDstream: DStream[ConsumerRecord[String, String]] = inputDstream.transform { rdd =>      offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges      rdd    }    val orderDstreamDetailWideDstream: DStream[OrderDetailWide] = inputGetOffsetDstream.map { record =>      val jsonStr: String = record.value()      val orderDetailWide: OrderDetailWide = JSON.parseObject(jsonStr, classOf[OrderDetailWide])      orderDetailWide    }    val orderWideWithSpuDstream: DStream[(String, Double)] = orderDstreamDetailWideDstream.map(orderWide=>(orderWide.spu_id+":"+orderWide.spu_name,orderWide.final_detail_amount))    val spuAmountDstream: DStream[(String, Double)] = orderWideWithSpuDstream.reduceByKey(_+_)    spuAmountDstream.foreachRDD { rdd =>      val resultArr: Array[(String, Double)] = rdd.collect()      if (resultArr != null && resultArr.size > 0) {        DBs.setup()        DB.localTx(implicit session => {          val dateTime: String = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date())          for ((spu, amount) <- resultArr) {            val spuArr: Array[String] = spu.split(":")            val spuId: String = spuArr(0)            val spuName: String = spuArr(1)            SQL("INSERT INTO spu_order_final_detail_amount_stat(stat_time,spu_id, spu_name, amount) VALUES (?,?,?,?)").bind(dateTime, spuId, spuName, amount).update().apply()          }          throw new RuntimeException("测试异常！！")          for (offset <- offsetRanges) {            //主键相同替换 主键不同新增            SQL("replace INTO offset_2020(group_id,topic, partition_id, topic_offset) VALUES (?,?,?,?)").bind(groupId, topic, offset.partition, offset.untilOffset).update().apply()          }        }        )      }    }    ssc.start()    ssc.awaitTermination()  }}

db.default.driver="com.mysql.jdbc.Driver"db.default.url="jdbc:mysql://hadoop2/gmall1122_rs?characterEncoding=utf-8&useSSL=false"db.default.user="root"db.default.password="123123"

DBs.setup()

DB.localTx(implicit session => {         SQL("INSERT INTO spu_order_final_detail_amount_stat(stat_time,spu_id, spu_name, amount) VALUES (?,?,?,?)").bind(dateTime, spuId, spuName, amount).update().apply()              SQL("replace INTO offset_2020(group_id,topic, partition_id, topic_offset) VALUES (?,?,?,?)").bind(groupId, topic, offset.partition, offset.untilOffset).update().apply()               })

xml version="1.0" encoding="UTF-8"?>  4.0.0      gmall2019_dw    com.atguigu.gmall2019.dw    1.0-SNAPSHOT    com.atguigu.gmall2019.dw.publisher  dw-publisher  0.0.1-SNAPSHOT  dw-publisher  Demo project for Spring Boot      1.8                org.springframework.boot       spring-boot-starter-web                   com.atguigu.gmall2019.dw       dw-common       1.0-SNAPSHOT               org.springframework.boot       spring-boot-starter-test       test    

server.port=8070

logging.level.root=error

spring.datasource.driver-class-name= com.mysql.jdbc.Driver

spring.datasource.url= jdbc:mysql://hadoop2/gmall1122_rs?characterEncoding=utf-8&useSSL=false

spring.datasource.data-username=rootspring.datasource.data-password=123123# mybatismybatis.mapperLocations=classpath:mapper/*.xmlmybatis.configuration.map-underscore-to-camel-case=true

控制层

PublisherController

实现接口的web发布

服务层

MySQLService

数据业务查询interface

MySQLServiceImpl

业务查询的实现类

数据层

TrademarkAmountSumMapper

数据层查询的interface

TrademarkAmountSum.xml

数据层查询的实现配置,实质上是Mapper接口的“实现类”。

主程序

GmallPublisherApplication

增加扫描包

@SpringBootApplication@MapperScan(basePackages = "com.atguigu.gmallXXXXXXX.publisher.mapper")public class Gmall2019PublisherApplication{  public static void main(String[] args) {     SpringApplication.run(Gmall2019PublisherApplication.class, args);  }}

package com.atguigu.gmall1122.publisher.controller;import com.alibaba.fastjson.JSON;import com.atguigu.gmall1122.publisher.service.MysqlService;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.web.bind.annotation.GetMapping;import org.springframework.web.bind.annotation.RequestParam;import org.springframework.web.bind.annotation.RestController;import java.util.List;import java.util.Map;@RestControllerpublic class DataVController {    //路径和参数随便定 ，但是返回值要看datav的需要    @Autowired    MysqlService mysqlService;    @GetMapping("trademark-sum")    public String trademarkSum(@RequestParam("start_date") String startDate, @RequestParam("end_date") String endDate){         if(startDate.length()==0||  endDate.length()==0){                return "参数不能为空！";         }        startDate = startDate.replace("_", " ");        endDate = endDate.replace("_", " ");        List

public interface MysqlService {    public List

@Servicepublic class MysqlServiceImpl implements MysqlService {    @Autowired    TrademarkAmountSumMapper trademarkAmountSumMapper;    @Override    public List

public interface TrademarkAmountSumMapper {    public List

xml version="1.0" encoding="UTF-8"?>>             select tm_name ,sum(amount) amount from `trademark_amount_sum_stat`        where stat_time >= #{start_Date}  and stat_time < #{end_Date}        group by tm_id ,tm_name        order by sum(amount) desc            

@GetMapping("trademark-sum")public String trademarkSum(@RequestParam("start_date") String startDate, @RequestParam("end_date") String endDate){     if(startDate.length()==0||  endDate.length()==0){            return "参数不能为空！";     }    startDate = startDate.replace("_", " ");    endDate = endDate.replace("_", " ");    List