大数据处理与开发课程设计——纽约出租车大数据分析_LHR13_大数据技术课程设计

网络 02-07 5363

注：完整代码见此处

一、设计目的综合应用所学的Hadoop/Spark/Storm/Mongdb等技术，设计并实现一个较为完整的小型大数据处理和分析系统。通过系统分析、系统设计、编程调试、撰写实验报告等环节，初步掌握大数据分析软件系统设计的方法和步骤，灵活运用Java高级编程等语言进行软件开发，提高分析问题和解决问题的能力。提高Java高级程序设计水平，培养必要的工程实践动手能力。理解 HDFS 文件系统的基本结构，理解 MapReduce 的工作原理，可以搭建 Hadoop 开发环境，并能进行相应的编程开发，具备运用 Hadoop 架构实现大数据处理的能力。理解 Spark 大数据计算框架、架构、计算模型，掌握 Spark 的关键技术，能够运用掌握的相关技术完成 Spark 开发环境的搭建，进行实时数据分析、机器学习与图形处理等。理解 Storm 的基本特性，理解 MongoDB 的基本知识，掌握 Storm、MongoDB 的基本操作，学会开发环境的搭建，并能够运用所学技术进行相应的应用开发。二、设计内容 2.1设计内容概述及设计要求

使用 Hadoop、Spark、Storm、MongoDB 进行大数据处理的相关开发与设计，主要包括： 1.基于 Hadoop 的大数据处理架构的搭建与应用开发； 2.Spark 分布式计算环境的搭建与应用开发； 3.二选一：安装配置 Storm，并进行简单开发；安装配置 MongoDB，并进行简单开发。本次课程设计采用的语言为Java，Web开发框架标准JSP/Springboot/SpringMVC可选。课程设计采用的大数据集规模不低于10GB数据。完成的大数据处理分析模块6-10个。其中有3个以上属于复杂处理分析任务。本次课程设计主要在台式机/笔记本电脑环境下完成。操作系统推荐以Linux为主。

2.2系统需求分析 (1) 目的

随着大数据时代的到来，人们对于海量数据内所蕴含的价值需求越来越大，而现有的软件技术也为海量数据的挖掘、清洗与分析提供了可能。通过数据分析，可以将一大批杂乱无章的数据中的信息集中和提炼出来，从而驱动研究人员得出结论或做出推测。本系统便是基于这个目的，通过对大数据集的读取分析，从而得出一些较为简单的结论。

(2) 系统概述

本系统采用Hadoop HDFS分布式文件处理系统来存储原始大数据集，采用专为大规模数据处理而设计的快速通用的计算引擎spark进行数据的分析处理，使用Java WEB框架springboot作为底层的应用开发框架，通过spring DATA JPA将数据的处理与分析结果保存在事务性数据库MYSQL中，并使用ECharts（一个优雅的使用 JavaScript 实现的开源可视化库）来作为数据的前端展示页面

(3) 可行性分析

本系统采用的均为项目生态中最为稳定常用的框架和软件。 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的磁盘空间用于储存数据。因此从技术层面上说是可行的。

2.3设计使用的大数据集

在纽约到处都是单向的、小的小巷，在任何特定的时间点，行人的数量几乎无法计算。更不用说路上塞满了汽车/摩托车/自行车了。在这样的城市生活，两地之间的交通出行可以坐出租车/Uber等。你不需要对交通或行人感到压力，但并不意味着你能及时到达目的地。所以你需要让你的司机尽可能走最短的路程。最短的时间，我们说的是时间。如果a线路比B线路长X公里*，但是比B线路快Y分钟*，那就选择B线路。因此需要知道哪条路线是最好的，我们需要能够预测在走特定路线的时候旅行会持续多长时间。以及其他值得我们去预测分析的东西。纽约出租车数据分析是kaggle竞赛的著名赛题之一，也是新手学习数据分析的经典练习案例，项目数据可以从kaggle官网或者github上下载。比赛数据集基于2016年纽约市黄色出租车出行记录数据，该数据可在Google Cloud Platform的Big Query中获得。该数据最初由纽约市出租车和豪华轿车委员会（TLC）发布。为了比赛的目的，对数据进行了采样和清理。档案说明

