使用Hadoop MapReduce进行大数据分析

Google在2001年推出图片搜索功能时，拥有2.5亿张索引图片。 不到十年后，这家搜索巨头就索引了超过100亿张图片。 每分钟有35个小时的内容上传到YouTube。 据说Twitter平均每天处理5500万条推文。 今年早些时候，其搜索功能每天记录6亿个查询。 这就是我们谈论大数据时的意思。

关于本系列
自从Java技术首次出现以来，Java开发环境发生了根本变化。 得益于成熟的开源框架和可靠的租用部署基础架构，现在可以快速，廉价地组装，测试，运行和维护Java应用程序。 在本系列中 ，Andrew Glover探索了使这种新的Java开发范例成为可能的技术和工具的范围。

如此大规模的数据曾经仅限于大型公司，大学和政府，这些实体能够购买非常昂贵的超级计算机，并需要员工来维持运行。 如今，随着存储成本的降低和处理能力的商品化，较小的公司和一些个人已经开始存储和挖掘相同的数据，从而引发了应用创新浪潮。

大数据革命的使能技术之一是MapReduce，这是Google开发的用于处理大规模分布式数据集的编程模型和实现。 在本文中，我介绍了Apache的开源MapReduce实现Hadoop，有人将其称为云计算的杀手级应用。

关于Hadoop

Apache的Hadoop框架本质上是一种用于分析大型数据集的机制，不一定需要将其存储在数据存储区中。 Hadoop抽象了MapReduce的海量数据分析引擎，使开发人员更易于使用。 Hadoop可以扩展到无数节点，并且可以处理与数据排序相关的所有活动和协调。

Hadoop的众多功能和配置使其成为了一个非常有用且功能强大的框架。 雅虎！ 无数其他组织发现它是一种有效的机制，可以分析大量的位和字节。 Hadoop也很容易在单个节点上工作。 您只需要一些数据即可分析和熟悉Java代码（包括泛型）。 Hadoop还可以与Ruby，Python和C ++一起使用。

有关MapReduce的更多信息
如果您是本系列的读者，那么您已经几次看到了MapReduce的实际应用。 在“ 使用CouchDB和Groovy的RESTClient进行REST结合 ”中，我演示了CouchDB如何利用MapReduce获取视图，然后在“ MongoDB：带有RDBMS移动的NoSQL数据存储 ”中再次使用了它，它是处理MongoDB文档的机制。

作为用于处理海量数据集的概念性框架，MapReduce经过了高度优化，可使用大量计算机来解决分布式问题。 顾名思义，该框架包含两个功能。 map功能旨在接收大数据输入并将其分成较小的部分，然后将其移交给可以对其执行某些操作的其他进程。 reduce功能可提取由map收集的各个答案，并将其呈现为最终输出。

在Hadoop中，您可以通过扩展Hadoop自己的基类来定义map并reduce实现。 这些实现由指定它们的配置以及输入和输出格式捆绑在一起。 Hadoop非常适合处理包含结构化数据的大文件。 Hadoop的一个特别方便的方面是它可以处理输入文件的原始解析，因此您一次只能处理一行。 因此，定义map功能实际上仅是确定要从输入的文本行中获取的内容的问题。

数据，数据无处不在！

美国政府产生了大量数据，其中许多数据对于普通公民来说是非常有趣的。 各种政府机构免费分发与美国经济健康状况和不断变化的社会人口状况有关的数据。 美国地质调查局（USGS）发布国际地震数据。

世界各地每天都发生多次小地震。 它们中的大多数都发生在地壳深处，因此没有人能感觉到它们，但听众仍在记录它们。 USGS以每周CSV（或逗号分隔值）文件的形式发布其地震数据。

平均每周文件不是很大-大约只有100KB左右。 尽管如此，它将作为学习Hadoop的基础。 记住这一点，虽然，Hadoop是能够处理更大的数据集。

追踪地震

我最近从USGS网站下载的CSV文件包含大约920行，如清单1所示：

清单1. USGS地震数据文件的行数

$> wc -l eqs7day-M1.txt 920 eqs7day-M1.txt

CVS文件的内容类似于清单2中的内容（即前两行）：

清单2. CVS文件的前两行

$> head -n 2 eqs7day-M1.txt 
Src,Eqid,Version,Datetime,Lat,Lon,Magnitude,Depth,NST,Region
ci,14896484,2,"Sunday, December 12, 2010 23:23:20 UTC",33.3040,-116.4130,1.0,11.70,22,"Southern California"

这就是我所说的信息丰富的文件，尤其是当您考虑到它总共920行时。 但是，我只想知道此文件报告的一周中的每一天发生了多少次地震。 然后，我想知道在那七天内哪个地区地震最多。

