尚硅谷大数据项目之实时项目2(日活需求)

八HDFS HA高可用8.1 HA概述1）所谓HA（high available），即高可用（7*24小时不中断服务）。

2）实现高可用最关键的策略是消除单点故障。

HA严格来说应该分成各个组件的HA机制：HDFS的HA和YARN的HA。

3）Hadoop2.0之前，在HDFS集群中NameNode存在单点故障（SPOF）。

4）NameNode主要在以下两个方面影响HDFS集群NameNode机器发生意外，如宕机，集群将无法使用，直到管理员重启NameNode机器需要升级，包括软件、硬件升级，此时集群也将无法使用HDFS HA功能通过配置Active/Standby两个nameNodes实现在集群中对NameNode的热备来解决上述问题。

如果出现故障，如机器崩溃或机器需要升级维护，这时可通过此种方式将NameNode很快的切换到另外一台机器。

8.2 HDFS-HA工作机制1）通过双namenode消除单点故障8.2.1 HDFS-HA工作要点1）元数据管理方式需要改变：内存中各自保存一份元数据；Edits日志只有Active状态的namenode节点可以做写操作；两个namenode都可以读取edits；共享的edits放在一个共享存储中管理（qjournal和NFS两个主流实现）；2）需要一个状态管理功能模块实现了一个zkfailover，常驻在每一个namenode所在的节点，每一个zkfailover负责监控自己所在namenode节点，利用zk进行状态标识，当需要进行状态切换时，由zkfailover 来负责切换，切换时需要防止brain split现象的发生。

3）必须保证两个NameNode之间能够ssh无密码登录。

4）隔离（Fence），即同一时刻仅仅有一个NameNode对外提供服务8.2.2 HDFS-HA自动故障转移工作机制前面学习了使用命令hdfs haadmin -failover手动进行故障转移，在该模式下，即使现役NameNode已经失效，系统也不会自动从现役NameNode转移到待机NameNode，下面学习如何配置部署HA自动进行故障转移。

尚硅谷大数据之hive尚硅谷大数据之hive》是一本关于Hive技术的书籍。

它旨在全面介绍Hive的主要内容和应用，并针对不同的读者群体提供有用的信息和指导。

Hive是一个开源的数据仓库基础设施工具，它构建在Hadoop 之上，用于处理大规模数据集。

通过使用Hive，用户可以使用类似于SQL的查询语言来访问和分析存储在Hadoop分布式文件系统中的数据。

这使得非技术背景的用户也能够利用Hive进行数据分析和查询。

本书主要包括以下内容：Hive基础知识：介绍Hive的基本概念、架构和组件。

读者将了解Hive如何与Hadoop生态系统中的其他工具集成，并研究如何安装和配置Hive。

Hive数据模型：详细解释Hive的数据模型，包括数据表、分区和桶等概念。

读者将研究如何创建、修改和管理Hive数据表，并了解如何利用分区和桶来提高查询性能。

Hive查询语言：深入介绍HiveQL，这是Hive的查询语言。

读者将研究如何编写各种类型的查询，包括基本的选择、过滤和聚合查询，以及复杂的连接和子查询。

Hive优化和性能调优：提供有关如何优化Hive查询性能的实用技巧和建议。

读者将研究如何使用索引、分区和桶来改善查询速度，以及如何使用适当的配置参数来优化Hive性能。

Hive高级特性：介绍Hive的一些高级特性和扩展，例如动态分区、外部表、UDF和UDAF等。

读者将了解如何利用这些功能来处理具有更复杂需求的数据分析场景。

本书适合各种读者群体，包括数据分析师、数据工程师、数据库管理员和对Hive技术感兴趣的研究者。

无论您是初学者还是有一定经验的专业人士，本书都将为您提供全面且易于理解的Hive研究资源。

2.简要介绍HiveHive是一个基于Hadoop的数据仓库基础架构，用于处理和分析大数据。

它提供了一个类似于SQL的查询语言，称为HiveQL，使用户能够对存储在Hadoop集群中的大规模数据进行查询和分析。

Hive的重要性在于它简化了大数据处理和分析的过程。

尚硅谷大数据技术之Hadoop（生产调优手册）（作者：尚硅谷大数据研发部）版本：V3.3第1章HDFS—核心参数1.1 NameNode内存生产配置1）NameNode内存计算每个文件块大概占用150byte，一台服务器128G内存为例，能存储多少文件块呢？128 * 1024 * 1024 * 1024 / 150Byte ≈9.1亿G MB KB Byte2）Hadoop2.x系列，配置NameNode内存NameNode内存默认2000m，如果服务器内存4G，NameNode内存可以配置3g。

在hadoop-env.sh文件中配置如下。

HADOOP_NAMENODE_OPTS=-Xmx3072m3）Hadoop3.x系列，配置NameNode内存（1）hadoop-env.sh中描述Hadoop的内存是动态分配的# The maximum amount of heap to use (Java -Xmx). If no unit # is provided, it will be converted to MB. Daemons will# prefer any Xmx setting in their respective _OPT variable.# There is no default; the JVM will autoscale based upon machine # memory size.# export HADOOP_HEAPSIZE_MAX=# The minimum amount of heap to use (Java -Xms). If no unit # is provided, it will be converted to MB. Daemons will# prefer any Xms setting in their respective _OPT variable.# There is no default; the JVM will autoscale based upon machine # memory size.# export HADOOP_HEAPSIZE_MIN=HADOOP_NAMENODE_OPTS=-Xmx102400m（2）查看NameNode占用内存[atguigu@hadoop102 ~]$ jps3088 NodeManager2611 NameNode3271 JobHistoryServer2744 DataNode3579 Jps[atguigu@hadoop102 ~]$ jmap -heap 2611Heap Configuration:MaxHeapSize = 1031798784 (984.0MB)（3）查看DataNode占用内存[atguigu@hadoop102 ~]$ jmap -heap 2744Heap Configuration:MaxHeapSize = 1031798784 (984.0MB)查看发现hadoop102上的NameNode和DataNode占用内存都是自动分配的，且相等。