train.csv-训练集（包含1458644旅行记录） ? test.csv-测试集（包含625134个行程记录）sample_submission.csv-格式正确的样本提交文件资料栏位 ? Medallion：UUIDHack_licence:UUIDVendor_id：比率id-每次旅行的唯一标识符vendor_id-指示与行程记录关联的提供商的代码Pickup_datetime-使用电表的日期和时间dropoff_datetime-仪表断开连接的日期和时间passenger_count-车辆中的乘客人数（驾驶员输入的值）Pickup_longitude-仪表使用的经度Pickup_Latitude-仪表使用的纬度dropoff_longitude-仪表脱离的经度dropoff_latitude-仪表脱离的纬度store_and_fwd_flag-此标志指示在发送给卖方之前，由于车辆没有与服务器的连接，因此行程记录是否已保存在车辆内存中-Y = 存储并转发；N =不是存储和继续旅行trip_duration-行程持续时间（以秒为单位）trip_distance：载客距离 2.4大数据可视化设计说明

基于HTML/HTML5和自选的Web页面开发框架，设计友好的Web界面，完成大数据的处理，问题发掘、统计和图表分析功能，从而实现一定的大数据可视化功能。

2.5设计结果

完成系统功能调试和系统测试，完成大数据分析系统部署，项目系统运行稳定，评估大数据处理与分析功能的性能和平均运行时间（分析代价），撰写相关设计报告文档。

三、概要设计 3.1功能模块图

系统由数据读入、数据预处理及初始化、数据分析、数据保存、欢迎页展示、各个分析结果图表展示，初始化分析结果功能组成，具体如下图所示：

图1 功能模块图 3.2各个分模块功能概要描述

数据读入与预处理及初始化：将.csv文件先去掉第一行字段名，然后逐行读入spark生成Cab类型的RDD，Cab是一个java类用于存储一条出租车的所有字段信息，读取时去掉字段不完整的信息。

数据分析：运用spark进行数据分析，从数据中获取有价值的信息： (1) 一个月内每日乘车趋势乘车距离趋势 (2) 各个乘车人数区间与时间段的对应关系 (3) 乘车距离的分布情况 (4) 长短途乘客比例（规定行驶路程5公里以上为长途车） (5) 一天中各个时间段的乘车人数 (6) 每周乘车趋势

数据保存：通过Spring DATA JPA数据分析结果保存在mysql中

欢迎页展示：输入网址进入分析系统欢迎页，欢迎页分别列出了各个分析结果图的链接和数据初始化的链接

各个分析结果的图表展示：前端拿到保存在Mysql的结果数据进行可视化分析

初始化分析：重新对数据进行分析并将分析结果保存

四．分模块详细设计 4.1 大数据分析模块一—数据的初始化（CatchData类）（1）大数据分析功能模块功能描述及涉及到的元数据

将纽约出租车大数据集中的每个.csv文件先去掉第一行字段名，然后逐行读入spark生成Cab类型的RDD，Cab是一个java类用于存储一条出租车的所有字段信息，读取时去掉字段不完整的信息。

（2）对应的大数据处理与分析算法描述

Spark的核心数据结构是RDD（弹性分布式数据集），其本质上是一个不可变的、分布式的元素集合。Spark中的RDD都被分为多个分区，这些分区运行在不同的节点上，RDD中可以包含Python、Java、Scala中任意类型的对象，在本项目中便是包含一个Cab类型的Java对象。 RDD有两个操作：转化操作（Transformation）、行动操作（Action） Cab对象中的属性全为String型，分别是纽约出租车大数据集中的每一个字段，这意味着一个Cab对象接收数据集中的一条完整数据记录再将数据读入并生成RDD时，若发现null值或者不合法的值，全部统一用String型的“null”代替，保证读取时的完整性与健壮性将所有的Cab类型数据记录全部封装成RDD型的对象cabrecord，并返回它

（3）关键设计/编程与重点代码实现 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; } } 4.2 大数据分析模块二-- 各个乘车人数区间在不同时间段的分布（GetPlatoonTime类） (1) 大数据分析功能模块功能描述及涉及到的元数据

通过对元数据字段中Pickup_datetime（使用电表的日期和时间）和passenger_count（车辆中的乘客人数）的分析，从中总结出各个时间段的区间以及各个乘车人数的区间，再对两者进行整体的统计分析，最后得出结论

(2) 对应的大数据处理与分析算法描述

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函数统计出各个特征的数据总数返回

(3) 关键设计/编程与重点代码实现 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; } } 4.3 大数据分析模块三—每周乘车趋势（WeekBoomDay类） (1) 大数据分析功能模块功能描述及涉及到的元数据

通过对纽约出租车大数据集中字段Pickup_datatime(使用电表的日期和时间)进行分析，通过算法将日期转换成相应的星期数，最后统计不同星期数的记录结果结果

(2) 对应的大数据处理与分析算法描述

