应用开发之linq和ef_应用开发

本系列的前两篇文章重点介绍了针对单节点和多节点集群的Hadoop的安装和配置。 这篇最后的文章探讨了Hadoop中的编程-特别是Ruby语言中的map和reduce应用程序的开发。 我之所以选择Ruby&＃xff0c;是因为它首先是您应该了解的一种很棒的面向对象的脚本语言&＃xff0c;其次&＃xff0c;您将在“ 相关主题”部分中找到大量参考资料&＃xff0c;以获取有关 Java™和Python语言的教程。 通过对MapReduce编程的探索&＃xff0c;我还将向您介绍流应用程序编程接口&＃xff08;API&＃xff09;。 该API提供了使用Java语言以外的其他语言开发应用程序的方法。

首先&＃xff0c;从功能的角度出发&＃xff0c;简要介绍一下映射和缩减&＃xff0c;然后更深入地研究Hadoop编程模型及其构架&＃xff0c;分配&＃xff0c;管理工作的架构和元素。

地图的起源与归约

那么&＃xff0c;启发MapReduce编程范例的功能要素是什么&＃xff1f; 1958年&＃xff0c;约翰·麦卡锡&＃xff08;John McCarthy&＃xff09;发明了一种名为Lisp的语言&＃xff0c;该语言既实现了数值计算又实现了符号计算&＃xff0c;但是以递归的形式实现&＃xff0c;而这种形式对于当今使用的大多数语言都是陌生的。 &＃xff08;实际上&＃xff0c;在Wikipedia上Lisp有着令人着迷的历史&＃xff0c;其中包括有用的教程&＃xff0c;非常值得阅读。&＃xff09;Lisp最早在IBM®704上实现&＃xff0c;这是第一台大规模生产的计算机&＃xff0c;它还支持另一个古老的收藏&＃xff1a;FORTRAN。

map函数起源于像Lisp这样的功能语言&＃xff0c;但现在在许多其他语言中很常见&＃xff0c;它是函数在一系列元素上的应用。 这是什么意思&＃xff1f; 清单1提供了与Scheme Shell&＃xff08;SCSH&＃xff09;的解释后的会话&＃xff0c;它是Lisp的派生词。 第一行定义了一个名为square的函数&＃xff0c;该函数接受一个参数并发出其平方根。 下一行说明了map函数的用法。 如图所示&＃xff0c;使用map &＃xff0c;您可以提供函数以及该函数所应用到的元素列表。 结果是一个包含平方元素的新列表。

清单1. SCSH中的map函数的演示

> (define square (lambda (x) (* x x)))
> (map square &＃39;(1 3 5 7))
&＃39;(1 9 25 49)
>

减少也适用于列表&＃xff0c;但通常将列表减少为标量值。 清单2中提供的示例说明了另一个SCSH函数&＃xff0c;用于将列表简化为标量-在这种情况下&＃xff0c;将值列表以&＃xff08;1 &＃43;&＃xff08;2 &＃43;&＃xff08;3 &＃43;&＃xff08;4 &＃43;&＃xff08;5&＃xff09;&＃xff09;&＃xff09;&＃xff09;&＃xff09;的形式求和。 请注意&＃xff0c;这是经典的函数式编程&＃xff0c;它依赖于迭代的递归。

清单2.减少SCSH的演示

> (define (list-sum lis) (if (null? lis) 0 (&＃43; (car lis) (list-sum (cdr lis)))))
> (list-sum &＃39;(1 2 3 4 5))
15
>

有趣的是&＃xff0c;在命令式语言中&＃xff0c;递归与迭代一样有效&＃xff0c;因为递归在后台被翻译成迭代。

Hadoop的编程模型

Google引入了MapReduce的想法&＃xff0c;将其作为用于处理或生成大量数据的编程模型。 在规范模型中&＃xff0c; map函数处理键值对&＃xff0c;从而产生一组中间的键值对。 然后&＃xff0c; reduce函数处理这些中间键/值对&＃xff0c;合并关联键的值&＃xff08;请参见图1&＃xff09;。 输入数据以这样一种方式进行分区&＃xff0c;即可以将其分布在一组机器中以进行并行处理。 以相同的方式&＃xff0c;并行处理生成的中间数据&＃xff0c;使该方法非常适合处理大量数据。