尚硅谷大数据技术之Zookeeper(作者：尚硅谷大数据研发部)版本：V2.0第1章 Zookeeper 入门1.1 概述Zookeeper 是一个开源的分布式的，为分布式应用提供协调服务的Apache 项目。

Zookeeper 工作机制业务功能服务器1Zookeeper 从设计模式角度来理解：是一个基于观察者模式设计的分布式服务管理框架，它负责存储和管理大家都关心的数据，然后接受观察者的注册，一旦这些数据的状态发生变化，Zookeeper 就将负责通知已经在Zookeeper 上注册的那些观察者做出相应的反应。

业务功能服务器2业务功能服务器3客户端2客户端3Zookeeper 集群1服务端启动时去注册信息（创建都是临时节点）/servers/server1hadoop10180nodes/server2hadoop10290nodes /server3hadoop10395nodes2获取到当前在线服务器列表，并且注册监听3服务器节点下线4服务器节点上下线事件通知5process(){重新再去获取服务器列表，并注册监听}客户端1Zookeeper=文件系统+通知机制1.2 特点1）Zookeeper ：一个领导者（Leader ），多个跟随者（Follower ）组成的集群。

2）集群中只要有半数以上节点存活，Zookeeper 集群就能正常服务。

3）全局数据一致：每个Server 保存一份相同的数据副本，Client 无论连接到哪个Server ，数据都是一致的。

4）更新请求顺序进行，来自同一个Client 的更新请求按其发送顺序依次执行。

5）数据更新原子性，一次数据更新要么成功，要么失败。

6）实时性，在一定时间范围内，Client 能读到最新数据。

Zookeeper 特点1.3 数据结构数据结构ZooKeeper 数据模型的结构与Unix 文件系统很类似，整体上可以看作是一棵树，每个节点称做一个ZNode 。

尚硅谷数据仓库实战之3数仓搭建数仓笔记数据仓库和数据集市详解：ODS、DW、DWD、DWM、DWS、ADS：尚硅谷数仓实战之1项目需求及架构设计：尚硅谷数仓实战之2数仓分层+维度建模：尚硅谷数仓实战之3数仓搭建：B站直达：百度网盘：，提取码：yyds 阿里云盘：，提取码：335o第4章数仓搭建-ODS层1）保持数据原貌不做任何修改，起到备份数据的作用。

2）数据采用LZO压缩，减少磁盘存储空间。

100G数据可以压缩到10G以内。

3）创建分区表，防止后续的全表扫描，在企业开发中大量使用分区表。

4）创建外部表。

在企业开发中，除了自己用的临时表，创建内部表外，绝大多数场景都是创建外部表。

4.2 ODS层（业务数据）ODS层业务表分区规划如下在这里插入图片描述ODS层业务表数据装载思路如下在这里插入图片描述4.2.1 活动信息表第5章数仓搭建-DIM层5.1 商品维度表（全量）1.建表语句2.分区规划在这里插入图片描述3.数据装载在这里插入图片描述5.6 用户维度表（拉链表）5.6.1 拉链表概述1）什么是拉链表在这里插入图片描述2）为什么要做拉链表在这里插入图片描述3）如何使用拉链表在这里插入图片描述4）拉链表形成过程在这里插入图片描述5.6.2 制作拉链表1.建表语句2.分区规划在这里插入图片描述3.数据装载在这里插入图片描述1）首日装载拉链表首日装载，需要进行初始化操作，具体工作为将截止到初始化当日的全部历史用户导入一次性导入到拉链表中。

目前的ods_user_info表的第一个分区，即2023-06-14分区中就是全部的历史用户，故将该分区数据进行一定处理后导入拉链表的9999-99-99分区即可。

2）每日装载（1）实现思路在这里插入图片描述第6章数仓搭建-DWD层1）对用户行为数据解析。

2）对业务数据采用维度模型重新建模。

6.1 DWD层（用户行为日志）6.1.1 日志解析思路1）日志结构回顾（1）页面埋点日志在这里插入图片描述（2）启动日志在这里插入图片描述2）日志解析思路在这里插入图片描述6.1.2 json_object函数使用Mysql中也有响应的JSON处理函数，不过性能没有hive高。

尚硅谷新学生培训计划一、培训目的尚硅谷作为一家技术培训机构，一直以来致力于为学生提供优质的技术培训服务，帮助学生快速提升自己的技术水平，顺利就业。

针对即将到来的新学期，为了增强新学生的学习能力和就业竞争力，我们制定了以下新学生培训计划。

二、培训内容1. 入学培训在新学生入学之前，将安排一次入学培训，内容包括尚硅谷的学习体系介绍、学习方法指导，帮助学生明确学习目标，并做好心理准备。

2. 专业课程培训根据学生报名的专业方向，为学生安排相应课程培训，包括但不限于Java开发、前端开发、大数据、人工智能等方向的课程。

课程内容包括理论知识讲解、实际案例分析、项目实战等，以提升学生的专业知识和实际操作能力。

3. 项目实训在专业课程培训结束后，将为学生安排项目实训，让学生在实际项目中应用所学知识，提升实际操作能力和团队协作能力。

4. 就业培训在项目实训结束后，将为学生安排就业指导课程，包括简历制作、面试技巧、职业规划等内容，帮助学生提升就业竞争力，并顺利就业。

5. 毕业证书颁发在学生顺利完成培训并通过考核后，将颁发尚硅谷的毕业证书，为学生提供一份权威的学习证明。

三、培训方式1. 实体课程尚硅谷提供优质的线下课程教学环境，为学生提供舒适的学习场所和专业的师资力量，让学生能够更好地融入学习氛围，深度学习课程内容。

2. 在线课程尚硅谷也提供优质的在线课程教学服务，学生可以通过网络平台进行学习，灵活安排学习时间，更好地掌握课程知识。

3. 项目实训项目实训将以团队合作的方式进行，学生将分组完成实际项目，培养学生的团队协作能力和项目管理能力。

四、培训要求1. 学习积极性学生应保持良好的学习积极性，勤奋学习，认真完成每一堂课程和作业。

2. 团队协作在项目实训环节，学生应积极参与团队合作，配合团队完成项目任务，培养团队合作精神。

3. 过程管理学生应合理规划学习时间，对学习进度进行有效管理，确保按时完成培训计划。

