赞
踩
注:完整代码见此处
使用 Hadoop、Spark、Storm、MongoDB 进行大数据处理的相关开发与设计,主要包括:
1.基于 Hadoop 的大数据处理架构的搭建与应用开发;
2.Spark 分布式计算环境的搭建与应用开发;
3.二选一:安装配置 Storm,并进行简单开发;
安装配置 MongoDB,并进行简单开发。
本次课程设计采用的语言为Java,Web开发框架标准JSP/Springboot/SpringMVC可选。
课程设计采用的大数据集规模不低于10GB数据。完成的大数据处理分析模块6-10个。其中有3个以上属于复杂处理分析任务。
本次课程设计主要在台式机/笔记本电脑环境下完成。操作系统推荐以Linux为主。
随着大数据时代的到来,人们对于海量数据内所蕴含的价值需求越来越大,而现有的软件技术也为海量数据的挖掘、清洗与分析提供了可能。通过数据分析,可以将一大批杂乱无章的数据中的信息集中和提炼出来,从而驱动研究人员得出结论或做出推测。本系统便是基于这个目的,通过对大数据集的读取分析,从而得出一些较为简单的结论。
本系统采用Hadoop HDFS分布式文件处理系统来存储原始大数据集,采用专为大规模数据处理而设计的快速通用的计算引擎spark进行数据的分析处理,使用Java WEB框架springboot作为底层的应用开发框架,通过spring DATA JPA将数据的处理与分析结果保存在事务性数据库MYSQL中,并使用ECharts(一个优雅的使用 JavaScript 实现的开源可视化库)来作为数据的前端展示页面
本系统采用的均为项目生态中最为稳定常用的框架和软件。
HDFS有着高容错性(fault-tolerant)的特点,并且设计用来部署在低廉的(low-cost)硬件上。而且它提供高吞吐量(high throughput)来访问应用程序的数据,适合那些有着超大数据集(large data set)的应用程序。
Spark:Apache Spark 是专为大规模数据处理而设计的快速通用的计算引擎。Spark是UC Berkeley AMP lab (加州大学伯克利分校的AMP实验室)所开源的类Hadoop MapReduce的通用并行框架,Spark拥有Hadoop MapReduce所具有的优点;但不同于MapReduce的是——Job中间输出结果可以保存在内存中,从而不再需要读写HDFS,因此Spark能更好地适用于数据挖掘与机器学习等需要迭代的MapReduce的算法。
SpringBoot是由Pivotal团队在2013年开始研发、2014年4月发布第一个版本的全新开源的轻量级框架。它基于Spring4.0设计,不仅继承了Spring框架原有的优秀特性,而且还通过简化配置来进一步简化了Spring应用的整个搭建和开发过程。另外SpringBoot通过集成大量的框架使得依赖包的版本冲突,以及引用的不稳定性等问题得到了很好的解决。
MySQL是一个关系型数据库管理系统,由瑞典MySQL AB 公司开发,属于 Oracle 旗下产品。MySQL 是最流行的关系型数据库管理系统之一,在 WEB 应用方面,MySQL是最好的 RDBMS (Relational Database Management System,关系数据库管理系统) 应用软件之一。
Spring DATA JPA:Spring Data JPA是较大的Spring Data系列的一部分,可轻松实现基于JPA的存储库。该模块处理对基于JPA的数据访问层的增强支持。它使构建使用数据访问技术的Spring支持的应用程序变得更加容易。
ECharts:ECharts 是一个使用 JavaScript 实现的开源可视化库,涵盖各行业图表,满足各种需求。ECharts 遵循 Apache-2.0 开源协议,免费商用,且兼容当前绝大部分浏览器(IE8/9/10/11,Chrome,Firefox,Safari等)及兼容多种设备,可随时随地任性展示
运行电脑CPU为Intel® Core™ i5-8300H CPU @ 2.30GHz × 8,内存为8g,同时拥有40g的磁盘空间用于储存数据。因此从技术层面上说是可行的。
在纽约到处都是单向的、小的小巷,在任何特定的时间点,行人的数量几乎无法计算。更不用说路上塞满了汽车/摩托车/自行车了。在这样的城市生活,两地之间的交通出行可以坐出租车/Uber等。你不需要对交通或行人感到压力,但并不意味着你能及时到达目的地。所以你需要让你的司机尽可能走最短的路程。最短的时间,我们说的是时间。如果a线路比B线路长X公里*,但是比B线路快Y分钟*,那就选择B线路。因此需要知道哪条路线是最好的,我们需要能够预测在走特定路线的时候旅行会持续多长时间。以及其他值得我们去预测分析的东西。
纽约出租车数据分析是kaggle竞赛的著名赛题之一,也是新手学习数据分析的经典练习案例,项目数据可以从kaggle官网或者github上下载。
比赛数据集基于2016年纽约市黄色出租车出行记录数据,该数据可在Google Cloud Platform的Big Query中获得。该数据最初由纽约市出租车和豪华轿车委员会(TLC)发布。为了比赛的目的,对数据进行了采样和清理。
档案说明
基于HTML/HTML5和自选的Web页面开发框架,设计友好的Web界面,完成大数据的处理,问题发掘、统计和图表分析功能,从而实现一定的大数据可视化功能。
完成系统功能调试和系统测试,完成大数据分析系统部署,项目系统运行稳定,评估大数据处理与分析功能的性能和平均运行时间(分析代价),撰写相关设计报告文档。
系统由数据读入、数据预处理及初始化、数据分析、数据保存、欢迎页展示、各个分析结果图表展示,初始化分析结果功能组成,具体如下图所示:
图1 功能模块图
数据读入与预处理及初始化:将.csv文件先去掉第一行字段名,然后逐行读入spark生成Cab类型的RDD,Cab是一个java类用于存储一条出租车的所有字段信息,读取时去掉字段不完整的信息。
数据分析:运用spark进行数据分析,从数据中获取有价值的信息:
(1) 一个月内每日乘车趋势乘车距离趋势
(2) 各个乘车人数区间与时间段的对应关系
(3) 乘车距离的分布情况
(4) 长短途乘客比例(规定行驶路程5公里以上为长途车)
(5) 一天中各个时间段的乘车人数
(6) 每周乘车趋势
数据保存:通过Spring DATA JPA数据分析结果保存在mysql中
欢迎页展示:输入网址进入分析系统欢迎页,欢迎页分别列出了各个分析结果图的链接和数据初始化的链接
各个分析结果的图表展示:前端拿到保存在Mysql的结果数据进行可视化分析
初始化分析:重新对数据进行分析并将分析结果保存
将纽约出租车大数据集中的每个.csv文件先去掉第一行字段名,然后逐行读入spark生成Cab类型的RDD,Cab是一个java类用于存储一条出租车的所有字段信息,读取时去掉字段不完整的信息。
Spark的核心数据结构是RDD(弹性分布式数据集),其本质上是一个不可变的、分布式的元素集合。Spark中的RDD都被分为多个分区,这些分区运行在不同的节点上,RDD中可以包含Python、Java、Scala中任意类型的对象,在本项目中便是包含一个Cab类型的Java对象。
RDD有两个操作:转化操作(Transformation)、行动操作(Action)
Cab对象中的属性全为String型,分别是纽约出租车大数据集中的每一个字段,这意味着一个Cab对象接收数据集中的一条完整数据记录
再将数据读入并生成RDD时,若发现null值或者不合法的值,全部统一用String型的“null”代替,保证读取时的完整性与健壮性
将所有的Cab类型数据记录全部封装成RDD型的对象cabrecord,并返回它
package com.lhr13.newyorkcab.data; import com.lhr13.newyorkcab.pojo.Cab; import org.apache.spark.api.java.JavaRDD; import org.apache.spark.api.java.JavaSparkContext; import org.apache.spark.api.java.function.Function; import scala.Serializable; public class CatchData implements Serializable { public JavaRDD<Cab> CatchData(JavaSparkContext sc) throws Exception { JavaRDD<String> file = sc.textFile("hdfs://localhost:9000/newYorkCab/trip_data_10.csv").cache(); JavaRDD<Cab> cabrecord = file.map((Function<String, Cab>) s -> { Cab cab; String[] strings = s.split(","); if (strings.length == 14) { // 防止空值 for (int i = 0; i < 14; i++) { if (strings[i] == null) { strings[i] = "null"; } } cab = new Cab(strings[0], strings[1], strings[2], strings[3], strings[4], strings[5], strings[6], strings[7], strings[8], strings[9], strings[10], strings[11], strings[12], strings[13]); } else { String nullString = "null"; cab = new Cab(nullString, nullString, nullString, nullString, nullString, nullString, nullString, nullString, nullString, nullString, nullString, nullString, nullString, nullString); } return cab; }); return cabrecord; } }
通过对元数据字段中Pickup_datetime(使用电表的日期和时间)和passenger_count(车辆中的乘客人数)的分析,从中总结出各个时间段的区间以及各个乘车人数的区间,再对两者进行整体的统计分析,最后得出结论
RDD的转化操作是返回一个新的RDD的操作,转化操作常见的有map、filter、union等,转化出来的RDD是惰性求值的,只有在行动操作中这些RDD才会被计算。通过转化操作,可以从已有的RDD中派生出新的RDD,Spark会使用谱系图来记录不同RDD之间的依赖关系,Spark需要用这些信息来按需计算每个RDD,也可以依靠谱系图在持久化的RDD丢失部分数据时恢复所丢失的数据。
GetPlatoonTime类会首先运行CatchData类,从中拿到封装好了的cabrecord类,使用filter函数去除对象为null或者Pickup_datetime、passenger_count为“null”的cab对象,再使用map函数重新返回元素均为String型的RDD,每一个String元素分为两部分,前面的为区间在[11,15]的数字,分别代表morning、midday、afternoon、evening、midnight、before dawn五个时间段,最后为区间在[1,3]的数字,分别代表less than 3 passagers、3-5passagers、more than 5 passagers三种乘客数量区间。如“112”代表“清晨且乘客数在3-5之间”
最后通过countByValue函数统计出各个特征的数据总数返回
package com.lhr13.newyorkcab.data; import com.lhr13.newyorkcab.pojo.Cab; import org.apache.spark.SparkConf; import org.apache.spark.api.java.JavaRDD; import org.apache.spark.api.java.JavaSparkContext; import org.apache.spark.api.java.function.Function; import scala.Serializable; import java.util.Map; public class GetPlatoonTime implements Serializable { public Map<String, Long> run() throws Exception { SparkConf conf = new SparkConf().setAppName("NewYarkCab2").setMaster("local"); System.setProperty("hadoop.home.dir", "/usr/local/hadoop"); JavaSparkContext sc = new JavaSparkContext(conf); JavaRDD<Cab> cabrecord = new CatchData().CatchData(sc); JavaRDD<Cab> wash = cabrecord.filter((Function<Cab, Boolean>) cab -> cab != null && cab.getPickup_datatime() != "null" && cab.getPassenger_count() != "null" && Integer.valueOf(cab.getPassenger_count()) > 0); JavaRDD<String> more = wash.map((Function<Cab, String>) cab -> { String dayornight; String passcount = null; Integer time = Integer.valueOf(cab.getPickup_datatime().substring(11,13)); if (time > 6 && time <= 10) { dayornight = "11"; }else if (time > 10 && time <= 14) { dayornight = "12"; }else if (time > 14 && time <= 18) { dayornight = "13"; }else if (time > 18 && time <= 22) { dayornight = "14"; }else if (time > 22 || time <= 2){ dayornight = "15"; }else { dayornight = "16"; } Integer count = Integer.valueOf(cab.getPassenger_count()); if (count >= 1 && count < 3) { passcount = "1"; }else if (count >= 3 && count < 6){ passcount = "2"; }else if (count >= 6) { passcount = "3"; } return passcount + "," + dayornight; }); Map<String, Long> morePassagers = more.countByValue(); sc.close(); return morePassagers; } }
通过对纽约出租车大数据集中字段Pickup_datatime(使用电表的日期和时间)进行分析,通过算法将日期转换成相应的星期数,最后统计不同星期数的记录结果结果
有时候需要对数据集进行实际的计算,从而得到所期望的结果。行动操作(action)是第二类型的操作,它会将最终执行的结果返回到驱动器程序,或者写入外部储存系统。由于行动操作需要实际的输出,所以其会强制执行那些求之必须用到的RDD的转换操作,或者可以说,转换操作的终点便是行动操作。
类会首先运行CatchData类,从中拿到封装好了的cabrecord类,使用filter函数去除对象为null或者Pickup_datetime为“null”的cab对象。再通过map函数生成一个只包含String型的上车日期的RDD。再一次使用map函数,函数中定义weeknum函数,这是专门用来将日期数转化为星期数的函数,通过这个map函数得到一个String型的星期数的RDD,最后使用行动操作函数countByValue,对RDD中星期数相同的元素数目进行统计,最后生成一个元素均为<String,Long>型的Map对象
package com.lhr13.newyorkcab.data; import com.lhr13.newyorkcab.pojo.Cab; import org.apache.spark.SparkConf; import org.apache.spark.api.java.JavaRDD; import org.apache.spark.api.java.JavaSparkContext; import org.apache.spark.api.java.function.Function; import scala.Serializable; import java.text.ParseException; import java.text.SimpleDateFormat; import java.util.Calendar; import java.util.Date; import java.util.Map; public class WeekBoomDay implements Serializable { public Map<String, Long> run() throws Exception { SparkConf conf = new SparkConf().setAppName("NewYarkCab2").setMaster("local"); System.setProperty("hadoop.home.dir", "/usr/local/hadoop"); JavaSparkContext sc = new JavaSparkContext(conf); JavaRDD<Cab> cabrecord = new CatchData().CatchData(sc); JavaRDD<Cab> wash = cabrecord.filter((Function<Cab, Boolean>) cab -> cab.getPickup_datatime() != "null" && cab != null); JavaRDD<String> day = wash.map((Function<Cab, String>) cab -> cab.getPickup_datatime().substring(0,10)); JavaRDD<String> day2 = day.map(new Function<String, String>() { @Override public String call(String dates) throws Exception { int numday = weeknum(dates); if (numday == 1) { return "1"; }else if (numday == 2) { return "2"; }else if (numday == 3) { return "3"; }else if (numday == 4) { return "4"; }else if (numday == 5) { return "5"; }else if (numday == 6) { return "6"; }else { return "7"; } } private int weeknum(String dates) { Calendar cal = Calendar.getInstance(); SimpleDateFormat f = new SimpleDateFormat("yyyy-MM-dd"); Date d=null; try { d=f.parse(dates); } catch (ParseException e) { e.printStackTrace(); } cal.setTime(d); int w=cal.get(Calendar.DAY_OF_WEEK)-1; if(w==0) w=7; return w; } }); Map<String, Long> map = day2.countByValue(); sc.close(); return map; } }
通过对纽约出租车大数据集中字段Pickup_datatime(使用电表的日期和时间)进行分析,统计不同天数的打车次数.
BoomDay类会首先运行CatchData类,从中拿到封装好了的cabrecord类,使用filter函数去除对象为null或者Pickup_datetime为“null”的cab对象。再通过map函数生成一个只包含String型的上车日期的RDD。
对RDD中天数相同的元素数目进行统计,最后生成一个元素均为<String,Long>型的Map对象
package com.lhr13.newyorkcab.data; import com.lhr13.newyorkcab.dao.DayDAO; import com.lhr13.newyorkcab.pojo.Cab; import org.apache.spark.SparkConf; import org.apache.spark.api.java.JavaRDD; import org.apache.spark.api.java.JavaSparkContext; import org.apache.spark.api.java.function.Function; import org.springframework.beans.factory.annotation.Autowired; import scala.Serializable; import java.util.Map; public class BoomDay implements Serializable { @Autowired private DayDAO dayDAO; public Map<String, Long> run() throws Exception { SparkConf conf = new SparkConf().setAppName("NewYarkCab2").setMaster("local"); System.setProperty("hadoop.home.dir", "/usr/local/hadoop"); JavaSparkContext sc = new JavaSparkContext(conf); JavaRDD<Cab> cabrecord = new CatchData().CatchData(sc); JavaRDD<Cab> wash = cabrecord.filter((Function<Cab, Boolean>) cab -> cab.getPickup_datatime() != "null" && cab != null); JavaRDD<String> day = wash.map((Function<Cab, String>) cab -> cab.getPickup_datatime().substring(8,10)); Map<String, Long> map = day.countByValue(); sc.close(); return map; } }
通过对纽约出租车大数据集中字段Trip_distance(行车距离)进行分析,统计不同乘车距离的打车次数.
Distance类会首先运行CatchData类,从中拿到封装好了的cabrecord类,使用filter函数去除对象为null或者trip_distance(载客距离)为“null”的cab对象。再通过map函数生成一个只包含String型的载客距离的RDD。
为防止样本过多,所以对乘车公里数进行四舍五入
对RDD中天数相同的距离进行统计,最后生成一个元素均为<String,Long>型的Map对象
package com.lhr13.newyorkcab.data; import com.lhr13.newyorkcab.pojo.Cab; import org.apache.spark.SparkConf; import org.apache.spark.api.java.JavaRDD; import org.apache.spark.api.java.JavaSparkContext; import org.apache.spark.api.java.function.Function; import org.springframework.stereotype.Component; import scala.Serializable; import java.util.Map; @Component public class DistanceCount implements Serializable { public Map<String, Long> run() throws Exception { SparkConf conf = new SparkConf().setAppName("NewYarkCab2").setMaster("local"); System.setProperty("hadoop.home.dir", "/usr/local/hadoop"); JavaSparkContext sc = new JavaSparkContext(conf); JavaRDD<Cab> cabrecord = new CatchData().CatchData(sc); JavaRDD<Cab> wash = cabrecord.filter((Function<Cab, Boolean>) cab -> cab != null && cab.getTrip_distance() != "null" ); JavaRDD<String> dtdistinct = wash.map((Function<Cab, String>) cab -> String.valueOf(Math.round(Double.valueOf(cab.getTrip_distance())))); Map<String, Long> spmap = dtdistinct.countByValue(); sc.close(); return spmap; } }
通过对纽约出租车大数据集的Trip_distance进行分析,统计长短途乘车的比例
Distance类会首先运行CatchData类,从中拿到封装好了的cabrecord类,使用filter函数去除对象为null或者trip_distance(载客距离)为“null”的cab对象。再通过map函数生成一个只包含String型的载客距离的RDD,元素只有“long”和“short”两种,分别表示长途车和短途车,划定界限是:大于5km为长途,小于等于5km为短途
对RDD中天数相同的距离进行统计,最后生成一个元素均为<String,Long>型的Map对象
package com.lhr13.newyorkcab.data; import com.lhr13.newyorkcab.dao.DayDAO; import com.lhr13.newyorkcab.pojo.Cab; import org.apache.spark.SparkConf; import org.apache.spark.api.java.JavaRDD; import org.apache.spark.api.java.JavaSparkContext; import org.apache.spark.api.java.function.Function; import org.springframework.beans.factory.annotation.Autowired; import scala.Serializable; import java.util.Map; public class LongOrShort implements Serializable { @Autowired private DayDAO dayDAO; public Map<String, Long> run() throws Exception { SparkConf conf = new SparkConf().setAppName("NewYarkCab2").setMaster("local"); System.setProperty("hadoop.home.dir", "/usr/local/hadoop"); JavaSparkContext sc = new JavaSparkContext(conf); JavaRDD<Cab> cabrecord = new CatchData().CatchData(sc); JavaRDD<Cab> wash = cabrecord.filter((Function<Cab, Boolean>) cab -> cab.getTrip_distance() != "null" && cab != null); JavaRDD<String> distanc = wash.map((Function<Cab, String>) cab -> { String distance = cab.getTrip_distance(); if (Double.valueOf(distance) > 5.00) { return "long"; }else { return "short"; } }); Map<String, Long> map = distanc.countByValue(); sc.close(); return map; } }
通过对纽约出租车大数据集的Pickup_datatime进行分析,统计一天之中打车分布情况
BoonDay类会首先运行CatchData类,从中拿到封装好了的cabrecord类,使用filter函数去除对象为null或者Pickup_datetime为“null”的cab对象。再通过map函数生成一个只包含String型的上车时间的RDD。
对RDD中小时数相同的元素数目进行统计,最后生成一个元素均为<String,Long>型的Map对象
package com.lhr13.newyorkcab.data; import com.lhr13.newyorkcab.pojo.Cab; import org.apache.spark.SparkConf; import org.apache.spark.api.java.JavaRDD; import org.apache.spark.api.java.JavaSparkContext; import org.apache.spark.api.java.function.Function; import scala.Serializable; import java.util.Map; public class MoreCustomerTime implements Serializable { public Map<String, Long> run() throws Exception { SparkConf conf = new SparkConf().setAppName("NewYarkCab2").setMaster("local"); System.setProperty("hadoop.home.dir", "/usr/local/hadoop"); JavaSparkContext sc = new JavaSparkContext(conf); JavaRDD<Cab> cabrecord = new CatchData().CatchData(sc); JavaRDD<Cab> wash = cabrecord.filter((Function<Cab, Boolean>) cab -> cab.getPickup_datatime() != "null" && cab != null); JavaRDD<String> more = wash.map((Function<Cab, String>) cab -> cab.getPickup_datatime().substring(11,13)); Map<String, Long> morePassagers = more.countByValue(); sc.close(); return morePassagers; } }
系统测试机型
笔记本型号:戴尔G7-7588
平台:Ubuntu
处理器核心:八核16线程
CPU类型:Intel® Core™ i5-8300H CPU @ 2.30GHz × 8
操作系统:Ubuntu Linux 16.02
显卡类型:GeForce GTX 1050 Ti/PCIe/SSE2
硬盘容量:1T SSD
内存容量:8G
安装Hadoop/Spark/Storm/Mongdb等软件,安装后截图如下:
采用大数据存储工具HDFS转入数据。转入后总的数据记录74901541条。
在浏览器中输入链接:http://localhost:8080/show/welcome
进入系统主页面如下图所示:
图1 欢迎页
图二 数据初始化页面
点击对应分析功能模块链接后,分别进入对应的大数据分析功能子模块网页。
(1) 测试输入
由于电脑配置的制约,在进行10g的大数据测试时会耗费大量的时间,而且系统经常性卡死,为了方便测试,只选用一个1g数据集。
(2) 运行结果
代码执行后,运行结果界面如下图所示:
结论:从图中可以得出,一个月内的乘车数量呈周期性变化
结论:在打车的人数中,三人以下的占据绝对比例。同时一天之内,人们更加倾向于在晚上打车,可能是因为下班回家,或者放学后朋友聚会
结论:短途车占据绝对比例,可能因为长途人们更倾向于搭乘节省经济的公交车
结论:凌晨五点是打车人数最低的时候,晚上七点是高峰期,在早八点到下午三点乘车人数相对平衡
结论:一周内,周四是打车人数相对最多的,周天周一相对较少。我认为这可能和假期与工作时间与人们的情绪有关,周四第二天周五意味着放假,而周日周一意味着新的忙碌的一周的开始
结论:乘车历程中,次数最多的集中在2-6km
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。