尚硅谷大数据项目之实时分析系统--灵活分析(作者：尚硅谷大数据研发部)版本：V 1.6第1章需求分析1.1 灵活查询的场景数仓中存储了大量的明细数据，但是hadoop存储的数仓计算必须经过mr ，所以即时交互性非常糟糕。
为了方便数据分析人员查看信息，数据平台需要提供一个能够根据文字及选项等条件，进行灵活分析判断的数据功能。
2.2 需求详细输入参数返回结果第2章架构分析2.1 T+1 模式2.1.1 实现步骤1）利用sqoop等工具，从业务数据库中批量抽取数据；2）利用数仓作业，在dws层组织宽表（用户购买行为）；3）开发spark的批处理任务，把dws层的宽表导入到ES中；4）从ES读取数据发布接口，对接可视化模块。
2.1.2 特点优点：可以利用在离线作业处理好的dws层宽表，直接导出一份到ES进行快速交互的分析。
缺点：因为要用离线处理的后的结果在放入ES,所以时效性等同于离线数据。
2.2 T+0 模式2.2.1 实现步骤1）利用canal抓取对应的数据表的实时新增变化数据,推送到Kafka；2）在spark-streaming中进行转换，过滤，关联组合成宽表的结构；3）保存到ES中；4）从ES读取数据发布接口，对接可视化模块。
2.2.2 特点优点：实时产生数据，时效性非常高。
缺点：因为从kafka中得到的是原始数据，所以要利用spark-streaming要进行加工处理，相对来说要比批处理方式麻烦，比如join操作。
第3章实时采集数据3.1 在canal 模块中增加要追踪的表代码public class CanalHandler {private List<CanalEntry.RowData> rowDatasList;String tableName;CanalEntry.EventType eventType;public CanalHandler(List<CanalEntry.RowData> rowDatasList, String tableName, CanalEntry.EventType eventType) {this.rowDatasList = rowDatasList;this.tableName = tableName;this.eventType = eventType;}//根据不同业务的类型发送不同主题public void handle(){if(eventType.equals(CanalEntry.EventType.INSERT)&&tableName.equals("order_info")){sendRowList2Kafka(GmallConstants.KAFKA_TOPIC_ORDER); }elseif((eventType.equals(CanalEntry.EventType.INSERT)||eventType.equals(CanalEntry.EventType.UPDATE))&&tableName.equals("user_info")){ sendRowList2Kafka(GmallConstants.KAFKA_TOPIC_USER); }elseif(eventType.equals(CanalEntry.EventType.INSERT)&&tableName.equals("order_detail")){sendRowList2Kafka(GmallConstants.KAFKA_TOPIC_ORDER_DETAIL);}}// 统一处理发送kafkaprivate void sendRowList2Kafka(String kafkaTopic){for (CanalEntry.RowData rowData : rowDatasList) {List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();JSONObject jsonObject = new JSONObject();for (CanalEntry.Column column : afterColumnsList) {System.out.println(column.getName()+"--->"+column.getValue());jsonObject.put(column.getName(),column.getValue()); }try {Thread.sleep(new Random().nextInt(5)*1000);} catch (InterruptedException e) {e.printStackTrace();}MyKafkaSender.send(kafkaTopic,jsonObject.toJSONString());}}}第4章实时数据处理4.1 数据处理流程4.2 双流join（难点）4.2.1 程序流程图4.2.2 代码4.2.2.1 样例类OrderDetailcase class OrderDetail( id:String ,order_id: String,sku_name: String,sku_id: String,order_price: String,img_url: String,sku_num: String) SaleDetailimport java.text.SimpleDateFormatimport java.utilcase class SaleDetail(var order_detail_id:String =null, var order_id: String=null,var order_status:String=null,var create_time:String=null,var user_id: String=null,var sku_id: String=null,var user_gender: String=null,var user_age: Int=0,var user_level: String=null,var sku_price: Double=0D,var sku_name: String=null,var dt:String=null) {def this(orderInfo:OrderInfo,orderDetail: OrderDetail) {thismergeOrderInfo(orderInfo)mergeOrderDetail(orderDetail)}def mergeOrderInfo(orderInfo:OrderInfo): Unit ={if(orderInfo!=null){this.order_id=orderInfo.idthis.order_status=orderInfo.order_statusthis.create_time=orderInfo.create_timethis.dt=orderInfo.create_dateer_id=er_id}}def mergeOrderDetail(orderDetail: OrderDetail): Unit ={if(orderDetail!=null){this.order_detail_id=orderDetail.idthis.sku_id=orderDetail.sku_idthis.sku_name=orderDetail.sku_namethis.sku_price=orderDetail.order_price.toDouble}}def mergeUserInfo(userInfo: UserInfo): Unit ={if(userInfo!=null){er_id=userInfo.idval formattor = new SimpleDateFormat("yyyy-MM-dd")val date: util.Date = formattor.parse(userInfo.birthday) val curTs: Long = System.currentTimeMillis()val betweenMs= curTs-date.getTimeval age=betweenMs/1000L/60L/60L/24L/365Ler_age= age.toInter_gender=userInfo.genderer_level=er_level}}}UserInfocase class UserInfo(id:String ,login_name:String,user_level:String,birthday:String,gender:String)4.2.2.2 SaleAppdef main(args: Array[String]): Unit = {val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("sale_app")val ssc = new StreamingContext(sparkConf,Seconds(5))val inputOrderDstream: InputDStream[ConsumerRecord[String, String]] = MyKafkaUtil.getKafkaStream(GmallConstant.KAFKA_TOPIC_ORDER,ssc)val inputOrderDetailDstream: InputDStream[ConsumerRecord[String, String]] = MyKafkaUtil.getKafkaStream(GmallConstant.KAFKA_TOPIC_ORDER_DETAIL ,ssc)//整理转换val orderInfoDstream: DStream[OrderInfo] = inputOrderDstream.map { record =>val jsonString: String = record.value()// 1 转换成case classval orderInfo: OrderInfo = JSON.parseObject(jsonString, classOf[OrderInfo])// 2 脱敏电话号码 1381*******val telTuple: (String, String) = orderInfo.consignee_tel.splitAt(4)orderInfo.consignee_tel = telTuple._1 + "*******"// 3 补充日期字段val datetimeArr: Array[String] = orderInfo.create_time.split(" ")orderInfo.create_date = datetimeArr(0) //日期val timeArr: Array[String] = datetimeArr(1).split(":")orderInfo.create_hour = timeArr(0) //小时orderInfo}val orderDetailDStream: DStream[OrderDetail] = inputOrderDetailDstream.map { record =>val jsonString: String = record.value()val orderDetail: OrderDetail = JSON.parseObject(jsonString, classOf[OrderDetail])orderDetail}// 双流join 前要把流变为kv结构val orderInfoWithKeyDstream: DStream[(String, OrderInfo)] = orderInfoDstream.map(orderInfo =>(orderInfo.id,orderInfo))val orderDetailWithKeyDstream: DStream[(String, OrderDetail)] = orderDetailDStream.map(orderDetail=>(orderDetail.order_id,orderDe tail))//为了不管是否能够关联左右，都要保留左右两边的数据采用full joinval fullJoinDStream: DStream[(String, (Option[OrderInfo], Option[OrderDetail]))] = orderInfoWithKeyDstream.fullOuterJoin(orderDetailWithKeyDstream) val saleDetailDstream: DStream[SaleDetail] =fullJoinDStream.mapPartitions { partitionItr =>val jedis: Jedis = RedisUtil.getJedisClientimplicit val formats = org.json4s.DefaultFormatsval saleDetailList = ListBuffer[SaleDetail]()for ((orderId, (orderInfoOption, orderDetailOption)) <- partitionItr) {if (orderInfoOption != None) {println(" 主表有数据！")val orderInfo: OrderInfo = orderInfoOption.get// 1 组合关联if (orderDetailOption != None) {println(" 主表有数据！且从表有数据成功关联")val orderDetail: OrderDetail = orderDetailOption.get//组合成一个SaleDetailval saleDetail = new SaleDetail(orderInfo, orderDetail)// 存放到sale集合中saleDetailList += saleDetail}//2写缓存key 类型: string key名[order_info:order_id] value -> orderinfoJsonprintln(" 主表有数据！写入缓存")val orderInfoKey = "order_info:" + orderId// fastjson无法转换 case class 为json// val orderInfoJson: String = JSON.toJSONString(orderInfo) // json4sval orderInfoJson: String = Serialization.write(orderInfo)jedis.setex(orderInfoKey, 300, orderInfoJson)// 3 查询缓存val orderDetailKey = "order_detail:" + orderIdval orderDetailJson: String = jedis.get(orderDetailKey)val orderDetailSet: util.Set[String] = jedis.smembers(orderDetailKey)import collection.JavaConversions._for ( orderDetailJson <- orderDetailSet ) {println(" 查询到从表缓存数据进行关联")val orderDetail: OrderDetail = JSON.parseObject(orderDetailJson, classOf[OrderDetail])val saleDetail = new SaleDetail(orderInfo, orderDetail)saleDetailList += saleDetail}} else if (orderDetailOption != None) { //主表没有数据从表有数据 println("主表没有数据从表有数据 ")val orderDetail: OrderDetail = orderDetailOption.get//1 查询缓存查询主表println("查询主表缓存")val orderInfoKey = "order_info:" + orderIdval orderInfoJson: String = jedis.get(orderInfoKey)if (orderInfoJson != null && orderInfoJson.size > 0) {val orderInfo: OrderInfo = JSON.parseObject(orderInfoJson, classOf[OrderInfo])val saleDetail = new SaleDetail(orderInfo, orderDetail)saleDetailList += saleDetail}// 2 从表写缓存 // 从表缓存设计问题 //要体现一个主表下多个从表的结构1：n keytype: set key order_detail:order_id members -> 多个order_detailjsonprintln("写从表缓存")val orderDetailKey = "order_detail:" + orderIdval orderDetailJson: String = Serialization.write(orderDetail)jedis.sadd(orderDetailKey,orderDetailJson)jedis.expire(orderDetailKey,300)//jedis.setex(orderDetailKey, 300, orderDetailJson)}}jedis.close()saleDetailList.toIterator}saleDetailDstream.foreachRDD{rdd=>println(rdd.collect().mkString("\n"))}ssc.start()ssc.awaitTermination()}4.2.2.3 样例类转换成为JSON字符串pom.xml<dependency><groupId>org.json4s</groupId><artifactId>json4s-native_2.11</artifactId><version>3.5.4</version></dependency>import org.json4s.native.Serializationimplicit val formats=org.json4s.DefaultFormatsval orderInfoJson: String = Serialization.write(orderInfo)4.3 采集userInfo进入缓存val inputUserDstream: InputDStream[ConsumerRecord[String, String]] = MyKafkaUtil.getKafkaStream(GmallConstants.KAFKA_TOPIC_USER,ssc)// 把userInfo 保存到缓存中inputUserDstream.map{record=>val userInfo: UserInfo = JSON.parseObject(record.value(), classOf[UserInfo])userInfo}.foreachRDD{rdd:RDD[UserInfo]=>val userList: List[UserInfo] = rdd.collect().toListval jedis: Jedis = RedisUtil.getJedisClientimplicit val formats=org.json4s.DefaultFormatsfor (userInfo <- userList ) { // string set list hash zset//设计user_info redis type hash key user_info , fielduser_id ,value user_info_jsonval userkey="user_info"val userJson: String = Serialization.write(userInfo)jedis.hset(userkey,userInfo.id,userJson)}jedis.close()}4.4 反查缓存并关联userInfoval fullSaleDetailDstream: DStream[SaleDetail] = saleDetailDstream.mapPartitions { saleIter =>val jedis: Jedis = RedisUtil.getJedisClientval userList: ListBuffer[SaleDetail] = ListBuffer[SaleDetail]() for (saleDetail <- saleIter) {val userInfoJson: String = jedis.hget("user_info", er_id)val userinfo: UserInfo = JSON.parseObject(userInfoJson, classOf[UserInfo])saleDetail.mergeUserInfo(userinfo)userList += saleDetail}jedis.close()userList.toIterator}4.5 保存购买明细进入ES中4.5.1 ES索引建立PUT gmall190408_sale_detail{"mappings" : {"_doc" : {"properties" : {"order_detail_id" : {"type" : "keyword"},"order_id" : {"type" : "keyword"},"create_time" : {"type" : "date","format" : "yyyy-MM-dd HH:mm:ss"},"dt" : {"type" : "date"},"order_status" : {"type" : "keyword"},"sku_id" : {"type" : "keyword"},"sku_name" : {"type" : "text","analyzer": "ik_max_word"},"sku_price" : {"type" : "float"},"user_age" : {"type" : "long"},"user_gender" : {"type" : "keyword"},"user_id" : {"type" : "keyword"},"user_level" : {"type" : "keyword","index" : false}}}}}4.5.2 保存ES代码fullSaleDetailDstream.foreachRDD{rdd=>val saleDetailList: List[SaleDetail] = rdd.collect().toListval saleDetailWithKeyList: List[(String, SaleDetail)] = saleDetailList.map(saleDetail=>(saleDetail.order_detail_id,saleDe tail))MyEsUtil.insertBulk(GmallConstants.ES_INDEX_SALE_DETAIL,saleDetai lWithKeyList)}第5章灵活查询数据接口开发5.1 传入路径及参数5.2 返回值5.3 编写DSL语句GET gmall190408_sale_detail/_search {"query": {"bool": {"filter": {"term": {"dt": "2019-02-14"}},"must": [{"match":{"sku_name": {"query": "小米手机","operator": "and"}}}]}} ,"aggs": {"groupby_age": {"terms": {"field": "user_age"}}},"from": 0,"size": 2,}5.4 代码开发5.4.1 代码清单5.4.2 pom.xml<!--- ES依赖包--><dependency><groupId>io.searchbox</groupId><artifactId>jest</artifactId><version>5.3.3</version></dependency><dependency><groupId>net.java.dev.jna</groupId><artifactId>jna</artifactId><version>4.5.2</version></dependency><dependency><groupId>org.codehaus.janino</groupId><artifactId>commons-compiler</artifactId><version>2.7.8</version></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-elasticsearch</artifactId> </dependency>5.4.3 配置application.properties#esspring.elasticsearch.jest.uris=http://hadoop102:92005.4.4 PublisherServiceImpl@Overridepublic Map getSaleDetail(String date, String keyword, int pageSize, int pageNo) {SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();//过滤匹配BoolQueryBuilder boolQueryBuilder = new BoolQueryBuilder();boolQueryBuilder.filter(new TermQueryBuilder("dt",date));boolQueryBuilder.must(newMatchQueryBuilder("sku_name",keyword).operator(MatchQueryBuilder. Operator.AND));searchSourceBuilder.query(boolQueryBuilder);// 性别聚合TermsBuilder genderAggs = AggregationBuilders.terms("groupby_user_gender").field("user_gend er").size(2);searchSourceBuilder.aggregation(genderAggs);// 年龄聚合TermsBuilder ageAggs =AggregationBuilders.terms("groupby_user_age").field("user_age").s ize(100);searchSourceBuilder.aggregation(ageAggs);// 行号= （页面-1） * 每页行数searchSourceBuilder.from((pageNo-1)*pageSize);searchSourceBuilder.size(pageSize);System.out.println(searchSourceBuilder.toString());Search search = new Search.Builder(searchSourceBuilder.toString()).addIndex(GmallCons tant.ES_INDEX_SALE_DETAIL).addType("_doc").build();Map resultMap=new HashMap(); //需要总数，明细，2个聚合的结果try {SearchResult searchResult = jestClient.execute(search);//总数Long total = searchResult.getTotal();//明细List<SearchResult.Hit<Map, Void>> hits = searchResult.getHits(Map.class);List<Map> saleDetailList=new ArrayList<>();for (SearchResult.Hit<Map, Void> hit : hits) {saleDetailList.add(hit.source) ;}//年龄聚合结果Map ageMap=new HashMap();List<TermsAggregation.Entry> buckets = searchResult.getAggregations().getTermsAggregation("groupby_user_ age").getBuckets();for (TermsAggregation.Entry bucket : buckets) {ageMap.put(bucket.getKey(),bucket.getCount());}//性别聚合结果Map genderMap=new HashMap();List<TermsAggregation.Entry> genderbuckets = searchResult.getAggregations().getTermsAggregation("groupby_user_ gender").getBuckets();for (TermsAggregation.Entry bucket : genderbuckets) {genderMap.put(bucket.getKey(),bucket.getCount());}resultMap.put("total",total);resultMap.put("list",saleDetailList);resultMap.put("ageMap",ageMap);resultMap.put("genderMap",genderMap);} catch (IOException e) {e.printStackTrace();}return resultMap;}5.4.5 Bean@Data@AllArgsConstructorpublic class Option {String name;Double value;}@Data@AllArgsConstructorpublic class Stat {String title;List<Option> options;}需要pom.xml增加依赖<dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><optional>true</optional></dependency>5.4.6 PublisherController@GetMapping("sale_detail")public String getSaleDetail(@RequestParam("date")String date ,@RequestParam("startpage") int startpage,@RequestParam("size") int size,@RequestParam("keyword")String keyword){Map saleMap = publisherService.getSaleDetail(date, keyword, startpage, size);Long total = (Long)saleMap.get("total");List<Map> saleDetailList = (List)saleMap.get("detail");Map ageMap =(Map) saleMap.get("ageMap");Map genderMap =(Map) saleMap.get("genderMap");// genderMap 整理成为 OptionGroupLong femaleCount =(Long) genderMap.get("F");Long maleCount =(Long) genderMap.get("M");double femaleRate = Math.round(femaleCount * 1000D / total) / 10D;double maleRate = Math.round(maleCount * 1000D / total) / 10D; List<Option> genderOptions = new ArrayList<>();genderOptions.add( new Option("男", maleRate));genderOptions.add( new Option("女", femaleRate));OptionGroup genderOptionGroup = new OptionGroup("性别占比", genderOptions);// ageMap 整理成为 OptionGroupLong age_20Count=0L;Long age20_30Count=0L;Long age30_Count=0L;for (Object o : ageMap.entrySet()) {Map.Entry entry = (Map.Entry) o;String agekey =(String) entry.getKey();int age = Integer.parseInt(agekey);Long ageCount =(Long) entry.getValue();if(age <20){age_20Count+=ageCount;}else if(age>=20&&age<30){age20_30Count+=ageCount;}else{age30_Count+=ageCount;}}Double age_20rate=0D;Double age20_30rate=0D;Double age30_rate=0D;age_20rate = Math.round(age_20Count * 1000D / total) / 10D;age20_30rate = Math.round(age20_30Count * 1000D / total) / 10D; age30_rate = Math.round(age30_Count * 1000D / total) / 10D;List<Option> ageOptions=new ArrayList<>();ageOptions.add(new Option("20岁以下",age_20rate));ageOptions.add(new Option("20岁到30岁",age20_30rate));ageOptions.add(new Option("30岁以上",age30_rate));OptionGroup ageOptionGroup = new OptionGroup("年龄占比", ageOptions);List<OptionGroup> optionGroupList=new ArrayList<>();optionGroupList.add(genderOptionGroup);optionGroupList.add(ageOptionGroup);SaleInfo saleInfo = new SaleInfo(total, optionGroupList, saleDetailList);return JSON.toJSONString(saleInfo);}5.5 对接可视化模块。