五、培训评估在培训过程中，将对学生的学习情况进行定期评估，包括课堂表现、作业情况、项目实训成果等，对合格的学生进行肯定与奖励，对不合格的学生进行及时的辅导和调整。

尚硅谷大数据技术之ELK(作者：尚硅谷大数据研发部)版本：V1.5第1章Elasticsearch概述1.1 什么是搜索？百度：我们比如说想找寻任何的信息的时候，就会上百度去搜索一下，比如说找一部自己喜欢的电影，或者说找一本喜欢的书，或者找一条感兴趣的新闻（提到搜索的第一印象）。

百度!= 搜索1）互联网的搜索：电商网站，招聘网站，新闻网站，各种app2）IT系统的搜索：OA软件，办公自动化软件，会议管理，日程管理，项目管理。

搜索，就是在任何场景下，找寻你想要的信息，这个时候，会输入一段你要搜索的关键字，然后就期望找到这个关键字相关的有些信息1.2 如果用数据库做搜索会怎么样？用数据库来实现搜索，是不太靠谱的。

通常来说，性能会很差的。

1.3 什么是全文检索和Lucene？1）全文检索，倒排索引全文检索是指计算机索引程序通过扫描文章中的每一个词，对每一个词建立一个索引，指明该词在文章中出现的次数和位置，当用户查询时，检索程序就根据事先建立的索引进行查找，并将查找的结果反馈给用户的检索方式。

这个过程类似于通过字典中的检索字表查字的过程。

全文搜索搜索引擎数据库中的数据。

2）lucene，就是一个jar包，里面包含了封装好的各种建立倒排索引，以及进行搜索的代码，包括各种算法。

我们就用java开发的时候，引入lucene jar，然后基于lucene的api 进行去进行开发就可以了。

1.4 什么是Elasticsearch？Elasticsearch，基于Lucene，隐藏复杂性，提供简单易用的RestfulAPI接口、JavaAPI 接口（还有其他语言的API接口）。

关于Elasticsearch的一个传说，有一个程序员失业了，陪着自己老婆去英国伦敦学习厨师课程。

程序员在失业期间想给老婆写一个菜谱搜索引擎，觉得Lucene实在太复杂了，就开发了一个封装了Lucene的开源项目：Compass。

后来程序员找到了工作，是做分布式的高性能项目的，觉得Compass不够，就写了Elasticsearch，让Lucene变成分布式的系统。

02_尚硅⾕⼤数据技术之Hadoop（⼊门）尚硅⾕⼤数据技术之 Hadoop（⼊门）（作者：尚硅⾕⼤数据研发部）版本：V3.3第 1 章 Hadoop 概述1.1 Hadoop 是什么1） Hadoop是⼀个由Apache基⾦会所开发的分布式系统基础架构。

2）主要解决，海量数据的存储和海量数据的分析计算问题。

3）⼴义上来说，Hadoop通常是指⼀个更⼴泛的概念——Hadoop⽣态圈。

1.2 Hadoop 发展历史（了解）1）Hadoop创始⼈Doug Cutting，为了实现与Google类似的全⽂搜索功能，他在Lucene框架基础上进⾏优化升级，查询引擎和索引引擎。

2）2001年年底Lucene成为Apache基⾦会的⼀个⼦项⽬。

3）对于海量数据的场景，Lucene框架⾯对与Google同样的困难，存储海量数据困难，检索海量速度慢。

4）学习和模仿Google解决这些问题的办法：微型版Nutch。

5）可以说Google是Hadoop的思想之源（Google在⼤数据⽅⾯的三篇论⽂）GFS --->HDFSMap-Reduce --->MRBigTable --->HBase6）2003-2004年，Google公开了部分GFS和MapReduce思想的细节，以此为基础Doug Cutting等⼈⽤了2年业余时间实现了DFS和MapReduce机制，使Nutch性能飙升。

7）2005 年Hadoop 作为 Lucene的⼦项⽬ Nutch的⼀部分正式引⼊Apache基⾦会。

8）2006 年 3 ⽉份，Map-Reduce和Nutch Distributed File System （NDFS）分别被纳⼊到 Hadoop 项⽬中，Hadoop就此正式诞⽣，标志着⼤数据时代来临。

9）名字来源于Doug Cutting⼉⼦的玩具⼤象1.3 Hadoop 三⼤发⾏版本（了解）Hadoop 三⼤发⾏版本：Apache、Cloudera、Hortonworks。

第1章需求分析1.1 简介实时预警，是一种经常出现在实时计算中的业务类型。

根据日志数据中系统报错异常，或者用户行为异常的检测，产生对应预警日志。

预警日志通过图形化界面的展示，可以提醒监控方，需要及时核查问题，并采取应对措施。

1.2 需求说明需求：同一设备，5分钟内三次及以上用不同账号登录并领取优惠劵，并且在登录到领劵过程中没有浏览商品。

达到以上要求则产生一条预警日志。

同一设备，每分钟只记录一次预警。

1.3 预警日志格式第2章整体流程设计2.1 框架流程2.2 开发思路1）从kafka中消费数据，根据条件进行过滤筛选，生成预警日志；2）预警日志保存到ElasticSearch中；3）利用Kibana快速搭建可视化图形界面。