图1. MapReduce处理的简化视图

为了快速上手&＃xff0c;请从map的角度看一下图1的体系结构&＃xff0c;并减少字数&＃xff08;因为在本文中您将开发map并减少应用程序&＃xff09;。 提供输入数据后&＃xff08;进入Hadoop文件系统[HDFS]&＃xff09;&＃xff0c;首先将其分区&＃xff0c;然后&＃xff08;通过作业跟踪器&＃xff09;分配给地图工作人员。 尽管图2中的示例显示了一个简短的句子被分区的情况&＃xff0c;但是通常由于以下原因&＃xff0c;要分区的工作量在128MB大小范围内&＃xff1a;花费很少的时间进行设置&＃xff0c;因此有更多的工作要做以最小化这开销。 地图工作人员&＃xff08;在规范示例中&＃xff09;将工作分解为单个向量&＃xff0c;这些向量包含标记词和初始值&＃xff08;在本例中为1&＃xff09;。 映射任务完成时&＃xff08;由任务跟踪程序在Hadoop中定义&＃xff09;&＃xff0c;将工作提供给reduce worker。 reduce worker将密钥还原为唯一的集合&＃xff0c;该值表示找到的密钥数量。

图2.简单的MapReduce示例

请注意&＃xff0c;此过程可以在同一台计算机上或不同计算机上进行&＃xff0c;或者使用不同的数据分区依次或并行进行&＃xff0c;结果仍然相同。

尽管规范视图&＃xff08;用于使用字数统计生成搜索索引&＃xff09;是查看Hadoop的一种方法&＃xff0c;但事实证明&＃xff0c;这种计算模型可以通用地应用于许多计算问题&＃xff0c;您将看到。

Hadoop的灵活性

从图2所示的简单示例中&＃xff0c;请注意&＃xff0c;两个主要元素是map和reduce流程。 尽管对于这些过程的工作方式有传统的看法&＃xff0c;但map和reduce以这种方式运行并不需要体系结构。 这是Hadoop的真正功能-灵活地实施map并reduce以解决特定应用程序方式运行的流程。 字数示例非常有用&＃xff0c;并且适用于许多问题&＃xff0c;但是其他模型仍然适用于此通用框架。 所需要做的就是开发地图并减少应用程序&＃xff0c;使流程对Hadoop可见。

在其他应用程序中&＃xff0c;Hadoop甚至被用于实现诸如神经网络&＃xff0c;支持向量机和k- means聚类之类的算法的机器学习应用程序&＃xff08; 有关更多信息&＃xff0c;请参见参考资料一节&＃xff09;。

数据流

尽管Hadoop是基于Java的框架&＃xff0c;但可以使用Java语言以外的其他语言编写地图并简化应用程序。 流使这成为可能。 Hadoop中的streaming实用程序实现了一种数据流胶水。 使用streaming实用程序&＃xff0c;您可以定义自己的映射并减少可执行文件&＃xff08;每个文件都从标准输入[stdin]中获取输入&＃xff0c;并通过标准输出[stdout]提供输出&＃xff09;&＃xff0c;并且streaming实用程序适当地读写数据&＃xff0c;从而以需要&＃xff08;请参见清单3&＃xff09;。

清单3.使用Hadoop流实用程序

hadoop jar $HADOOP_HOME/hadoop-streaming.jar \-input inputData-output outputData-mapper map_exec-reducer reduce_exec

清单3展示了如何在Hadoop中使用streaming实用程序&＃xff0c;而图3以图形方式显示了如何定义流。 请注意&＃xff0c;这是流使用的一个简单示例。 有许多选项可用于调整数据的解析方式&＃xff0c;调整图像的调用方式&＃xff0c;为分区器或合并器指定替换图像以及其他配置调整&＃xff08; 有关更多信息&＃xff0c;请参阅“ 参考资料”部分&＃xff09;。

图3.图形流示例

Ruby示例