我的第一个想法是，我可以使用简单的grep命令来搜索每天的地震次数。 查看该文件，我看到它的数据从12月12日开始。所以我对该字符串进行了grep -c ，结果如清单3所示：

清单3. 12月12日发生了几次地震

$> grep -c 'December 12' eqs7day-M1.txt 
98

安装Hadoop
如果您以前尚未安装Hadoop，请立即进行安装。 首先， 下载最新的二进制文件 ，将其解压缩，然后在路径上设置Hadoop的bin目录。 这样做使您可以直接执行hadoop命令。 使用Hadoop要求您执行其hadoop命令，而不是像看到的那样调用java命令。 您可以将选项传递给hadoop命令，例如可以在其中找到Java二进制文件的位置（例如，表示map并reduce实现）。 就我而言，我创建了一个jar文件，并告诉Hadoop我想在jar中运行哪些作业。 我还将运行我的应用程序所需的所有其他二进制文件添加到Hadoop的类路径中。

现在我知道在12月12日有98项记录，即98项记录的地震。 我可以顺其自然，为12月11日，10日做grep ，依此类推。 但这对我来说听起来很乏味。 更糟糕的是，为了实现这一目标，我需要知道文件中的日期。 我并不十分在乎，在某些情况下，我可能无法访问该信息。 真的，我只想知道任何七天之内每一天的数字，而我可以使用Hadoop轻松获得该信息。

Hadoop只需要少量信息即可回答我的第一个和第二个问题：即要处理的输入以及如何处理map和reduce 。 我还必须提供一份将所有内容联系在一起的工作。 但是在开始编写该代码之前，我将花几分钟时间确保CSV数据一切正常。

用opencsv解析数据

除了地震CSV文件的第一行（即标题）之外，每行都是一系列数据值，以逗号分隔。 我主要对三个数据感兴趣：每次地震的日期，位置和震级。 为了获得这些数据，我将使用一个名为opencsv的漂亮开放源代码库，该库有助于解析CSV文件。

作为测试优先的人，我将从编写一个快速的JUnit测试开始，以验证是否可以从从CSV文件获得的示例行中获取所需的信息，如清单4所示：

清单4.解析CSV行

public class CSVProcessingTest {private final String LINE = "ci,14897012,2,\"Monday, December 13, 2010 " +"14:10:32 UTC\",33.0290,-115." +"5388,1.9,15.70,41,\"Southern California\"";@Testpublic void testReadingOneLine() throws Exception {String[] lines = new CSVParser().parseLine(LINE);assertEquals("should be Monday, December 13, 2010 14:10:32 UTC","Monday, December 13, 2010 14:10:32 UTC", lines[3]);assertEquals("should be Southern California","Southern California", lines[9]);assertEquals("should be 1.9", "1.9", lines[6]);}
}

如清单4所示 ， opencsv使使用逗号分隔的值变得非常容易。 解析器仅返回String的数组，因此可以获取位置值（只需回想一下Java语言中的数组和集合访问是从零开始的）。

转换日期格式

使用MapReduce时， map功能的工作是从一些值中选取要使用的值以及一些键。 也就是说， map主要用于并返回两个元素：键和值。 回到我以前的要求，我想首先找出每天发生多少地震。 因此，当我分析地震文件时，我将发出两个值：我的密钥将是日期，而该值将是一个计数器。 然后，我的reduce函数将对计数器求和（它们只是值为1的整数），从而为我提供了目标地震文件中日期发生的次数。

因为我对24小时感兴趣，所以我必须在每个文件中删除日期的时间方面。 在清单5中，我编写了一个快速测试，验证了如何将传入文件中的特定日期格式转换为更通用的24小时日期：

清单5.日期格式转换

@Test
public void testParsingDate() throws Exception {String datest = "Monday, December 13, 2010 14:10:32 UTC";SimpleDateFormat formatter = new SimpleDateFormat("EEEEE, MMMMM dd, yyyy HH:mm:ss Z");Date dt = formatter.parse(datest);formatter.applyPattern("dd-MM-yyyy");String dtstr = formatter.format(dt);assertEquals("should be 13-12-2010", "13-12-2010", dtstr);
}

在清单5中 ，我使用了SimpleDateFormat Java对象将日期String （采用UTC 2010年12月13日星期一14:10:32 UTC的CSV文件格式）格式化为更通用的13-12-2010。

Hadoop的地图和缩小

现在，我已经确定了如何处理CSV文件及其日期格式，现在可以开始实现map并reduce Hadoop中的功能了。 此过程需要了解Java泛型，因为Hadoop倾向于显式类型安全。

