用 Hadoop 进行分布式并行编程

Sky-Tiger

更进一步
本文讨论了一个简单的（伪分布式）Hadoop 集群的安装和初始配置。在这里，我选用 Cloudera 的 Hadoop 发行版是因为它简化了 Hadoop 的安装和初始配置。可以在 apache.org 找到许多 Hadoop 发行版（包括源代码）。更多信息见 参考资料。
但是，如果缺少硬件资源，无法扩展 Hadoop 集群以满足自己的需要，那么应该怎么办？由于 Hadoop 非常流行，可以很方便地在云计算基础设施中使用预构建的 Hadoop VM 和租用的服务器运行它。Amazon 在 Amazon Elastic Compute Cloud (Amazon EC2) 中提供 Amazon Machine Image (AMI) 和计算资源。另外，Microsoft 最近宣布将在它的 Windows® Azure Services Platform 中支持 Hadoop。
通过本文很容易看出 Hadoop 显著简化了处理大型数据集的分布式计算。本系列中的下一篇文章通过更多示例讨论如何在多节点集群中配置 Hadoop。

Sky-Tiger

更进一步
本文讨论了一个简单的（伪分布式）Hadoop 集群的安装和初始配置。在这里，我选用 Cloudera 的 Hadoop 发行版是因为它简化了 Hadoop 的安装和初始配置。可以在 apache找到许多 Hadoop 发行版（包括源代码）。更多信息见 参考资料。
但是，如果缺少硬件资源，无法扩展 Hadoop 集群以满足自己的需要，那么应该怎么办？由于 Hadoop 非常流行，可以很方便地在云计算基础设施中使用预构建的 Hadoop VM 和租用的服务器运行它。Amazon 在 Amazon Elastic Compute Cloud (Amazon EC2) 中提供 Amazon Machine Image (AMI) 和计算资源。另外，Microsoft 最近宣布将在它的 Windows® Azure Services Platform 中支持 Hadoop。
通过本文很容易看出 Hadoop 显著简化了处理大型数据集的分布式计算。本系列中的下一篇文章通过更多示例讨论如何在多节点集群中配置 Hadoop。

Sky-Tiger

更进一步
本文讨论了一个简单的（伪分布式）Hadoop 集群的安装和初始配置。在这里，我选用 Cloudera 的 Hadoop 发行版是因为它简化了 Hadoop 的安装和初始配置。可以在 apache.org 找到许多 Hadoop 发行版（包括源代码）。更多信息见 参考资料。
但是，如果缺少硬件资源，无法扩展 Hadoop 集群以满足自己的需要，那么应该怎么办？由于 Hadoop 非常流行，可以很方便地在云计算基础设施中使用预构建的 Hadoop VM 和租用的服务器运行它。Amazon 在 Amazon Elastic Compute Cloud (Amazon EC2) 中提供 Amazon Machine Image (AMI) 和计算资源。另外，Microsoft 最近宣布将在它的 Windows Azure Services Platform 中支持 Hadoop。
通过本文很容易看出 Hadoop 显著简化了处理大型数据集的分布式计算。本系列中的下一篇文章通过更多示例讨论如何在多节点集群中配置 Hadoop。

Sky-Tiger

