用python + hadoop streaming 分布式编程（一） — 原理介绍，样例程序与本地调试

下面我将通过标题、代码块等格式来详细讲解“用python + hadoop streaming 分布式编程（一） -- 原理介绍，样例程序与本地调试”的完整攻略：

简介

本文将介绍如何使用Python语言和Hadoop Streaming框架进行分布式编程，并且提供样例程序和本地调试方法。Hadoop Streaming框架是Hadoop自带的工具之一，允许开发者在不使用Java语言的情况下，使用各种支持标准输入输出的脚本语言进行MapReduce编程。因此，Hadoop Streaming + Python的组合可以为没有Java编程经验的开发者提供一个良好的解决方案。

预备知识

• Hadoop的基本概念和架构
• Python编程语言
• Hadoop Streaming基本的命令格式和参数含义

实现过程

为了使用Python + Hadoop Streaming进行分布式编程，需要进行以下步骤：

步骤1：编写map函数和reduce函数的Python代码

在使用Hadoop Streaming时，需要将map和reduce函数编写成Python脚本，并且将脚本保存在一个可执行文本文件中。以WordCount为例，map函数和reduce函数的代码如下：

# map函数
#!/usr/bin/env python
import sys
for line in sys.stdin:
    words = line.strip().split()
    for word in words:
        print word + '\t' + '1'

# reduce函数
#!/usr/bin/env python
import sys
current_word = None
current_count = 0
for line in sys.stdin:
    word, count = line.strip().split('\t')
    count = int(count)
    if current_word == word:
        current_count += count
    else:
        if current_word:
            print current_word + '\t' + str(current_count)
        current_word = word
        current_count = count
if current_word == word:
    print current_word + '\t' + str(current_count)

步骤2：使用Hadoop Streaming运行Python脚本

Hadoop Streaming的命令格式为：

hadoop jar $HADOOP_HOME/share/hadoop/tools/lib/hadoop-streaming-<version>.jar [options]

其中，$HADOOP_HOME为Hadoop的安装路径，<version>为Hadoop的版本号。

下面是一个使用Hadoop Streaming运行WordCount示例的命令：

hadoop jar $HADOOP_HOME/share/hadoop/tools/lib/hadoop-streaming-<version>.jar \
-input /path/to/input/file \
-output /path/to/output/directory \
-mapper /path/to/mapper/script \
-reducer /path/to/reducer/script \
-file /path/to/mapper/script \
-file /path/to/reducer/script

其中，/path/to/input/file为输入文件路径，/path/to/output/directory为输出目录路径，/path/to/mapper/script和/path/to/reducer/script为之前编写的map函数和reduce函数所在的Python脚本路径。在运行命令时，需要使用-file参数指定Python脚本的路径，这样Hadoop才能够将脚本发送到集群的计算机节点上执行。

步骤3：本地调试

为了避免在Hadoop集群上反复测试，可以使用本地模拟的方式进行调试。这里提供两种方法：

方法1：使用cat命令+管道

使用cat命令模拟输入文件，将其与之前编写的map函数和reduce函数的Python脚本进行结合：

cat /path/to/input/file | /path/to/mapper/script | /path/to/reducer/script

方法2：使用Python脚本

编写本地的Python脚本，利用Python的标准输入输出流来模拟Hadoop Streaming中的输入输出流。下面是一个简单的示例代码：

import sys

# mock input
input_data = [
    'hello world',
    'hello python',
    'hello hadoop',
    'world python',
    'world hadoop',
    'python hadoop'
]

# mock stdin
sys.stdin = input_data

# run mapper
from /path/to/mapper/script import mapper
for line in sys.stdin:
    mapper(line)

# run reducer
from /path/to/reducer/script import reducer
for key in keys:
    reducer(key, values)

需要注意的是，此方法在模拟Hadoop Streaming时，缺乏了输入文件的完整路径和输出目录的路径，因此需要在Python脚本中手动处理输入输出路径。

示例说明

示例1：WordCount

上面的例子就是一个针对WordCount的演示。使用Hadoop Streaming和Python编写了一个简单的map函数和reduce函数，根据不同的字符分隔，将各自的取值加一，然后计数器就++，最后将结果输出。

在本地终端给出如下的示例输入：

apple banana apple lemon orange banana orange
watermelon apple orange bananana
cherry apple banana
grape watermelon apple banana orange

在执行上述命令之后，Hadoop Streaming会在计算集群的节点上并行运行编写的Python脚本，最后将运算结果收集到指定的输出目录下。接下来给出最终的示例输出：

apple   4
banana  4
cherry  1
grape   1
lemon   1
orange  4
watermelon      2

示例2：统计访问量最大的IP地址

示例2展示如何使用Hadoop Streaming和Python编写一个程序来统计访问量最大的IP地址。我们已经提前下载好了access.log文件，这是一个web服务器日志文件，里面包含有所有访问网站的客户端IP地址。接下来，模拟一下本地和Hadoop集群上执行此程序的过程：

本地调试

以上代码中，我们将从access.log文件中读取IP地址，并且将每个IP地址作为“键”，将数量1作为“值”，传递给Hadoop Streaming来进行统计。为了方便本地测试，我们可以将access.log文件复制到本地，并在命令行中执行以下命令：

cat access.log | python count_ip.py | sort -n -k 2 -r | head -n 10

上述命令将输出10个出现次数最多的IP地址，并按相应的出现次数进行排序。值得注意的是，为了保证正确的排序方式，需要使用sort命令并指定正确的参数（-n以数字方式进行排序，-k 2指定按照第二个字段进行排序，-r指定倒序排列）。

集群运行

同样需要在Hadoop集群上进行map-reduce处理。

下面是在Linux系统上运行的完整命令：

hadoop jar /opt/hadoop-3.1.3/share/hadoop/tools/lib/hadoop-streaming-3.1.3.jar \
-file /path/to/count_ip_mapper.py -mapper /path/to/count_ip_mapper.py \
-file /path/to/count_ip_reducer.py -reducer /path/to/count_ip_reducer.py \
-input /path/to/access.log \
-output /path/to/output/directory \

我们使用hadoop jar命令执行了Hadoop Streaming的 Jar包，并通过-file、-mapper和-reducer等参数指定了Python脚本的路径和命令。同时，我们也使用了-input和-output参数来指定输入文件路径和输出目录路径。

命令执行结束之后，输出的目录下将会生成一个part-xxxxx文件，其中包含了按照IP出现次数从多到少排序的所有结果，用户可以根据需要进行进一步的处理。

总结

到此结束本次的教程。在本文中，我们提供了使用Python和Hadoop Streaming进行分布式编程的完整攻略，其中包括Hadoop Streaming的基本命令格式、本地调试方法和示例程序的说明。凭借这些知识，用户就可以轻松利用Python和Hadoop Streaming来开发各种大数据处理应用。