有时候需要对数据集进行实际的计算，从而得到所期望的结果。行动操作（action）是第二类型的操作，它会将最终执行的结果返回到驱动器程序，或者写入外部储存系统。由于行动操作需要实际的输出，所以其会强制执行那些求之必须用到的RDD的转换操作，或者可以说，转换操作的终点便是行动操作。类会首先运行CatchData类，从中拿到封装好了的cabrecord类，使用filter函数去除对象为null或者Pickup_datetime为“null”的cab对象。再通过map函数生成一个只包含String型的上车日期的RDD。再一次使用map函数，函数中定义weeknum函数，这是专门用来将日期数转化为星期数的函数，通过这个map函数得到一个String型的星期数的RDD，最后使用行动操作函数countByValue，对RDD中星期数相同的元素数目进行统计，最后生成一个元素均为<String，Long>型的Ｍａｐ对象

(3) 关键设计/编程与重点代码实现 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; } } 4.4 大数据分析模块四—一个月内每日乘车趋势乘车距离趋势（ＢｏｏｎＤａｙ类） (1) 大数据分析功能模块功能描述及涉及到的元数据

通过对纽约出租车大数据集中字段Pickup_datatime(使用电表的日期和时间)进行分析，统计不同天数的打车次数.

(2) 对应的大数据处理与分析算法描述

BoomDay类会首先运行CatchData类，从中拿到封装好了的cabrecord类，使用filter函数去除对象为null或者Pickup_datetime为“null”的cab对象。再通过map函数生成一个只包含String型的上车日期的RDD。对RDD中天数相同的元素数目进行统计，最后生成一个元素均为<String，Long>型的Ｍａｐ对象

(3) 关键设计/编程与重点代码实现 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; } } 4.5 大数据分析模块四—一个月内每日乘车趋势乘车距离趋势（DistanceCount类） (1) 大数据分析功能模块功能描述及涉及到的元数据

通过对纽约出租车大数据集中字段Trip_distance（行车距离）进行分析，统计不同乘车距离的打车次数.

(2) 对应的大数据处理与分析算法描述

Distance类会首先运行CatchData类，从中拿到封装好了的cabrecord类，使用filter函数去除对象为null或者trip_distance（载客距离）为“null”的cab对象。再通过map函数生成一个只包含String型的载客距离的RDD。为防止样本过多，所以对乘车公里数进行四舍五入对RDD中天数相同的距离进行统计，最后生成一个元素均为<String，Long>型的Ｍａｐ对象

通过对纽约出租车大数据集的Trip_distance进行分析，统计长短途乘车的比例

(2) 对应的大数据处理与分析算法描述

Distance类会首先运行CatchData类，从中拿到封装好了的cabrecord类，使用filter函数去除对象为null或者trip_distance（载客距离）为“null”的cab对象。再通过map函数生成一个只包含String型的载客距离的RDD，元素只有“long”和“short”两种，分别表示长途车和短途车，划定界限是：大于5km为长途，小于等于5km为短途对RDD中天数相同的距离进行统计，最后生成一个元素均为<String，Long>型的Ｍａｐ对象

(3) 关键设计/编程与重点代码实现 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; } } 4.7 大数据分析模块四—一个月内每日乘车趋势乘车距离趋势（MoreCustomerTime类） (1) 大数据分析功能模块功能描述及涉及到的元数据

通过对纽约出租车大数据集的Pickup_datatime进行分析，统计一天之中打车分布情况

(2) 对应的大数据处理与分析算法描述（含机器学习方法）

ＢｏｏｎＤａｙ类会首先运行CatchData类，从中拿到封装好了的cabrecord类，使用filter函数去除对象为null或者Pickup_datetime为“null”的cab对象。再通过map函数生成一个只包含String型的上车时间的RDD。对RDD中小时数相同的元素数目进行统计，最后生成一个元素均为<String，Long>型的Ｍａｐ对象

系统测试机型笔记本型号：戴尔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

5.1 开发环境及大数据转换

安装Hadoop/Spark/Storm/Mongdb等软件，安装后截图如下：

采用大数据存储工具HDFS转入数据。转入后总的数据记录74901541条。 5.2 系统主界面

在浏览器中输入链接：http://localhost:8080/show/welcome

进入系统主页面如下图所示：

图1 欢迎页

图二数据初始化页面

点击对应分析功能模块链接后，分别进入对应的大数据分析功能子模块网页。

5.3大数据分析模块一正常测试输入和运行结果

(1) 测试输入由于电脑配置的制约，在进行10g的大数据测试时会耗费大量的时间，而且系统经常性卡死，为了方便测试，只选用一个1g数据集。

(2) 运行结果代码执行后，运行结果界面如下图所示：

结论：从图中可以得出，一个月内的乘车数量呈周期性变化