掌握了streaming实用程序的基本知识之后&＃xff0c;您就可以编写一个简单的Ruby映射并简化应用程序&＃xff0c;并了解如何在Hadoop框架内使用这些流程。 此处的示例与规范的MapReduce应用程序一起使用&＃xff0c;但是稍后您将看到其他应用程序&＃xff08;以及如何在map和reduce形式中实现它们&＃xff09;。

从映射器开始。 该脚本从stdin获取文本输入&＃xff0c;对其进行标记化&＃xff0c;然后向stdout发出一组键/值对。 像大多数面向对象的脚本语言一样&＃xff0c;此任务几乎太简单了。 清单4中显示了mapper脚本&＃xff08;带有一些注释和空格&＃xff0c;使它更大一些&＃xff09;。 该程序使用迭代器从stdin中读取一行&＃xff0c;并使用另一个迭代器将该行拆分为单独的标记。 然后将每个标记&＃xff08;单词&＃xff09;以1的关联值&＃xff08;由制表符分隔&＃xff09;发送到stdout。

清单4. Ruby映射脚本&＃xff08;map.rb&＃xff09;

#!/usr/bin/env ruby# Our input comes from STDIN
STDIN.each_line do |line|# Iterate over the line, splitting the words from the line and emitting# as the word with a count of 1.line.split.each do |word|puts "#{word}\t1"endend

接下来&＃xff0c;查看reduce应用程序。 这个稍微复杂一点&＃xff0c;但是使用Ruby哈希&＃xff08;关联数组&＃xff09;来简化约简操作&＃xff08;参见清单5&＃xff09;。 该脚本再次处理来自stdin的输入数据&＃xff08;由streaming实用程序传递&＃xff09;&＃xff0c;并将行拆分为一个单词和一个值。 然后检查哈希值中的单词&＃xff1b; 如果找到&＃xff0c;则将计数添加到元素。 否则&＃xff0c;您将在单词的哈希中创建一个新条目&＃xff0c;然后加载计数&＃xff08;在映射程序中应为1 &＃xff09;。 处理完所有输入后&＃xff0c;您只需遍历哈希并将键值对发送到stdout。

清单5. Ruby reduce脚本&＃xff08;reduce.rb&＃xff09;

#!/usr/bin/env ruby# Create an empty word hash
wordhash &＃61; {}# Our input comes from STDIN, operating on each line
STDIN.each_line do |line|# Each line will represent a word and countword, count &＃61; line.strip.split# If we have the word in the hash, add the count to it, otherwise# create a new one.if wordhash.has_key?(word)wordhash[word] &＃43;&＃61; count.to_ielsewordhash[word] &＃61; count.to_iendend# Iterate through and emit the word counters
wordhash.each {|record, count| puts "#{record}\t#{count}"}

完成map和reduce脚本后&＃xff0c;从命令行对其进行测试。 请记住使用chmod &＃43;x将这些文件更改为可执行文件。 首先生成一个输入文件&＃xff0c;如清单6所示。

清单6.生成输入文件

# echo "Hadoop is an implementation of the map reduce framework for " \"distributed processing of large data sets." > input
#

使用此输入&＃xff0c;您现在可以测试您的mapper脚本&＃xff0c;如清单7所示。回想一下&＃xff0c;该脚本只是将输入标记化为键值对&＃xff0c;其中每个值将为1 &＃xff08;非唯一输入&＃xff09;。

清单7.测试映射器脚本

# cat input | ruby map.rb
Hadoop 1
is 1
an 1
implementation 1
of 1
the 1
map 1
reduce 1
framework 1
for 1
distributed 1
processing 1
of 1
large 1
data 1
sets. 1
#

到目前为止&＃xff0c;一切都很好。 现在&＃xff0c;以原始流形式&＃xff08;Linux®管道&＃xff09;将整个应用程序组合在一起。 在清单8中&＃xff0c;您将输入传递给地图脚本&＃xff0c;对输出进行排序&＃xff08;可选步骤&＃xff09;&＃xff0c;然后将所得的中间数据传递给reducer脚本。

