基于Java实现Spark2x新闻网大数据实时分析可视化系统项目【100012794】

# 基于Spark2.x新闻网大数据实时分析可视化系统项目 ## 一、业务需求分析 捕获用户浏览日志信息 实时分析前20名流量最高的新闻话题 实时统计当前线上已曝光的新闻话题 统计哪个时段用户浏览量最高 ## 二、系统架构图设计  ## 三、系统数据流程设计  ## 四、集群资源规划设计  ## 五、步骤详解 考虑到实际情况，本人集群配置共三个节点（node5、node6、node7）。 ### 5.1Zookeeper分布式集群部署 参考 [博客](https://blog.csdn.net/u011254180/article/details/79480234) ### 5.2. Hadoop2.X HA架构与部署 参考 [博客](https://blog.csdn.net/u011254180/article/details/77922331) ### 5.3. HBase分布式集群部署与设计 参考 [博客](https://blog.csdn.net/u011254180/article/details/80171500) ### 5.4. Kafka分布式集群部署 参考 [博客](https://blog.csdn.net/u011254180/article/details/79481088) ### 5.5.Flume部署及数据采集准备 node6与node7中flume数据采集到node5中，而且node6和node7的flume配置文件大致相同，node7中将a2改为a3，如下 ```js a2.sources = r1 a2.sinks = k1 a2.channels = c1 a2.sources.r1.type = exec a2.sources.r1.command = tail -F /opt/data/weblog-flume.log a2.sources.r1.channels = c1 a2.channels.c1.type = memory a2.channels.c1.capacity = 1000 a2.channels.c1.transactionCapacity = 1000 a2.channels.c1.keep-alive = 5 a2.sinks.k1.type = avro a2.sinks.k1.channel = c1 a2.sinks.k1.hostname = node5 a2.sinks.k1.port = 5555 ``` ## 5.6Flume+HBase+Kafka集成与开发 #### 1.下载Flume源码并导入Idea开发工具 1）将apache-flume-1.7.0-src.tar.gz源码下载到本地解压 2）通过idea导入flume源码 打开idea开发工具，选择File——》Open  然后找到flume源码解压文件，选中flume-ng-hbase-sink，点击ok加载相应模块的源码。\  #### 2.官方flume与hbase集成的参数介绍  #### 3.下载日志数据并分析 到搜狗实验室 下载用户查询日志 1）介绍 搜索引擎查询日志库设计为包括约1个月(2008年6月)Sogou搜索引擎部分网页查询需求及用户点击情况的网页查询日志数据集合。为进行中文搜索引擎用户行为分析的研究者提供基准研究语料。 2）格式说明 数据格式为:访问时间\t用户ID\t[查询词]\t该URL在返回结果中的排名\t用户点击的顺序号\t用户点击的URL 其中，用户ID是根据用户使用浏览器访问搜索引擎时的Cookie信息自动赋值，即同一次使用浏览器输入的不同查询对应同一个用户ID  #### 4.node5聚合节点与HBase和Kafka的集成配置 node5通过flume接收node6与node7中flume传来的数据，并将其分别发送至hbase与kafka中，配置内容如下 ```js a1.sources = r1 a1.channels = kafkaC hbaseC a1.sinks = kafkaSink hbaseSink a1.sources.r1.type = avro a1.sources.r1.channels = hbaseC kafkaC a1.sources.r1.bind = node5 a1.sources.r1.port = 5555 a1.sources.r1.threads = 5 # ****************************flume + hbase****************************** a1.channels.hbaseC.type = memory a1.channels.hbaseC.capacity = 10000 a1.channels.hbaseC.transactionCapacity = 10000 a1.channels.hbaseC.keep-alive = 20 a1.sinks.hbaseSink.type = asynchbase a1.sinks.hbaseSink.table = weblogs a1.sinks.hbaseSink.columnFamily = info a1.sinks.hbaseSink.serializer = org.apache.flume.sink.hbase.KfkAsyncHbaseEventSerializer a1.sinks.hbaseSink.channel = hbaseC a1.sinks.hbaseSink.serializer.payloadColumn = datetime,userid,searchname,retorder,cliorder,cliurl # ****************************flume + kafka****************************** a1.channels.kafkaC.type = memory a1.channels.kafkaC.capacity = 10000 a1.channels.kafkaC.transactionCapacity = 10000 a1.channels.kafkaC.keep-alive = 20 a1.sinks.kafkaSink.channel = kafkaC a1.sinks.kafkaSink.type = org.apache.flume.sink.kafka.KafkaSink a1.sinks.kafkaSink.brokerList = node5:9092,node6:9092,node7:9092 a1.sinks.kafkaSink.topic = weblogs a1.sinks.kafkaSink.zookeeperConnect = node5:2181,node6:2181,node7:2181 a1.sinks.kafkaSink.requiredAcks = 1 a1.sinks.kafkaSink.batchSize = 1 a1.sinks.kafkaSink.serializer.class = kafka.serializer.StringEncoder ``` #### 5.对日志数据进行格式处理 1）将文件中的tab更换成逗号 ```sh cat weblog.log|tr "\t" "," > weblog.log ``` 2）将文件中的空格更换成逗号 ```sh cat weblog2.log|tr " " "," > weblog.log ``` #### 6.自定义SinkHBase程序设计与开发 1）模仿SimpleAsyncHbaseEventSerializer自定义KfkAsyncHbaseEventSerializer实现类，修改一下代码即可。 ```java @Override public List<PutRequest> getActions() { List<PutRequest> actions = new ArrayList<PutRequest>(); if (payloadColumn != null) { byte[] rowKey; try { /*---------------------------代码修改开始---------------------------------*/ // 解析列字段 String[] columns = new String(this.payloadColumn).split(","); // 解析flume采集过来的每行的值 String[] values = new String(this.payload).split(","); for(int i=0;i < columns.length;i++){ byte[] colColumn = columns[i].getBytes(); byte[] colValue = values[i].getBytes(Charsets.UTF_8); // 数据校验：字段和值是否对应 if(colColumn.length != colValue.length) break; // 时间 String datetime = values[0].toString(); // 用户id String userid = values[1].toString(); // 根据业务自定义Rowkey rowKey = SimpleRowKeyGenerator.getKfkRowKey(userid,datetime); // 插入数据 PutRequest putRequest = new PutRequest(table, rowKey, cf, colColumn, colValue); actions.add(putRequest); /*---------------------------代码修改结束---------------------------------*/ } } catch (Exception e) { throw new FlumeException("Could not get row key!", e); } } return actions; } ``` 2）在SimpleRowKeyGenerator类中，根据具体业务自定义Rowkey生成方法 ```java public static byte[] getKfkRowKey(String userid, String datetime) throws UnsupportedEncodingException { return (userid + "-" + datetime + "-" + String.valueOf(System.currentTimeMillis())).getBytes("UTF8"); } ``` #### 7.自定义编译程序打jar包 1）在idea工具中，选择File——》ProjectStructrue  2）左侧选中Artifacts，然后点击右侧的+号，最后选择JAR——》From modules with dependencies  3）然后直接点击ok  4）删除其他依赖包，只把flume-ng-hbase-sink打成jar包就可以了。  ![](https://www.writebug.com/myre