• 实时统计前10名流量最高的业务信息

• 可视化图表展示实时统计结果

• 统计一天中用户访问业务最多的时间段 (视频)

• 统计用户访问最多的业务网站

1. 其中01节点和02节点负责产生日志。产生日志的每台服务其上安装Flume进行日志采集，然后把各自采集到的日志数据发送给同一个FLume服务器（03）节点进行日志合并；

2. 将合并后的日志数据以副本的方式分成两路（两路数据一致）：一路进行实时计算，另一路进行离线计算。降雪要实时计算的数据发送到实时消息系统Kafka进行中转，将需要离线计算的数据存储到HBase分布式数据库中；

3. 使用Spark Streaming作为Kafka的消费者，按批次从Kafka中获取数据进行实时计算，并将计算结果存储于MySQL关系数据库中；

4. 使用SparkSQL查询HBase中的数据进行离线数据，将计算结果存储于MySQL关系数据库中；

5. 使用WebSocket实时获取MySQL中的数据，通过可视化组件展示流量最高的前10项业务。
模块架构如下：
数据来源于脚本自动生成的日志文件。
使用FLume+Kafka构建数据传输层，Flume可以将大量日志数据从许多不同源（两个centos服务器节点）进行收集、聚合，最终移动到一个集中的数据中心（centos03）进行存储；Kafka基于Zookeeper，高吞吐量低延迟分布式实时处理消息，在本模块作为消息中间件进行日志的中转。
HDFS、HBase、Hive、MySQL构成数据存储层。Hive是一个基于hadoop的数据仓库架构，使用SQL语句读写管理大型分布式数据集，可以将SQL语句转换为MapReduce任务执行，MySQL用于存储用户行为数据的计算结果。
Spark引擎为数据计算层核心，Spark SQL可以结合HBase进行数据查询和分析，Spark streaming可以进行实时数据流的处理。

1. 启动Hadoop集群
在01节点执行命令： start-all.sh 停止集群： stop-all.sh
HDFS Web: http://192.168.170.133:9870

2. 启动Kafka中转用户行为数据（Kafka依赖于zookeeper）
①启动zookeeper
在每个节点的安装目录 cd /opt/modules/apache-zookeeper-3.6.3-bin下执行：bin/zkServer.sh start
②启动Kafka集群
cd /opt/modules/kafka_2.12-3.1.0
bin/kafka-server-start.sh -daemon config/server.properties
常用命令：
Ⅰ.Kafka中存在的主题查询
bin/kafka-topics.sh --list --bootstrap-server centos01:9092



Ⅱ. 创建生产者
bin/kafka-console-producer.sh \ --broker-list centos01:9092,centos02:9092,centos03:9092 \ --topic topictest
Ⅲ. 创建消费者
bin/kafka-console-consumer.sh \ --bootstrap-server centos01:9092,centos02:9092,centos03:9092 \ --topic topictest

3. 启动Flume多节点数据采集



Centos01节点和centos02节点的Flume负责实时监控日志文件数据，并将监控到的新数据发送到centos03节点中的Flume合并，centos03节点接收数据并以副本的形式分为两路，需要实时计算的数据发送到Kafka进行中转(见视频“日志消息经flume汇总写入Kafka”,从视频可以看出，flume收集来自两个节点的数据，并且同时写入了kafka消费者和存储到hbase中)，需要离线计算的数据存储到HBase分布式数据库中。

4. centos03节点执行命令(flume安装目录下)：cd /opt/modules/apache-flume-1.9.0-bin
bin/flume-ng agent \ --conf conf \ --conf-file /opt/modules/apache-flume-1.9.0-bin/conf/flume-kafka-hbase.properties \ --name a1 \ -Dflume.root.logger=INFO,console

5. Centos01、centos02节点启动flume：
bin/flume-ng agent \ --conf conf \ --conf-file /opt/modules/apache-flume-1.9.0-bin/conf/flume-conf.properties \ --name a1 \ -Dflume.root.logger=INFO,console

6. 启动Hbase集群 cd /opt/modules/hbase-2.4.9
在centos01节点上执行：bin/start-hbase.sh
常用命令：
① 启动Hbase shell: bin/hbase shell

7. 启动Hive
1. 启动mysql服务：sudo systemctl start mysqld
2. hive进入hive命令行模式

8. 启动Spark
在centos01节点上进入spark安装目录，启动Spark集群：
sbin/start-all.sh
Spark WebUI : http://centos01:8080



Spark shell :bin/spark-shell --master spark://centos01:7077 --jars /opt/softwares/mysql-connector-java-8.0.11.jar
val jdbcDF = spark.read.format("jdbc")
.option("url", "jdbc:mysql://192.168.170.133:3306/spark_db")
.option("driver", "com.mysql.cj.jdbc.Driver")
.option("dbtable", "student")
.option("user","root")
.option("password", "123456")
.load()

9. Spark提交程序
bin/spark-submit --conf spark.dynamicAllocation.enabled=false --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.2.1 --master spark://centos01:7077 --class spark.demo.StructuredStreamingKafkaFinal /opt/softwares/spark.demo.jar

10. 产生实时数据流+运行前端
java -cp /opt/softwares/spark.demo.jar spark.demo.ReadWriteDa /opt/modules/data/SogouQ.reduced /opt/modules/data/user_behavior_info
java -cp /opt/softwares/spark.demo.jar \ spark.demo.ReadWriteData \ /opt/modules/data/SogouQ.reduced \ /opt/modules/data/user_behavior_info
前端在IDEA中运行APP.java main方法。
视频见sparkstreaming,提交StructuredStreamingKafkaFinal.scala程序到spark集群，spark将实时监控Kafka中数据，并更新计算结果到mysql。利用centos01、02节点中日志数据文件模拟多数据源-->centos01和centos02同时产生数据，centos03汇总计算并经由spark将计算结果存储至mysql。
视频“sparkstreaming（带可视化）”中补充了基于SpringBoot做的简易可视化，利用WebSocket实现服务端和客户端通信，并实时绘制每天的搜索业务关键词数量。



java -cp /opt/softwares/user_analyse_web.jar

11. Sparkdemo
• 代码结构



• Spark SQL







• spark streaming

Attachments