清单8.使用Linux管道的简单MapReduce

# cat input | ruby map.rb | sort | ruby reduce.rb
large 1
of 2
framework 1
distributed 1
data 1
an 1
the 1
reduce 1
map 1
sets. 1
Hadoop 1
implementation 1
for 1
processing 1
is 1
#

Ruby与Hadoop

在外壳环境中按预期方式运行地图和归约脚本后&＃xff0c;将其与Hadoop一起进行测试。 我将跳过Hadoop设置任务&＃xff08;请参阅本系列的第1 部分或第2部分 &＃xff0c;以启动并运行Hadoop&＃xff09;。

第一步是在HDFS中为您的输入数据创建一个输入目录&＃xff0c;然后提供一个示例文件&＃xff0c;您将在该文件上测试脚本。 清单9演示了此步骤&＃xff08;有关这些步骤的更多信息&＃xff0c;请参见第1部分或第2部分 &＃xff09;。

清单9.为MapReduce流程创建输入数据

# hadoop fs -mkdir input
# hadoop dfs -put /usr/src/linux-source-2.6.27/Documentation/memory-barriers.txt input
# hadoop fs -ls input
Found 1 items
-rw-r--r-- 1 root supergroup 78031 2010-06-04 17:36 /user/root/input/memory-barriers.txt
#

接下来&＃xff0c;使用streaming实用程序&＃xff0c;使用自定义脚本调用Hadoop&＃xff0c;指定输入数据和输出位置&＃xff08;请参见清单10&＃xff09;。 请注意&＃xff0c;在此示例中&＃xff0c; -file选项只是告诉Hadoop将Ruby脚本打包为作业提交的一部分。

清单10.将Hadoop流与自定义Ruby MapReduce脚本一起使用