map 和 reduce 的起源
是什么功能性元素激发了 MapReduce 编程范例的创立？在 1958 年，John McCarthy 发明了名为 Lisp 的语言，其实现了数值和符号计算，但在递归形式下此语言非常不同于现在所使用的大多数语言。（在维基百科全书上记述着 Lisp 那段迷人的历史，同时包括一个有用的教程 — 值得您花费时间来阅读。）Lisp 最先是在 IBM® 704 中实现的，IBM® 704 是第一种大规模生产的计算机，也支持其他旧的语言，如 FORTRAN。
map 函数，源于功能性语言（如 Lisp）但如今在其他语言中也很常见，其中包含了一系列元素的函数的应用程序。这意味着什么？ 清单 1 通过 Scheme Shell (SCSH) 提供解释会话，即一个 Lisp 衍生。第一行定义一个名为 square 的函数，该函数可接受参数并发出其平方根。下一行说明 map 函数的使用。如图所示，通过 map，为已应用的函数提供您的函数和一系列元素。结果是一个包含平方元素的新列表。
清单 1. SCSH 上的 map 函数演示
> (define square (lambda (x) (* x x)))
> (map square '(1 3 5 7))
'(1 9 25 49)
>
Reduce 也适用于列表但是通常将列表缩减为标量值。清单 2中提供的示例说明用于将列表缩减为标量的其他 SCSH 函数 — 在这种情况下，用 (1 + (2 + (3 + (4 + (5))))) 的格式汇总值的列表。请注意这是典型的功能性编程，取决于迭代上的递归。
清单 2. SCSH 上的 reduce 演示
> (define (list-sum lis) (if (null? lis) 0 (+ (car lis) (list-sum (cdr lis)))))
> (list-sum '(1 2 3 4 5))
15
>
有趣的是要注意递归与迭代在命令性语言中同样高效，因为递归在幕后被转化成迭代。

Sky-Tiger

Hadoop 的编程模型

[size=1.166em]Google 引用 MapReduce 的概念作为处理或生成大型数据集的编程模型。在规范模型中，map 函数处理键值对，这将得出键值对的中间集。然后 reduce 函数会处理这些中间键值对，并合并相关键的值（请参考图 1）。输入数据使用这样一种方法进行分区，即在并行处理的计算机集群中分区的方法。使用相同的方法，已生成的中间数据将被并行处理，这是处理大量数据的理想方法。

图 1. MapReduce 处理的简化视图

[size=1.166em]对于快速刷新器来说，查看图 1 的体系结构，从 map 和 reduce 角度来进行字数统计（因为您将在本文中开发 map 和 reduce 应用程序)。在提供输入数据时（进入 Hadoop 文件系统 [HDFS]），首先分段，然后分配给 map 工作线程（通过作业跟踪器）。虽然 图 2 中的示例显示了一个被分段的简短语句，但是分段的工作数量通常在 128MB 范围内，其原因是建立工作只需要很少的时间，因为有更多的工作要做，以便最大限度地减少这种开销。map 工作线程（在规范的示例中）将工作分割成包含已标记单词和初始值（在此情况下是 1）的单个矢量。在 map 任务完成时（如通过任务跟踪器在 Hadoop 中所定义的），提供工作给 reduce 工作线程。通过代表所发现的键的数量的值，reduce 工作线程将许多键缩减为一个惟一的集合。

图 2. 简单的 MapReduce 示例

[size=1.166em]请注意此过程可在相同的或不同的计算机中出现或者使用不同的数据分区来按顺序或并行完成，且结果仍然是相同的。

[size=1.166em]虽然规范的视图（用于使用字数统计生成搜索索引）是一种用来查看 Hadoop 方法，但结果是此计算模型被常规地应用到可计算问题上，正如您将要看到的那样。

Sky-Tiger

map 和 reduce 的起源
是什么功能性元素激发了 MapReduce 编程范例的创立？在 1958 年，John McCarthy 发明了名为 Lisp 的语言，其实现了数值和符号计算，但在递归形式下此语言非常不同于现在所使用的大多数语言。（在维基百科全书上记述着 Lisp 那段迷人的历史，同时包括一个有用的教程 — 值得您花费时间来阅读。）Lisp 最先是在 IBM® 704 中实现的，IBM704 是第一种大规模生产的计算机，也支持其他旧的语言，如 FORTRAN。
map 函数，源于功能性语言（如 Lisp）但如今在其他语言中也很常见，其中包含了一系列元素的函数的应用程序。这意味着什么？ 清单 1 通过 Scheme Shell (SCSH) 提供解释会话，即一个 Lisp 衍生。第一行定义一个名为 square 的函数，该函数可接受参数并发出其平方根。下一行说明 map 函数的使用。如图所示，通过 map，为已应用的函数提供您的函数和一系列元素。结果是一个包含平方元素的新列表。
清单 1. SCSH 上的 map 函数演示
> (define square (lambda (x) (* x x)))
> (map square '(1 3 5 7))
'(1 9 25 49)
>
Reduce 也适用于列表但是通常将列表缩减为标量值。清单 2中提供的示例说明用于将列表缩减为标量的其他 SCSH 函数 — 在这种情况下，用 (1 + (2 + (3 + (4 + (5))))) 的格式汇总值的列表。请注意这是典型的功能性编程，取决于迭代上的递归。
清单 2. SCSH 上的 reduce 演示
> (define (list-sum lis) (if (null? lis) 0 (+ (car lis) (list-sum (cdr lis)))))
> (list-sum '(1 2 3 4 5))
15
>
有趣的是要注意递归与迭代在命令性语言中同样高效，因为递归在幕后被转化成迭代。

Sky-Tiger

map 和 reduce 的起源
是什么功能性元素激发了 MapReduce 编程范例的创立？在 1958 年，John McCarthy 发明了名为 Lisp 的语言，其实现了数值和符号计算，但在递归形式下此语言非常不同于现在所使用的大多数语言。（在维基百科全书上记述着 Lisp 那段迷人的历史，同时包括一个有用的教程 — 值得您花费时间来阅读。）Lisp 最先是在 IBM 704 中实现的，IBM 704 是第一种大规模生产的计算机，也支持其他旧的语言，如 FORTRAN。
map 函数，源于功能性语言（如 Lisp）但如今在其他语言中也很常见，其中包含了一系列元素的函数的应用程序。这意味着什么？ 清单 1 通过 Scheme Shell (SCSH) 提供解释会话，即一个 Lisp 衍生。第一行定义一个名为 square 的函数，该函数可接受参数并发出其平方根。下一行说明 map 函数的使用。如图所示，通过 map，为已应用的函数提供您的函数和一系列元素。结果是一个包含平方元素的新列表。
清单 1. SCSH 上的 map 函数演示
> (define square (lambda (x) (* x x)))
> (map square '(1 3 5 7))
'(1 9 25 49)
>
Reduce 也适用于列表但是通常将列表缩减为标量值。清单 2中提供的示例说明用于将列表缩减为标量的其他 SCSH 函数 — 在这种情况下，用 (1 + (2 + (3 + (4 + (5))))) 的格式汇总值的列表。请注意这是典型的功能性编程，取决于迭代上的递归。
清单 2. SCSH 上的 reduce 演示
> (define (list-sum lis) (if (null? lis) 0 (+ (car lis) (list-sum (cdr lis)))))
> (list-sum '(1 2 3 4 5))
15
>
有趣的是要注意递归与迭代在命令性语言中同样高效，因为递归在幕后被转化成迭代。

Sky-Tiger

Hadoop 的灵活性
从 图 2 中所示的简单示例看，需注意 map 和 reduce 过程这两个主要元素。虽然这里存在一个这些过程如何工作的传统视图，但是它不是 map 和 reduce 体系结构所需要的。这就是 Hadoop 的真实力量 — 其灵活性用来实现在某种程度上活动的 map 和 reduce 过程，这解决了一个特定的应用程序。虽然字数统计示例对于大量的问题是有用且适用的，但是其他的模型仍然在此总体框架内适用。所需的就是使 map 和 reduce 应用程序的开发过程对于 Hadoop 可见。
在其他的应用程序中，Hadoop 已经被用于实现包括神经网络算法的计算机学习应用程序，支持矢量计算机以及 k-means 集群（要获得更多信息，请参考 参考资料 部分）。

Sky-Tiger

数据流

[size=1.166em]虽然 Hadoop 是一个基于 Java 的框架，但是其有可能在 Java 语言以外的语言中编写 msp 和 reduce 应用程序。Hadoop 内的 流 实用工具实现了一种数据流胶的类型。通过 流 实用工具，您可以定义您自己的可执行 map 和 reduce（使用每一个从标准输入 [stdin] 提取的输入和每一个通过标准输出 [stdout] 提供的输出），且 流 实用工具可适当地读取和写入数据，根据需要调用您的应用程序（请参考清单 3）。

清单 3. 使用 Hadoop 流实用工具hadoop jar $HADOOP_HOME/hadoop-流.jar \        -input inputData        -output outputData        -mapper map_exec        -reducer reduce_exec

[size=1.166em]清单 3 说明如何在 Hadoop 内使用 流 实用工具，图 3 图形化地显示了如何定义流。请注意这是一个流使用的简单示例。大量的选项可用于制定如何解析数据、制定如何调用图像、为分区器和合成器指定替换图像以及调整其他配置（要获得更多信息，请参考 参考资料 部分）。

图 3. 图形流示例

Sky-Tiger

Ruby 示例
通过已经获得的在 流 实用工具基本理解上的经验，您已经准备编写一个简单的 Ruby map 和 reduce 应用程序并查看如何在 Hadoop 框架中使用过程。虽然此处的示例伴随着规范的 MapReduce 应用程序，但是稍后您将看到其他的应用程序（取决于您将如何用 map 和 reduce 格式实现它们）。
首选是 mapper。此脚本从 stdin 提取文本输入，首先标记它，然后将一系列键值对发送到 stdout。像大多数面向对象的脚本语言一样，这个任务几乎太简单了。如清单 4 中所示的 mapper 脚本（通过一些注释和空白区域可给与其大一点的大小）。此程序使用一个迭代器来从 stdin 中读取一行，同时另一个迭代器将该行分割成单个的标记。使用为 1 的相关值（通过选项卡分隔）将每一个标记（单词）发送到 stdout。
清单 4. Ruby map 脚本（map.rb）
#!/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"
  end

end
下一步，查看 reduce 应用程序。虽然此应用程序稍微有些复杂，但是使用 Ruby hash（关联阵列）可简化 reduce 操作（请参考清单 5）。此脚本可通过来自 stdin （通过 流 实用工具传递）的输入数据再次工作且将该行分割成一个单词或值。而后该 hash 会检查该单词；如果发现，则将计数添加到元素。否则，您需要在该单词的 hash 中创建新的条目，然后加载计数（应该是来自 mapper 过程的 1）。在所有输入都被处理以后，通过 hash 可简单迭代且将键值对发送到 stdout。
清单 5. Ruby reduce 脚本（reduce.rb）
#!/usr/bin/env ruby

# Create an empty word hash
wordhash = {}

# Our input comes from STDIN, operating on each line
STDIN.each_line do |line|

  # Each line will represent a word and count
  word, count = 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] += count.to_i
  else
    wordhash[word] = count.to_i
  end

end

# Iterate through and emit the word counters
wordhash.each {|record, count| puts "#{record}\t#{count}"}
随着 map 和 reduce 脚本的完成，需从命令行测试它们。记得要使用 chmod +x 将这些文件更改为可执行。通过生成输入文件来启动，如清单 6 所示。
清单 6. 生成输入文件
# echo "Hadoop is an implementation of the map reduce framework for " \
        "distributed processing of large data sets." > input
#
通过单词输入，现在您可以测试您的 mapper 脚本，如清单 7 所示。回想此脚本简单地将输入标记到键值对，此处每个值都将是 1（非惟一输入）。
清单 7. 测试 mapper 脚本
# 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
#
到目前为止，一切都很顺利。现在，在原始流格式中将整个应用程序一起调出。在清单 8 中，通过 map 脚本传递您的输入、排序输出（可选步骤）、然后通过 reduce 脚本传递由此产生的中间数据。
清单 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
#