第3章实时计算模块3.1 筛选条件分析同一设备（分组）5分钟内（窗口）三次不同账号登录（用户）领取优惠券（行为）没有浏览商品（行为）同一设备每分钟只记录一次预警（去重）3.2 数据处理流程图3.3 代码开发3.3.1 事件日志样例类– EventInfo case class EventInfo(mid:String,uid:String,appid:String,area:String,os:String,ch:String,`type`:String,evid:String,pgid:String,npgid:String,itemid:String,var logDate:String, var logHour:String,var ts:Long)3.3.2 预警日志样例类–CouponAlertInfocase class CouponAlertInfo(mid:String,uids:java.util.HashSet[String],itemIds:java.util.HashSet[String],events:java.util.List[String],ts:Long)3.3.3 预警业务类– AlertAppimport com.alibaba.fastjson.JSONimport com.atguigu.gmall.constant.GmallConstantsimport com.atguigu.gmall2019.realtime.bean.{CouponAlertInfo, EventInfo}import com.atguigu.gmall2019.realtime.util.{MyEsUtil, MyKafkaUtil} import org.apache.kafka.clients.consumer.ConsumerRecordimport org.apache.spark.SparkConfimport org.apache.spark.streaming.dstream.{DStream, InputDStream} import org.apache.spark.streaming.{Seconds, StreamingContext}import scala.util.control.Breaks._object AlertApp {def main(args: Array[String]): Unit = {val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("event_app")val ssc = new StreamingContext(sparkConf,Seconds(5))val inputDstream: InputDStream[ConsumerRecord[String, String]] = MyKafkaUtil.getKafkaStream(GmallConstants.KAFKA_TOPIC_EVENT,ssc) //1 格式转换成样例类val eventInfoDstream: DStream[EventInfo] = inputDstream.map { record =>val jsonstr: String = record.value()val eventInfo: EventInfo = JSON.parseObject(jsonstr, classOf[EventInfo])eventInfo}//2 开窗口val eventInfoWindowDstream: DStream[EventInfo] = eventInfoDstream.window(Seconds(30),Seconds(5))//3同一设备分组val groupbyMidDstream: DStream[(String, Iterable[EventInfo])] = eventInfoWindowDstream.map(eventInfo=>(eventInfo.mid,eventInfo)). groupByKey()//4 判断预警// 在一个设备之内// 1 三次及以上的领取优惠券 (evid coupon) 且 uid都不相同// 2 没有浏览商品(evid clickItem)val checkCouponAlertDStream: DStream[(Boolean, CouponAlertInfo)] = groupbyMidDstream.map { case (mid, eventInfoItr) =>————————————————————————————— val couponUidsSet = new util.HashSet[String]()val itemIdsSet = new util.HashSet[String]()val eventIds = new util.ArrayList[String]()var notClickItem: Boolean = truebreakable(for (eventInfo: EventInfo <- eventInfoItr) {eventIds.add(eventInfo.evid) //用户行为if (eventInfo.evid == "coupon") {couponUidsSet.add(eventInfo.uid) //用户领券的uiditemIdsSet.add(eventInfo.itemid) //用户领券的商品id} else if (eventInfo.evid == "clickItem") {notClickItem = falsebreak()}})//组合成元组（标识是否达到预警要求，预警信息对象）(couponUidsSet.size() >= 3 && notClickItem, CouponAlertInfo(mid, couponUidsSet, itemIdsSet, eventIds, System.currentTimeMillis()))}//过滤val filteredDstream: DStream[(Boolean, CouponAlertInfo)] = checkCouponAlertDStream.filter{_._1}//增加一个id用于保存到es的时候进行去重操作val alertInfoWithIdDstream: DStream[(String, CouponAlertInfo)] = filteredDstream.map { case (flag, alertInfo) =>val period: Long = alertInfo.ts / 1000L / 60Lval id: String = alertInfo.mid + "_" + period.toString(id, alertInfo)}alertInfoWithIdDstream.foreachRDD{rdd=>rdd.foreachPartition{alertInfoWithIdIter=>MyEsUtil.insertBulk(GmallConstants.ES_INDEX_COUPON_ALERT ,alertIn foWithIdIter.toList)}}ssc.start()ssc.awaitTermination()}}第4章ElasticSearch 的保存4.1 ES集群搭建参考《ElasticSearch集群安装》手册—————————————————————————————4.2 ES上建好索引其实即使不提前建立索引，ES也是可以将数据保存进去的。

尚硅谷大数据课程大纲
尚硅谷的大数据课程大纲主要包括以下几个部分：
1. Java SE：学习Java的基础语法和面向对象编程思想，以及常用的Java
开发工具。

2. MySQL：学习关系型数据库管理系统MySQL，包括数据库设计、SQL
语言、存储过程、触发器等。

3. Linux：学习Linux操作系统的基本命令和常用工具，以及在Linux环境
下部署和配置应用程序。

4. Maven：学习使用Maven进行项目管理和构建，了解Maven的依赖管理、项目构建、项目报告等功能。

5. Shell：学习Shell脚本编程，以及在Linux环境下使用Shell脚本来自动化管理任务。

6. Hadoop：学习使用Hadoop进行大数据处理和分析，了解Hadoop的分布式文件系统、MapReduce编程模型、Hive、HBase等组件。

7. Zookeeper：学习使用Zookeeper进行分布式系统的一致性协调和管理。

8. Hive：学习使用Hive进行数据仓库建设和管理，以及Hive的数据建模
和SQL查询。

9. Flume：学习使用Flume进行大数据采集和传输，了解Flume的数据采集、数据清洗、数据传输等功能。

10. 大数据实践项目：通过实际的大数据项目，综合运用所学知识解决实际问题，提升大数据处理和分析能力。

以上是尚硅谷大数据课程大纲的简要介绍，具体内容可能会根据不同的版本和讲师有所差异。

第5章DML数据操作5.1 数据导入5.1.1 向表中装载数据（Load）1）语法hive> load data [local] inpath '/opt/module/datas/student.txt' [overwrite] into table student [partition (partcol1=val1,…)];（1）load data:表示加载数据（2）local:表示从本地加载数据到hive表；否则从HDFS加载数据到hive表（3）inpath:表示加载数据的路径（4）overwrite:表示覆盖表中已有数据，否则表示追加（5）into table:表示加载到哪张表（6）student:表示具体的表（7）partition:表示上传到指定分区2）实操案例（0）创建一张表hive (default)> create table student(id string, name string) row format delimited fields terminated by '\t';（1）加载本地文件到hivehive (default)> load data local inpath '/opt/module/datas/student.txt' into table default.student;（2）加载HDFS文件到hive中上传文件到HDFShive (default)> dfs -put /opt/module/datas/student.txt /user/atguigu/hive;加载HDFS上数据hive (default)> load data inpath '/user/atguigu/hive/student.txt' into table default.student;（3）加载数据覆盖表中已有的数据上传文件到HDFShive (default)> dfs -put /opt/module/datas/student.txt /user/atguigu/hive;加载数据覆盖表中已有的数据hive (default)> load data inpath '/user/atguigu/hive/student.txt' overwrite into table default.student;5.1.2 通过查询语句向表中插入数据（Insert）1）创建一张分区表hive (default)> create table student(id int, name string) partitioned by (month string) row format delimited fields terminated by '\t';2）基本插入数据3）基本模式插入（根据单张表查询结果）hive (default)> insert overwrite table student partition(month='201708')select id, name from student where month='201709';4）多插入模式（根据多张表查询结果）hive (default)> from studentinsert overwrite table student partition(month='201707')select id, name where month='201709'insert overwrite table student partition(month='201706')select id, name where month='201709';5.1.3 查询语句中创建表并加载数据（As Select）详见4.5.1章创建表。

硅谷大数据技术之Flink-CDC()版本：V2.0第1章CDC简介1.1什么是CDCCDC是Change Data Capture(变更数据获取)的简称。

核心思想是，监测并捕获数据库的变动（包括数据或数据表的插入、更新以及删除等），将这些变更按发生的顺序完整记录下来，写入到消息中间件中以供其他服务进行订阅及消费。

1.2CDC的种类CDC主要分为基于查询和基于Binlog两种方式，我们主要了解一下这两种之间的区别：1.3Flink-CDCFlink社区开发了flink-cdc-connectors组件，这是一个可以直接从MySQL、PostgreSQL 等数据库直接读取全量数据和增量变更数据的source组件。

目前也已开源，开源地址：https:///ververica/flink-cdc-connectors第2章FlinkCDC案例实操2.1DataStream方式的应用2.1.1导入依赖<dependencies><dependency><groupId>org.apache.flink</groupId><artifactId>flink-java</artifactId><version>1.12.0</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-streaming-java_2.12</artifactId><version>1.12.0</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-clients_2.12</artifactId><version>1.12.0</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-client</artifactId><version>3.1.3</version></dependency><dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId><version>5.1.49</version></dependency><dependency><groupId>com.alibaba.ververica</groupId><artifactId>flink-connector-mysql-cdc</artifactId><version>1.2.0</version></dependency><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>1.2.75</version></dependency></dependencies><build><plugins><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-assembly-plugin</artifactId><version>3.0.0</version><configuration><descriptorRefs><descriptorRef>jar-with-dependencies</descriptorRef></descriptorRefs></configuration><executions><execution><id>make-assembly</id><phase>package</phase><goals><goal>single</goal></goals></execution></executions></plugin></plugins></build>2.1.2编写代码import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;import com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction;import com.alibaba.ververica.cdc.debezium.StringDebeziumDeserializationSchema; import mon.restartstrategy.RestartStrategies;import org.apache.flink.runtime.state.filesystem.FsStateBackend;import org.apache.flink.streaming.api.CheckpointingMode;import org.apache.flink.streaming.api.datastream.DataStreamSource;import org.apache.flink.streaming.api.environment.CheckpointConfig;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import java.util.Properties;public class FlinkCDC{public static void main(String[]args)throws Exception{//1.创建执行环境StreamExecutionEnvironment env= StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);//2.Flink-CDC将读取binlog的位置信息以状态的方式保存在CK,如果想要做到断点续传,需要从Checkpoint或者Savepoint启动程序//2.1开启Checkpoint,每隔5秒钟做一次CKenv.enableCheckpointing(5000L);//2.2指定CK的一致性语义env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);//2.3设置任务关闭的时候保留最后一次CK数据env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckp ointCleanup.RETAIN_ON_CANCELLATION);//2.4指定从CK自动重启策略env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,2000L));//2.5设置状态后端env.setStateBackend(new FsStateBackend("hdfs://hadoop102:8020/flinkCDC"));//2.6设置访问HDFS的用户名System.setProperty("HADOOP_USER_NAME","atguigu");//3.创建Flink-MySQL-CDC的Source//initial(default):Performs an initial snapshot on the monitored database tables upon first startup,and continue to read the latest binlog.//latest-offset:Never to perform snapshot on the monitored database tables upon first startup,just read from the end of the binlog which means only have the changes since the connector was started.//timestamp:Never to perform snapshot on the monitored database tables upon first startup,and directly read binlog from the specified timestamp.The consumer will traverse the binlog from the beginning and ignore change events whose timestamp is smaller than the specified timestamp.//specific-offset:Never to perform snapshot on the monitored database tables upon first startup,and directly read binlog from the specified offset.DebeziumSourceFunction<String>mysqlSource=MySQLSource.<String>builder().hostname("hadoop102").port(3306).username("root").password("000000").databaseList("gmall-flink").tableList("gmall-flink.z_user_info")//可选配置项,如果不指定该参数,则会读取上一个配置下的所有表的数据，注意：指定的时候需要使用"db.table"的方式.startupOptions(StartupOptions.initial()).deserializer(new StringDebeziumDeserializationSchema()).build();//4.使用CDC Source从MySQL读取数据DataStreamSource<String>mysqlDS=env.addSource(mysqlSource);//5.打印数据mysqlDS.print();//6.执行任务env.execute();}}2.1.3案例测试1）打包并上传至Linux2）开启MySQL Binlog并重启MySQL3）启动Flink集群[atguigu@hadoop102flink-standalone]$bin/start-cluster.sh4）启动HDFS集群[atguigu@hadoop102flink-standalone]$start-dfs.sh5）启动程序[atguigu@hadoop102flink-standalone]$bin/flink run-c com.atguigu.FlinkCDC flink-1.0-SNAPSHOT-jar-with-dependencies.jar6）在MySQL的gmall-flink.z_user_info表中添加、修改或者删除数据7）给当前的Flink程序创建Savepoint[atguigu@hadoop102flink-standalone]$bin/flink savepoint JobIdhdfs://hadoop102:8020/flink/save8）关闭程序以后从Savepoint重启程序[atguigu@hadoop102flink-standalone]$bin/flink run-s hdfs://hadoop102:8020/flink/save/...-c com.atguigu.FlinkCDC flink-1.0-SNAPSHOT-jar-with-dependencies.jar2.2FlinkSQL方式的应用2.2.1添加依赖<dependency><groupId>org.apache.flink</groupId><artifactId>flink-table-planner-blink_2.12</artifactId><version>1.12.0</version></dependency>2.2.2代码实现import mon.restartstrategy.RestartStrategies;import org.apache.flink.runtime.state.filesystem.FsStateBackend;import org.apache.flink.streaming.api.CheckpointingMode;import org.apache.flink.streaming.api.environment.CheckpointConfig;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;public class FlinkSQL_CDC{public static void main(String[]args)throws Exception{//1.创建执行环境StreamExecutionEnvironment env= StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);StreamTableEnvironment tableEnv=StreamTableEnvironment.create(env);//2.创建Flink-MySQL-CDC的SourcetableEnv.executeSql("CREATE TABLE user_info("+"id INT,"+"name STRING,"+"phone_num STRING"+")WITH("+"'connector'='mysql-cdc',"+"'hostname'='hadoop102',"+"'port'='3306',"+"'username'='root',"+"'password'='000000',"+"'database-name'='gmall-flink',"+"'table-name'='z_user_info'"+")");tableEnv.executeSql("select*from user_info").print();env.execute();}}2.3自定义反序列化器2.3.1代码实现import com.alibaba.fastjson.JSONObject;import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;import com.alibaba.ververica.cdc.debezium.DebeziumDeserializationSchema; import com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction;import io.debezium.data.Envelope;import mon.restartstrategy.RestartStrategies;import mon.typeinfo.TypeInformation;import org.apache.flink.runtime.state.filesystem.FsStateBackend;import org.apache.flink.streaming.api.CheckpointingMode;import org.apache.flink.streaming.api.datastream.DataStreamSource;import org.apache.flink.streaming.api.environment.CheckpointConfig;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.util.Collector;import org.apache.kafka.connect.data.Field;import org.apache.kafka.connect.data.Struct;import org.apache.kafka.connect.source.SourceRecord;import java.util.Properties;public class Flink_CDCWithCustomerSchema{public static void main(String[]args)throws Exception{//1.创建执行环境StreamExecutionEnvironment env= StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);//2.创建Flink-MySQL-CDC的SourceProperties properties=new Properties();//initial(default):Performs an initial snapshot on the monitored database tables upon first startup,and continue to read the latest binlog.//latest-offset:Never to perform snapshot on the monitored database tables upon first startup,just read from the end of the binlog which means only have the changes since the connector was started.//timestamp:Never to perform snapshot on the monitored database tables upon first startup,and directly read binlog from the specified timestamp.The consumer will traverse the binlog from the beginning and ignore change events whose timestamp is smaller than the specified timestamp.//specific-offset:Never to perform snapshot on the monitored database tables upon first startup,and directly read binlog from the specified offset.DebeziumSourceFunction<String>mysqlSource=MySQLSource.<String>builder().hostname("hadoop102").port(3306).username("root").password("000000").databaseList("gmall-flink").tableList("gmall-flink.z_user_info")//可选配置项,如果不指定该参数,则会读取上一个配置下的所有表的数据,注意：指定的时候需要使用"db.table"的方式.startupOptions(StartupOptions.initial()).deserializer(new DebeziumDeserializationSchema<String>(){//自定义数据解析器@Overridepublic void deserialize(SourceRecord sourceRecord,Collector<String> collector)throws Exception{//获取主题信息,包含着数据库和表名mysql_binlog_source.gmall-flink.z_user_infoString topic=sourceRecord.topic();String[]arr=topic.split("\\.");String db=arr[1];String tableName=arr[2];//获取操作类型READ DELETE UPDATE CREATEEnvelope.Operation operation=Envelope.operationFor(sourceRecord);//获取值信息并转换为Struct类型Struct value=(Struct)sourceRecord.value();//获取变化后的数据Struct after=value.getStruct("after");//创建JSON对象用于存储数据信息JSONObject data=new JSONObject();for(Field field:after.schema().fields()){Object o=after.get(field);data.put((),o);}//创建JSON对象用于封装最终返回值数据信息JSONObject result=new JSONObject();result.put("operation",operation.toString().toLowerCase());result.put("data",data);result.put("database",db);result.put("table",tableName);//发送数据至下游collector.collect(result.toJSONString());}@Overridepublic TypeInformation<String>getProducedType(){return TypeInformation.of(String.class);}}).build();//3.使用CDC Source从MySQL读取数据DataStreamSource<String>mysqlDS=env.addSource(mysqlSource);//4.打印数据mysqlDS.print();//5.执行任务env.execute();}}。

尚硅谷大数据项目之电商数仓（系统业务数据仓库）(作者：尚硅谷大数据研发部)版本：V6.1.2第1章电商业务简介1.1 电商业务流程1.2 电商常识（SKU、SPU）SKU=Stock Keeping Unit（库存量基本单位）。

现在已经被引申为产品统一编号的简称，每种产品均对应有唯一的SKU号。

SPU（Standard Product Unit）：是商品信息聚合的最小单位，是一组可复用、易检索的标准化信息集合。

例如：iPhoneX手机就是SPU。

一台银色、128G内存的、支持联通网络的iPhoneX，就是SKU。

SPU 表示一类商品。

好处就是：可以共用商品图片，海报、销售属性等。

1.3 电商业务表结构电商表结构订单表order_info商品评论表comment_info 加购表cart_info 商品收藏表favor_info用户表user_info 支付流水表payment_info优惠卷领用表coupon_use优惠卷表coupon_info 活动表Activity_info 活动订单表Activity_order优惠规则表Activity_rule参与活动商品表Activity_sku省份表base_province地区表base_region订单详情表order_detailSKU 商品表sku_info商品三级分类表base_category3商品二级分类表base_category2商品一级分类表base_category1SPU 商品表spu_info品牌表base_trademark退单表order_redund_info订单状态表order_status_log编码字典表base_dic1.3.1 订单表（order_info ）标签 含义 id 订单编号 consignee 收货人 consignee_tel 收件人电话 final_total_amount 总金额 order_status 订单状态 user_id用户id delivery_address 送货地址 order_comment 订单备注out_trade_no 订单交易编号（第三方支付用)trade_body 订单描述(第三方支付用) create_time 创建时间operate_time 操作时间expire_time 失效时间tracking_no 物流单编号parent_order_id 父订单编号img_url 图片路径province_id 地区benefit_reduce_amount 优惠金额original_total_amount 原价金额feight_fee 运费1.3.2 订单详情表（order_detail）标签含义id 订单编号order_id 订单号sku_id 商品idsku_name sku名称（冗余)img_url 图片名称（冗余)order_price 商品价格(下单时sku价格）sku_num 商品数量create_time 创建时间1.3.3 SKU商品表（sku_info）标签含义id skuIdspu_id spuidprice 价格sku_name 商品名称sku_desc 商品描述weight 重量tm_id 品牌idcategory3_id 品类idsku_default_img 默认显示图片(冗余) create_time 创建时间1.3.4 用户表（user_info）标签含义id 用户idlogin_name 用户名称nick_name 用户昵称passwd 用户密码name 姓名phone_num 手机号email 邮箱head_img 头像user_level 用户级别birthday 生日gender 性别M男,F女create_time 创建时间operate_time操作时间1.3.5 商品一级分类表（base_category1）标签含义id idname 名称1.3.6 商品二级分类表（base_category2）标签含义id idname 名称category1_id 一级品类id1.3.7 商品三级分类表（base_category3）标签含义id idname 名称Category2_id 二级品类id1.3.8 支付流水表（payment_info）标签含义id 编号out_trade_no 对外业务编号order_id 订单编号user_id 用户编号alipay_trade_no 支付宝交易流水编号total_amount 支付金额subject 交易内容payment_type 支付类型payment_time 支付时间1.3.9 省份表（base_province）标签含义id idname 省份名称region_id 地区IDarea_code 地区编码iso_code 国际编码1.3.10 地区表（base_region）标签含义id 大区idregion_name 大区名称1.3.11 品牌表（base_trademark）标签含义tm_id 品牌idtm_name 品牌名称1.3.12 订单状态表（order_status_log）标签含义id 编号order_id 订单编号order_status 订单状态operate_time 操作时间1.3.13 SPU商品表（spu_info）标签含义id 商品idspu_name spu商品名称description 商品描述(后台简述）category3_id 三级分类idtm_id 品牌id1.3.14 商品评论表（comment_info）标签含义id 编号user_id 用户idsku_id 商品idspu_id spu_idorder_id 订单编号appraise 评价 1 好评 2 中评 3 差评comment_txt 评价内容create_time 创建时间1.3.15 退单表（order_refund_info）标签含义id 编号order_id 订单编号sku_id sku_idrefund_type 退款类型refund_amount 退款金额refund_reason_type 原因类型refund_reason_txt 原因内容create_time 创建时间1.3.16 加购表（cart_info）标签含义id 编号user_id 用户idsku_id SKU商品cart_price 放入购物车时价格sku_num 数量img_url 图片文件sku_name sku名称 (冗余)create_time 创建时间operate_time 修改时间is_ordered 是否已经下单order_time 下单时间1.3.17 商品收藏表（favor_info）标签含义id 编号user_id 用户名称sku_id 商品idspu_id spu_idis_cancel 是否已取消 0 正常 1 已取消create_time 创建时间cancel_time 修改时间1.3.18 优惠券领用表（coupon_use）标签含义id 编号coupon_id 购物券IDuser_id 用户IDorder_id 订单IDcoupon_status 购物券状态get_time 领券时间using_time 使用时间used_time 支付时间expire_time 过期时间1.3.19 优惠券表（coupon_info）标签含义id 购物券编号coupon_name 购物券名称coupon_type 购物券类型 1 现金券 2 折扣券 3 满减券 4 满件打折券condition_amount 满额数condition_num 满件数activity_id 活动编号benefit_amount 减金额benefit_discount 折扣create_time 创建时间range_type 范围类型 1、商品 2、品类 3、品牌spu_id 商品idtm_id 品牌idcategory3_id 品类idlimit_num 最多领用次数operate_time 修改时间expire_time 过期时间1.3.20 活动表（activity_info）标签含义id 活动idactivity_name 活动名称activity_type 活动类型activity_desc 活动描述start_time 开始时间end_time 结束时间create_time 创建时间1.3.21 活动订单关联表（activity_order）标签含义id 编号activity_id 活动idorder_id 订单编号create_time 发生日期1.3.22 优惠规则表（activity_rule）标签含义id 编号activity_id 活动idcondition_amount 满减金额condition_num 满减件数benefit_amount 优惠金额benefit_discount 优惠折扣benefit_level 优惠级别1.3.23 编码字典表（base_dic）标签含义dic_code 编号dic_name 编码名称parent_code 父编号create_time 创建日期operate_time 修改日期1.3.24 活动参与商品表（activity_sku）（暂不导入）标签含义id 编号activity_id 活动idsku_id sku_idcreate_time 创建时间1.4 时间相关表1.4.1 时间表（date_info）标签含义date_id 日期idweek_id 周idweek_day 周day 日month 月quarter 季度year 年is_workday 是否是周末holiday_id 假期id1.4.2 假期表（holiday_info）标签含义holiday_id 假期idholiday_name 假期名称1.4.3 假期年表（holiday_year）标签含义holiday_id 假期idholiday_name 假期名称start_date_id 假期开始时间end_date_id 假期结束时间第2章 业务数据采集模块Flume 消费MySQL 业务数据Nginx业务日志数据（后端埋点数据）Web/App 前端埋点日志服务器(Springboot)Nginx业务服务器(Springboot)日志服务器(Springboot)logFilelogFile业务服务器(Springboot)生产Flume生产FlumeKafkaKafka KafkaSparkSt reamingHbaseHDFSHive (ods dwd dws ads)消费FlumeMysql数据可视化SqoopWeb/App 业务交互PrestoKylin实时指标分析业务交互数据：业务流程中产生的登录、订单、用户、商品、支付等相关的数据，通常存储在DB 中，包括Mysql 、Oracle 埋点用户行为数据：用户在使用产品过程中，与客户端产品交互过程中产生的数据，比如页面浏览、点击、停留、评论、点赞、收藏等2.1 MySQL 安装 2.1.1 安装包准备1）卸载自带的Mysql-libs （如果之前安装过mysql ，要全都卸载掉）[atguigu@hadoop102 software]$ rpm -qa | grep -i-Emysql\|mariadb | xargs -n1 sudo rpm -e --nodeps 2）将安装包和JDBC 驱动上传到/opt/software ，共计6个01_mysql-community-common-5.7.29-1.el7.x86_64.rpm 02_mysql-community-libs-5.7.29-1.el7.x86_64.rpm03_mysql-community-libs-compat-5.7.29-1.el7.x86_64.rpm 04_mysql-community-client-5.7.29-1.el7.x86_64.rpm 05_mysql-community-server-5.7.29-1.el7.x86_64.rpm mysql-connector-java-5.1.48.jar2.1.2 安装MySQL1）安装mysql 依赖[atguigu@hadoop102 software]$ sudo rpm -ivh 01_mysql-community-common-5.7.29-1.el7.x86_64.rpm[atguigu@hadoop102 software]$ sudo rpm -ivh 02_mysql-community-libs-5.7.29-1.el7.x86_64.rpm[atguigu@hadoop102 software]$ sudo rpm -ivh 03_mysql-community-libs-compat-5.7.29-1.el7.x86_64.rpm 2）安装mysql-client[atguigu@hadoop102 software]$ sudo rpm -ivh 04_mysql-community-client-5.7.29-1.el7.x86_64.rpm 3）安装mysql-server[atguigu@hadoop102 software]$ sudo rpm -ivh 05_mysql-community-server-5.7.29-1.el7.x86_64.rpm4）启动mysql[atguigu@hadoop102 software]$ sudo systemctl start mysqld5）查看mysql密码[atguigu@hadoop102 software]$ sudo cat /var/log/mysqld.log | grep password2.1.3 配置MySQL配置只要是root用户+密码，在任何主机上都能登录MySQL数据库。

大数据项目开发案例一、项目背景。

某电商企业作为国内领先的电子商务企业，拥有庞大的用户群体和海量的交易数据。

随着业务的不断扩张，企业内部数据规模不断增长，传统的数据处理方式已经无法满足业务发展的需求。

因此，企业决定启动大数据项目，利用先进的大数据技术和工具来进行数据处理和分析，以实现更精准的营销、更高效的运营和更优质的用户体验。

二、需求分析。

在项目启动初期，企业进行了深入的需求调研和分析，主要包括以下几个方面的需求：1. 数据存储和管理，需要建立一套稳定、高效的数据存储和管理系统，能够支撑海量数据的存储和快速检索。

2. 数据处理和分析，需要实现对海量数据的实时处理和分析，能够为营销、运营和用户体验提供有效的数据支持。

3. 数据可视化，需要将处理和分析后的数据通过直观的可视化方式展现，帮助决策者更直观地了解业务状况和趋势变化。

三、技术架构。

基于以上需求，企业选择了Hadoop作为大数据存储和处理的基础框架，同时引入了Spark作为数据处理和分析的引擎。

此外，为了支持数据可视化，企业还引入了Elasticsearch和Kibana技术，实现了数据的实时监控和可视化展示。

整个技术架构如下图所示：（此处应有技术架构图）。

四、实施过程。

在项目实施过程中，企业充分发挥了各团队的专业能力，按照项目计划和里程碑，分阶段、有序地推进项目实施。

主要包括以下几个阶段：1. 数据采集和清洗，建立数据采集和清洗的流程，确保数据的完整性和准确性。

2. 数据存储和管理，搭建Hadoop集群，实现数据的分布式存储和管理。

3. 数据处理和分析，引入Spark技术，实现对海量数据的实时处理和分析。

4. 数据可视化，利用Elasticsearch和Kibana技术，实现数据的可视化展示和监控。

五、效果评估。

经过大数据项目的实施，企业取得了显著的效果，主要包括以下几个方面：1. 营销效果提升，通过对用户行为数据的分析，企业实现了更精准的营销，提高了营销效果和转化率。

尚硅谷大数据项目之实时分析系统--灵活分析(作者：尚硅谷大数据研发部)版本：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 对接可视化模块。

尚硅谷大数据技术之电商推荐系统第 1 章项目体系架构设计1.1 项目系统架构项目以推荐系统建设领域知名的经过修改过的中文亚马逊电商数据集作为依托，以某电商网站真实业务数据架构为基础，构建了符合教学体系的一体化的电商推荐系统，包含了离线推荐与实时推荐体系，综合利用了协同过滤算法以及基于内容的推荐方法来提供混合推荐。

提供了从前端应用、后台服务、算法设计实现、平台部署等多方位的闭环的业务实现。

用户可视化：主要负责实现和用户的交互以及业务数据的展示，主体采用AngularJS2 进行实现，部署在Apache 服务上。

综合业务服务：主要实现JavaEE 层面整体的业务逻辑，通过Spring 进行构建，对接业务需求。

部署在Tomcat 上。

数据存储部分】 责平台业务逻辑数据的存储。

缓存数据库： 项目采用 Redis 作为缓存数据库，主要用来支撑实时推荐系统部 分对于数据的高速获取需求。

【离线推荐部分】离线统计服务： 批处理统计性业务采用 Spark Core + Spark SQL 进行实现，实 现对指标类数据的统计任务。

离线推荐服务： 离线推荐业务采用 Spark Core + Spark MLlib 进行实现，采用 ALS 算法进行实现。

【实时推荐部分】日志采集服务： 通过利用 Flume-ng 对业务平台中用户对于商品的一次评分行为 进行采集，实时发送到 Kafka 集群。

消息缓冲服务： 项目采用 Kafka 作为流式数据的缓存组件，接受来自 Flume 的 数据采集请求。

并将数据推送到项目的实时推荐系统部分。

实时推荐服务： 项目采用 Spark Streaming 作为实时推荐系统，通过接收 Kafka 中缓存的数据，通过设计的推荐算法实现对实时推荐的数据处理，并将结构合并更 新到 MongoDB 数据库。

1.2 项目数据流程【系统初始化部分】0.通过Spark SQL 将系统初始化数据加载到MongoDB 中。