# hadoop jar /usr/lib/hadoop-0.20/contrib/streaming/hadoop-0.20.2&＃43;228-streaming.jar \-file /home/mtj/ruby/map.rb -mapper /home/mtj/ruby/map.rb \-file /home/mtj/ruby/reduce.rb -reducer /home/mtj/ruby/reduce.rb \-input input/* -output output
packageJobJar: [/home/mtj/ruby/map.rb, /home/mtj/ruby/reduce.rb, /var/lib/hadoop-0.20/...
10/06/04 17:42:38 INFO mapred.FileInputFormat: Total input paths to process : 1
10/06/04 17:42:39 INFO streaming.StreamJob: getLocalDirs(): [/var/lib/hadoop-0.20/...
10/06/04 17:42:39 INFO streaming.StreamJob: Running job: job_201006041053_0001
10/06/04 17:42:39 INFO streaming.StreamJob: To kill this job, run:
10/06/04 17:42:39 INFO streaming.StreamJob: /usr/lib/hadoop-0.20/bin/hadoop job ...
10/06/04 17:42:39 INFO streaming.StreamJob: Tracking URL: http://localhost:50030/...
10/06/04 17:42:40 INFO streaming.StreamJob: map 0% reduce 0%
10/06/04 17:43:17 INFO streaming.StreamJob: map 100% reduce 0%
10/06/04 17:43:26 INFO streaming.StreamJob: map 100% reduce 100%
10/06/04 17:43:29 INFO streaming.StreamJob: Job complete: job_201006041053_0001
10/06/04 17:43:29 INFO streaming.StreamJob: Output: output
#

最后&＃xff0c;通过hadoop实用程序使用cat文件系统操作探索输出&＃xff08;请参见清单11&＃xff09;。

清单11.探索Hadoop输出

# hadoop fs -ls /user/root/output
Found 2 items
drwxr-xr-x - root supergroup 0 2010-06-04 17:42 /user/root/output/_logs
-rw-r--r-- 1 root supergroup 23014 2010-06-04 17:43 /user/root/output/part-00000
# hadoop fs -cat /user/root/output/part-00000 | head -12
&＃43;--->| 4
immediate 2
Alpha) 1
enable 1
_mandatory_ 1
Systems 1
DMA. 2
AMD64 1
{*C,*D}, 2
certainly 2
back 2
this 23
#

因此&＃xff0c;在不到30行的脚本中&＃xff0c;您已经实现了map和reduce元素&＃xff0c;并演示了它们在Hadoop框架内的执行。 这是一个简单的示例&＃xff0c;但它说明了Hadoop背后的强大功能&＃xff0c;以及为什么它成为使用自定义或专有算法处理大型数据集的如此流行的框架。

Hadoop的其他应用程序

Hadoop可以用于许多应用程序中&＃xff0c;而不仅仅是简单地计算大型数据集的字数。 所需要做的就是以Hadoop基础架构可以使用的矢量形式表示数据。 尽管规范示例使用向量表示作为键和值&＃xff0c;但是如何定义值&＃xff08;例如&＃xff0c;多个值的集合&＃xff09;没有任何限制。 这种灵活性可以为Hadoop提供更丰富的应用程序的新机会。

正好适合MapReduce字数模型的一个有趣的应用程序是制表Web服务器访问的频率&＃xff08;在开创性的Google论文中讨论&＃xff09;。 对于此应用程序&＃xff0c;URL用作键&＃xff08;从Web服务器访问日志中提取&＃xff09;。 reduce过程的结果是基于Web服务器日志的给定Web站点每个URL的访问总数。

在机器学习应用程序中&＃xff0c;Hadoop已被用作扩展遗传算法以处理大量GA个体的方法&＃xff08;潜在解决方案&＃xff09;。 map过程执行传统的遗传算法&＃xff0c;从本地池中寻求最佳个体解决方案。 然后&＃xff0c;reduce应用程序将成为地图阶段中各个解决方案的竞赛。 这允许各个节点确定其最佳解决方案&＃xff0c;然后允许这些解决方案在分布式的优胜劣汰展示中在还原阶段竞争。

创建了另一个有趣的应用程序来识别垃圾邮件的僵尸网络。 此过程的第一步是对电子邮件进行分类&＃xff0c;以减少它们&＃xff08;基于一组指纹&＃xff09;为来自给定组织。 根据此过滤后的数据&＃xff0c;为以某种方式连接的电子邮件&＃xff08;例如&＃xff0c;引用电子邮件正文中的相同链接&＃xff09;构建了一个图形。 然后&＃xff0c;将这些相关的电子邮件减少到主机&＃xff08;静态或动态IP地址&＃xff09;以识别相关的僵尸网络。

在通过地图和简化图元查看世界的应用程序之外&＃xff0c;Hadoop可用作在计算机集群之间分配工作的一种方式。 映射和简化不一定强制特定类型的应用程序。 取而代之的是&＃xff0c;可以将Hadoop视为将数据和算法分配到主机以进行更快的并行处理的一种方式。

Hadoop应用程序生态系统

尽管Hadoop提供了一个灵活的框架&＃xff0c;但其他应用程序也可以将其接口转换为其他应用程序。 一个有趣的示例称为Hive&＃xff0c;它是一种具有自己的查询语言&＃xff08;称为Hive QL &＃xff09;的数据仓库基础结构。 Hive使Hadoop对具有结构化查询语言&＃xff08;SQL&＃xff09;背景的人更加熟悉&＃xff0c;但是它也支持传统的MapReduce基础结构来进行数据处理。

HBase是驻留在HDFS之上的另一个有趣的应用程序。 这是一个类似于Google BigTable的高性能数据库系统。 HBase代替了传统的文件处理&＃xff0c;使数据库表成为MapReduce处理的输入和输出形式。

最后&＃xff0c; Pig是Hadoop上用于分析大型数据集的平台。 Pig提供了一种高级语言&＃xff0c;可以进行编译以映射和减少应用程序。

更进一步

Hadoop 系列的最后一篇文章探讨了针对Hadoop框架的地图开发和减少Ruby中的应用程序。 希望从本文中&＃xff0c;您可以看到Hadoop的真正功能。 尽管Hadoop限制您使用特定的编程模型&＃xff0c;但是该模型非常灵活&＃xff0c;可以应用于大量应用程序。

翻译自: https://www.ibm.com/developerworks/java/library/l-hadoop-3/index.html