在使用Hadoop定义地图实现时，我仅扩展了Hadoop的Mapper类。 然后，我可以使用泛型为传出键和值指定显式类型。 type子句还描述了传入的键和值，在读取文件的情况下，它们分别是字节数和文本行。

我的EarthQuakesPerDateMapper类扩展了Hadoop的Mapper对象。 它显式地将输出键描述为Text对象，将其值描述为IntWritable ，它是Hadoop特定的类，本质上是一个整数。 还要注意，class子句中的前两种类型是LongWritable和Text ，它们分别是字节数和文本行。

由于类定义中的type子句，我在map方法中使用的参数类型以及该方法的输出都在context.write子句中设置。 如果尝试指定其他内容，则会遇到编译器问题，否则Hadoop将出错并显示一条描述类型不匹配的消息。

清单6.映射实现

public class EarthQuakesPerDateMapper extends Mapper<LongWritable, Text, Text, IntWritable> {@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException,InterruptedException {if (key.get() > 0) {try {CSVParser parser = new CSVParser();String[] lines = parser.parseLine(value.toString());SimpleDateFormat formatter = new SimpleDateFormat("EEEEE, MMMMM dd, yyyy HH:mm:ss Z");Date dt = formatter.parse(lines[3]);formatter.applyPattern("dd-MM-yyyy");String dtstr = formatter.format(dt);context.write(new Text(dtstr), new IntWritable(1));} catch (ParseException e) {}}}
}

清单6中的 map实现很简单：Hadoop基本上针对在输入文件中找到的每一行文本调用此类。 为了避免尝试处理CSV的标头，我首先检查字节数（ key对象）是否不为零。 然后，执行清单4和5中已经看到的操作：我获取传入的日期，对其进行转换，然后将其设置为传出密钥。 我还提供了一个计数：1.也就是说，我为每个日期编写了一个计数器，当调用reduce实现时，它将获得一个键和一组值。 在这种情况下，键将是日期及其值，如清单7所示：

清单7.映射输出和reduce输入的逻辑视图

"13-12-2010":[1,1,1,1,1,1,1,1]
"14-12-2010":[1,1,1,1,1,1]
"15-12-2010":[1,1,1,1,1,1,1,1,1]

注意， context.write(new Text(dtstr), new IntWritable(1)) （在清单6中 ）构建了清单7中所示的逻辑集合。 您可能已经知道， context是Hadoop数据结构，其中包含各种信息。 该context传递给reduce实现，该实现将采用这1个值并将其求和。 因此， reduce实现从逻辑上创建了清单8中所示的数据结构：

清单8. reduce输出的视图

"13-12-2010":8
"14-12-2010":6
"15-12-2010":9

我的reduce实现如清单9所示。 与Hadoop的Mapper ， Reducer被参数化：前两个参数是传入的键类型（ Text ）和值类型（ IntWritable ），后两个参数是输出类型：键和值，在这种情况下是相同的。

清单9. reduce实现

public class EarthQuakesPerDateReducer extends Reducer<Text, IntWritable, Text, IntWritable> {@Overrideprotected void reduce(Text key, Iterable<IntWritable> values, Context context)throws IOException, InterruptedException {int count = 0;for (IntWritable value : values) {count++;}context.write(key, new IntWritable(count));}
}

我的reduce实现非常简单。 正如我在清单7中指出的那样，传入值实际上是值的集合，在这种情况下，它意味着1个值的集合。 我要做的就是将它们加起来，然后写出代表日期和计数的新键值对。 然后，我的reduce代码基本上吐出了清单8中看到的行。 逻辑流程如下所示：

"13-12-2010":[1,1,1,1,1,1,1,1] -> "13-12-2010":8

当然，此清单的抽象形式是map -> reduce 。

定义Hadoop Job

现在，我已经编码了map并reduce实现，剩下要做的就是将所有内容链接到Hadoop Job 。 定义Job很简单：您提供输入和输出， map和reduce实现（如清单6和清单9所示 ）以及输出类型。 在这种情况下，我的输出类型与我的reduce实现所使用的输出类型相同。

清单10.一个工作关系映射并减少

