实时--1.1 日志

  <dependency>
       <groupId>org.springframework.boot</groupId>
       <artifactId>spring-boot-starter</artifactId>
       <exclusions>
         <exclusion>
           <groupId>org.springframework.boot</groupId>
           <artifactId>spring-boot-starter-logging</artifactId>
         </exclusion>
       </exclusions>
    </dependency>
    <dependency>
       <groupId>org.springframework.boot</groupId>
       <artifactId>spring-boot-starter-log4j</artifactId>
       <version>1.3.8.RELEASE</version>
    </dependency>
  </dependencies>

com.atguigu.gmall.logger.controller.LoggerController

@RestController
public class LoggerController {
    @Autowired
    KafkaTemplate kafkaTemplate;
    private static final Logger logger = LoggerFactory.getLogger(LoggerController.class);
    @PostMapping("log")
    public String doLog(@RequestParam("log") String log){

        JSONObject jsonObject = JSON.parseObject(log);
        jsonObject.put("ts", System.currentTimeMillis());

        //System.out.println(log);

        // 落盘成为日志文件     // log4j
        logger.info(jsonObject.toJSONString());
        //发送kafka
        if ("startup".equals(jsonObject.getString("type"))){
            kafkaTemplate.send(GmallConstant.KAFKA_TOPIC_STARTUP, jsonObject.toJSONString());
        }else {
            kafkaTemplate.send(GmallConstant.KAFKA_TOPIC_EVENT, jsonObject.toJSONString());
        }
        return "success";
    }
}

log4j.appender.atguigu.MyConsole=org.apache.log4j.ConsoleAppender  //怎么写这个日志；类型--控制台
log4j.appender.atguigu.MyConsole.target=System.err  //控制台有两种：System.out日志颜色黑色和System.err日志是红色的
log4j.appender.atguigu.MyConsole.layout=org.apache.log4j.PatternLayout    //自定义的，除了要打印的日志级别，还要打印什么
log4j.appender.atguigu.MyConsole.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss} %10p (%c:%M) - %m%n  //格式，p是日志级别，%m输出的内容，%n是换行；

log4j.appender.atguigu.File=org.apache.log4j.DailyRollingFileAppender  //每日滚动文件，每天产生一个文件；
log4j.appender.atguigu.File.file=/opt/module/applog/gmall/log/app.log  //输出的文件路径
log4j.appender.atguigu.File.DatePattern=‘.‘yyyy-MM-dd  　　　　　　//输出文件的后缀； 当天的日志是没有后缀的，一旦过了12点，就有后缀.‘yyyy-MM-dd‘，后缀是日志时间
log4j.appender.atguigu.File.layout=org.apache.log4j.PatternLayout //自定义格式
log4j.appender.atguigu.File.layout.ConversionPattern=%m%n  //要干干净净的打印信息；

log4j.logger.com.atguigu.gmall.logger.controller.LoggerController=info,atguigu.File  
//某一个类的路径，只监控某个类所产生的日志；log4j.rootLogger=error,atguigu.Myconsole表示根底的，除了上边指定的都是它，首先是精确匹配到info就日志输出就按照它的打印，它们后边的.File或.Myconsole都会输出

