想学习架构师构建流程请跳转：Java架构师系统架构设计
什么是智能推荐？ 定义： 根据用户行为习惯所提供的数据， 系统提供策略模型，自动推荐符合用户行为的信息。 例举： 比如根据用户对商品的点击数据（时间周期，点击频次）， 推荐类似的商品； 根据用户的评价与满意度， 推荐合适的品牌； 根据用户的使用习惯与点击行为，推荐类似的资讯。 应用案例：
什么是实时数仓 数据仓库（Data Warehouse）,可简写为DW或DWH，是一个庞大的数据存储集合，通过对各种业务数 据进行筛选与整合，生成企业的分析性报告和各类报表，为企业的决策提供支持。实时仓库是基于 Storm/Spark（Streaming）/Flink等实时处理框架，构建的具备实时性特征的数据仓库。
应用案例 分析物流数据， 提升物流处理效率。 
阿里巴巴菜鸟网络实时数仓设计： 
数仓分层处理架构（流式ETL）： ODS -> DWD -> DWS -> ADS ODS（Operation Data Store）：操作数据层， 一般为原始采集数据。 DWD(Data Warehouse Detail) ：明细数据层， 对数据经过清洗，也称为DWI。 DWS(Data Warehouse Service)：汇总数据层，基于DWD层数据， 整合汇总成分析某一个主题域的服 务数据，一般是宽表， 由多个属性关联在一起的表， 比如用户行为日志信息：点赞、评论、收藏等。 ADS(Application Data Store): 应用数据层， 将结果同步至RDS数据库中， 一般做报表呈现使用。 
- IoT数据分析
- 什么是IoT 物联网是新一代信息技术，也是未来发展的趋势，英文全称为： Internet of things（IOT），顾名 思义， 物联网就是万物相联。物联网通过智能感知、识别技术与普适计算等通信感知技术，广泛 应用于网络的融合中，也因此被称为继计算机、互联网之后世界信息产业发展的第三次浪潮。
- 应用案例 物联网设备运营分析：
华为Iot数据分析平台架构： 
- 智慧城市 城市中汽车越来越多， 川流不息，高德地图等APP通过技术手段采集了越来越多的摄像头、车流 的数据。 但道路却越来越拥堵，越来越多的城市开始通过大数据技术， 对城市实行智能化管理。 2018年， 杭州采用AI智慧城市，平均通行速度提高15%，监控摄像头日报警次数高达500次，识 别准确率超过92%，AI智慧城市通报占全体95%以上，在中国城市交通堵塞排行榜， 杭州从中国 第5名降至57名。
- 金融风控 风险是金融机构业务固有特性，与金融机构相伴而生。金融机构盈利的来源就是承担风险的风险溢 价。 金融机构中常见的六种风险：市场风险、信用风险、流动性风险、操作风险、声誉风险及法律风 险。其中最主要的是市场风险和信用风险。 线上信贷流程，通过后台大数据系统进行反欺诈和信用评估：
- 电商行业 用户在电商的购物网站数据通过实时大数据分析之后， 通过大屏汇总展示， 比如天猫的双11购物 活动，通过大屏， 将全国上亿买家的订单数据可视化，实时性的动态展示，包含总览数据，流式 TopN数据，多维区域统计数据等，极大的增强了对海量数据的可读性。
TopN排行： 
大数据Flink概述 大数据Flink入门案例
Flink 连接器包含数据源输入与汇聚输出两部分。Flink自身内置了一些基础的连接器，数据源输入包含文件、目录、Socket以及 支持从collections 和 iterators 中读取数据；汇聚输出支持把数据写入文件、标准输出（stdout）、标准错误输出（stderr）和 socket。 官方地址 Flink还可以支持扩展的连接器，能够与第三方系统进行交互。目前支持以下系统:
Flink还可以支持扩展的连接器，能够与第三方系统进行交互。目前支持以下系统:
- Apache Kafka (source/sink)
- Apache Cassandra (sink)
- Amazon Kinesis Streams (source/sink)
- Elasticsearch (sink)
- Hadoop FileSystem (sink)
- RabbitMQ (source/sink)
- Apache NiFi (source/sink)
- Twitter Streaming API (source)
- Google PubSub (source/sink)
- JDBC (sink)
常用的是Kafka、ES、HDFS以及JDBC。
Flink Connectors JDBC 如何使用？ 功能： 将集合数据写入数据库中
代码：
数据表：
自定义写入数据源 功能：读取Socket数据， 采用流方式写入数据库中。 代码：
自定义数据源
AccessLog:
测试数据：注意
自定义读取数据源 功能： 读取数据库中的数据， 并将结果打印出来。 代码：
通过Sink写入HDFS数据 功能： 将Socket接收到的数据， 写入至HDFS文件中。
依赖
代码：
数据源模拟实现：
代码:
读取HDFS文件数据
Hadoop环境安装
- 配置免密码登录 生成秘钥：
将秘钥写入认证文件：
修改认证文件权限：
- 配置环境变量 将Hadoop安装包解压， 将Hadoop加入环境变量/etc/profile：
执行生效：
- 修改Hadoop配置文件 1） 修改hadoop-env.sh文件
修改JAVA_HOME：
2）修改core-site.xml文件
这里的主机名称是flink。 3）修改hdfs-site.xml文件
4）修改mapred-site.xml文件
5）修改slaves文件
这里配置的是单节点， 指向本机主机名称。 6）修改yarn-site.xml
- 启动Hadoop服务
上传一个文件， 用于测试：
如果上传失败 1）可能是namenode没有启动，则执行如下命令
2）检查/etc/hosts文件配置
- 访问验证
ES服务安装
- 到官网下载地址下载6.8.1版本的gz压缩包， 不要下载最新版本， Spring Boot等项目可能未及时更新支持。
- 解压安装包
- ElasticSearch不能以Root身份运行， 需要单独创建一个用户
执行以上命令，创建一个名为elsearch用户， 并赋予目录权限。 4. 修改配置文件 vi config/elasticsearch.yml, 只需修改以下设置：
- 指定JDK版本 最新版的ElasticSearch需要JDK11版本， 下载JDK11压缩包， 并进行解压。
修改环境配置文件 vi bin/elasticsearch-env 参照以下位置， 追加一行， 设置JAVA_HOME， 指定JDK11路径。
- 关闭ConcMarkSweepGC
JDK9版本以后不建议使用ConcMarkSweepGC， 如果不想出现提示， 可以将其关闭 vi config/jvm.options 将UseConcMarkSweepGC注释：
- 启动ElasticSearch
- 切换用户
su elsearch
- 以后台常驻方式启动
bin/elasticsearch -d 7. 问题处理 出现max virtual memory areas vm.max_map_count [65530] is too low, increase to at least 错误信息 修改系统配置：
添加
执行生效
在文件末尾添加
重新切换用户即可：
Flink ES写入功能实现 功能： 将Socket流数据， 写入至ES服务。 依赖
代码：
查看index信息： http://192.168.116.141:9200/_cat/indices?v 查看具体数据： http://192.168.116.141:9200/flink-es/_search
Kafka安装
- 下载Kafka_2.12-1.1.1安装包
- 将安装包解压
- 修改kafka配置 只修改绑定IP， 因为是单节点， 其他按默认配置来。
如有多个IP地址， 绑定为对外访问的IP。 4. 启动zookeeper服务 kafka安装包内置了zookeeper，可以直接启动。
- 启动kafka服务
Flink Kafka 读取功能 功能： 通过flink读取kafka消息队列数据， 并打印显示。 依赖
代码：
通过kafka生产者命令测试验证
扩展点：kafka消息的消费处理策略：
Flink Kafka 写入功能 功能： 将Socket的流数据，通过flink 写入kafka 消息队列。 代码：
通过kafka消费者命令测试验证：
控制消息的发送处理模式:
提供了三种消息处理模式：
- Semantic.NONE ：Flink 不会有任何语义的保证，产生的记录可能会丢失或重复。
- Semantic.AT_LEAST_onCE （默认设置）：类似 FlinkKafkaProducer010 版本中的setFlushonCheckpoint(true) ，这可以保证不会丢失任何记录（虽然记录可能会重复）。
- Semantic.EXACTLY_onCE ：使用 Kafka 事务提供精准一次的语义。无论何时，在使用事务写入 Kafka时，都要记得为所有消费 Kafka 消息的应用程序设置所需的 isolation.level （ read_committed 或 read_uncommitted - 后者是默认值）。
- Kafka 的消息可以携带时间戳，指示事件发生的时间或消息写入 Kafka broker 的时间。
在实际应用场景中， 会存在各种复杂传输对象，同时要求较高的传输处理性能， 这就需要采用自定义的序列化方式做相应实现， 这里以Protobuf为例做讲解。 功能： kafka对同一Topic的生产与消费，采用Protobuf做序列化与反序列化传输， 验证能否正常解析数据。
- 通过protobuf脚本生成JAVA文件
通过批处理脚本，生成JAVA文件：
注意， 路径要配置正确。 2. 自定义序列化实现 添加POM依赖：
AccessLog对象：
序列话好之后会根据AccessLog对象得到一个序列号的文件
CustomSerialSchema:
3. 通过flink对kafka消息生产者的实现
开启Kafka消费者命令行终端，验证生产者的可用性：
- 通过flink对kafka消息订阅者的实现
通过flink的kafka生产者消息的发送， 对消费者的功能做测试验证。
- 总览数据
总销售量/总销售金额 TopN: 热销商品/商品类目/商品PV/商品UV
- 区域/分类数据
不同区域销售排名 不同分类销售排名
- 下载安装包 安装包 后台管理包
- 解压 解压安装包：
解压管理包：
- 初始化管理数据库 导入初始化数据脚本：
- 修改MySQL服务同步配置 编辑配置文件：
增加同步配置：
重启服务：
检查同步功能是否开启
创建同步用户：
赋予同步所需权限：
- 修改后台管理配置文件
配置内容：
先启动后台管理服务， 再启动Canal服务， 后台管理服务启动命令：
访问：http://192.168.116.141:8089/ 登录： admin/123456 6. Canal服务配置
配置内容：
启动Canal服务:
- 后台管理配置 修改Server管理配置：
修改Instance配置（如果没有， 则新建，载入模板即可）：
regex同步配置规则： 常见例子：
- 所有表：.* or …
- canal schema下所有表： canal…*
- canal下的以canal打头的表：canal.canal.*
- canal schema下的一张表：canal.test1
- 多个规则组合使用：canal…*,mysql.test1,mysql.test2 (逗号分隔)
功能实现流程：
- 订单数据源的实现
- flink代码功能实现
- Flink 与 Spring Boot的集成
- 测试验证，比对SQL:
- 数据呈现
kibana服务安装 Kibana是一个针对Elasticsearch的开源分析及可视化平台，用来搜索、查看交互存储在Elasticsearch索 引中的数据。 6. 到官网下载， Kibana安装包, 与之对应6.8.1版本， 选择Linux 64位版本下载，并进行解压。 7. Kibana启动不能使用root用户， 使用上面创建的elsearch用户， 进行赋权
- 修改配置文件 vi config/kibana.yml , 修改以下配置：
- 启动kibana
看到以下日志， 代表启动正常
如果出现启动失败的情况， 要检查集群各节点的日志， 确保服务正常运行状态
- 增加订单地址信息数据源
- 创建对应的表与实体 实体： OrderAddress BO： JoinOrderAddress（订单数据与地址数据的合并对象） BO： HotDimensionOrder（ES存储的映射对象）， 注意这里的ID唯一性， 如果是按省份统计， ID存储省份信息，如果是按地级市统计， ID则存储为市区信息。
- 改造订单数据源， 增加缓存写入， 地址信息数据源增加缓存的读取。
- 修改Canal的后台配置， 增加地址数据源的监听队列。
- 区域双流统计的核心代码实现： 1）增加双流的kafka配置， 每个流监听不同的数据队列。 2）每个流要加上时间水印， 设定时间窗， 设定值比后面聚合的时间窗稍小一些。 3）根据订单ID做join匹配。 4） 根据区域做汇总统计（省份、城市）。 5） 将数据写入至ES。
- 测试验证 验证SQL:
- 增加订单支付流水数据源
- 创建对应的表与实体 实体： OrderPayment BO： JoinOrderAddress
- 修改Canal的后台配置， 增加地址数据源的监听队列。
- 核心代码实现： 1）实现订单支付流水数据源的监听处理。 2）定义CEP处理规则，解析出支付成功的订单。
- 测试验证 检查订单状态是未支付 -》 已支付的数据
检查超时的数据： 初始状态为0， 指定时间之内没有已支付的数据。 6. 拓展实现， 热门商品统计排行，只统计支付成功的数据。
功能： 统计商品在一段时间内的UV（Unique Visitor） 核心代码：
功能： 统计商品在一段时间内的UV（采用布隆过滤器） 核心代码：