public class EarthQuakesPerDayJob {public static void main(String[] args) throws Throwable {Job job = new Job();job.setJarByClass(EarthQuakesPerDayJob.class);FileInputFormat.addInputPath(job, new Path(args[0]));FileOutputFormat.setOutputPath(job, new Path(args[1]));job.setMapperClass(EarthQuakesPerDateMapper.class);job.setReducerClass(EarthQuakesPerDateReducer.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(IntWritable.class);System.exit(job.waitForCompletion(true) ? 0 : 1);}
}

在清单10中 ，我将所有内容与一个采用两个参数的main方法绑定在一起：地震CSV文件所在的目录，以及应写入结果报告的目录（Hadoop更喜欢创建此目录）。

为了执行这个小框架，我需要将这些类加起来。 我还需要告诉Hadoop在哪里可以找到opencsv二进制文件。 然后，我可以通过命令行执行Hadoop，如清单11所示：

清单11.执行Hadoop

$> export HADOOP_CLASSPATH=lib/opencsv-2.2.jar
$> hadoop jar target/quake.jar com.b50.hadoop.quake.EarthQuakesPerDayJob~/temp/mreduce/in/ ~/temp/mreduce/out

运行此代码，当Hadoop开始工作时，您会在屏幕上看到一堆文本。 请记住，与使用Hadoop处理的大型犬相比，我正在使用的CSV文件只是一只小狗。 根据您的处理能力，Hadoop应该在几秒钟内完成。

完成后，您几乎可以使用任何编辑器查看输出文件的内容。 另一个选择是直接使用hadoop命令，如清单12所示：

清单12.读取Hadoop的输出

$> hadoop dfs -cat part-r-00000 
05-12-2010      43
06-12-2010      143
07-12-2010      112
08-12-2010      136
09-12-2010      178
10-12-2010      114
11-12-2010      114
12-12-2010      79

如果您像我一样， 清单12中的第一件事就是每天的地震数量之多-仅12月9日就是178次！ 希望您还会注意到Hadoop确实做了我想要做的事情：整齐地列出了我范围内每个日期的地震发生次数。

编写另一个Mapper

接下来，我想找出发生地震的地点，并以某种方式快速测量哪个位置记录了我的日期范围内最多的地震。 嗯，正如您可能已经猜到的那样，Hadoop使此操作变得容易。 在这种情况下，关键不是日期，而是位置。 因此，我编写了一个新的Mapper类。

清单13.一个新的地图实现

public class EarthQuakeLocationMapper extends Mapper<LongWritable, Text, Text,IntWritable> {@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException,InterruptedException {if (key.get() > 0) {String[] lines = new CSVParser().parseLine(value.toString());context.write(new Text(lines[9]), new IntWritable(1));}}
}

在清单13中 ，我没有获取日期并进行转换，而是获取位置，该位置是CSV数组中的最后一个位置项。

除了要列出大量地点和地点之外，我还希望将结果限制在任何在7天之内发生10次以上地震的地点。

清单14.更多地震发生在哪里？

public class EarthQuakeLocationReducer extends Reducer<Text, IntWritable, Text,IntWritable> {@Overrideprotected void reduce(Text key, Iterable<IntWritable> values, Context context)throws IOException, InterruptedException {int count = 0;for (IntWritable value : values) {count++;}if (count >= 10) {context.write(key, new IntWritable(count));}}
}

清单14中的代码与清单9十分相似。 但是，在这种情况下，我将输出限制为10个或更多。 接下来，我可以将map绑定在一起，并通过另一个Job实现进行reduce ，添加内容，然后像往常一样执行Hadoop以获取新答案。

发出hadoop dfs命令将显示我请求的新值：

清单15.地震地点

$> hadoop dfs -cat part-r-00000 
Andreanof Islands, Aleutian Islands, Alaska     24
Arkansas        40
Baja California, Mexico 101
Central Alaska  74
Central California      68
Greater Los Angeles area, California    16
Island of Hawaii, Hawaii        16
Kenai Peninsula, Alaska 11
Nevada  15
Northern California     114
San Francisco Bay area, California      21
Southern Alaska 97
Southern California     115
Utah    19
western Montana 11

清单15的要点是什么？ 首先，从墨西哥到阿拉斯加的北美西海岸是个摇摇欲坠的地方。 其次，阿肯色州显然位于断层线附近，这是我没有意识到的。 最后，如果您居住在加利福尼亚北部或南部（许多软件开发人员都这样做），那么周围的地面大约每13分钟晃动一次。

结论

使用Hadoop分析数据既简单又高效，我什至还没有完全了解数据分析所提供的内容。 Hadoop实际上是设计为以分布式方式运行的，它可以处理运行map和reduce的各个节点的协调。 举例来说，在本文中，我在一个带有单个微弱文件的JVM中运行了Hadoop。

Hadoop本身就是一个了不起的工具，它周围还有一个完整的，不断发展的生态系统，从子项目到基于云的Hadoop服务。 Hadoop生态系统展示了该项目背后的丰富社区。 从该社区涌现出的许多工具证明了大数据分析作为全球业务活动的可行性。 借助Hadoop，分布式数据挖掘和分析可用于各种软件创新者和企业家，包括但不限于像Google和Yahoo!这样的大手笔。

转载自：https://blog.csdn.net/cuxiong8996/article/details/107152959