object DauApp {
      def main(args: Array[String]): Unit = {
        val conf: SparkConf = new SparkConf().setAppName("gmall").setMaster("local[*]")
        val streamingContext: StreamingContext = new StreamingContext(new SparkContext(conf),Seconds(5))

        val inputStream: InputDStream[ConsumerRecord[String, String]] = MyKafkaUtil.getKafkaStream(GmallConstant.KAFKA_TOPIC_STARTUP, streamingContext)
        //  1 把当日已访问过的用户保存起来 redis
        //  2  以当日已访问用户清单为依据 ，过滤掉再次访问的请求

        // 转换case class  补全日期格式
        val startupLogDStream: DStream[StartUpLog] = inputStream.map { record =>
          val jsonStr: String = record.value()
          val startUpLog: StartUpLog = JSON.parseObject(jsonStr, classOf[StartUpLog])
          //把日期进行补全
          val dateTimeString: String = new SimpleDateFormat("yyyy-MM-dd HH:mm").format(new Date(startUpLog.ts))
          val dateTimeArray: Array[String] = dateTimeString.split(" ")
          startUpLog.logDate = dateTimeArray(0)
          startUpLog.logHour = dateTimeArray(1).split(":")(0)
          startUpLog.logHourMinute = dateTimeArray(1)
          startUpLog
        }
        // 去重操作
        val filterDStream: DStream[StartUpLog] = startupLogDStream.transform { rdd =>
          //driver 每时间间隔执行一次
          println("过滤前：" + rdd.count())
          val jedis: Jedis = RedisUtil.getJedisClient
          val curDate: String = new SimpleDateFormat("yyyy-MM-dd").format(new Date())
          val key: String = "dau:" + curDate
          val dauSet: util.Set[String] = jedis.smembers(key) //当日日活用户清单
        //使用广播变量
        val dauBC: Broadcast[util.Set[String]] = streamingContext.sparkContext.broadcast(dauSet)

          val filterRDD: RDD[StartUpLog] = rdd.filter {
            startuplog =>
              !dauBC.value.contains(startuplog.mid)
          }
          println("过滤后:" + filterRDD.count())
          filterRDD
        }

        // 考虑到 新的访问可能会出现重复 ，所以以mid为key进行去重，每个mid为小组 每组取其中一个
        val startupLogGroupDStream: DStream[(String, Iterable[StartUpLog])] = filterDStream.map{startuplog => (startuplog.mid, startuplog)}.groupByKey()

        val startupLogFilterDistinctDStream: DStream[StartUpLog] = startupLogGroupDStream.flatMap {
          case (mid, startupLogIter) =>
            val startupLogOneIter: Iterable[StartUpLog] = startupLogIter.take(1)
            startupLogOneIter
        }

        //  1 把当日已访问过的用户保存到 redis
        startupLogFilterDistinctDStream.foreachRDD{rdd =>
          rdd.foreachPartition{startupLogItr =>
            val jedis: Jedis = RedisUtil.getJedisClient
            val startupList: List[StartUpLog] = startupLogItr.toList
            for (elem <- startupList) {
              val key: String = "dau:" + elem.logDate
              jedis.sadd(key, elem.mid)
            }
            jedis.close()
            //保存到ES
            MyEsUtil.insertEsBatch(GmallConstant.ES_INDEX_DAU, startupList)
          }
    }
    streamingContext.start()
    streamingContext.awaitTermination()

  }

}

##############在ES中创建index
PUT gmall_dau
{
  "mappings": {
    "_doc":{
      "properties":{
         "mid":{
           "type":"keyword" ,
         },
         "uid":{
           "type":"keyword"
         },
         "area":{
           "type":"keyword"
         },
         "os":{
           "type":"keyword"
         },
         "ch":{
           "type":"keyword"
         },
         "vs":{
           "type":"keyword"
         },
         "logDate":{
           "type":"keyword"
         },
         "logHour":{
           "type":"keyword"
         },
         "logHourMinute":{
           "type":"keyword"
         },
         "ts":{
           "type":"long"
         } 
      }
    }
  }
}

GET /gmall_dau/_search
{
  "query": {
    "bool": {
      "filter": {
        "term": {
          "logDate": "2019-05-04"
        }
      }
    }
  }
}

######groupby操作  聚合aggregation
GET /gmall_dau/_search
{
  "query": {
    "bool": {
      "filter": {
        "term": {
          "logDate": "2019-04-30"
        }
      }
    }
  },
  "aggs": {
    "groupby_logHour": {
      "terms": {
        "field": "logHour.keyword",
        "size": 24
      }
    }
  }
}