Python 海量数据处理之 _Hadoop（三）程序调用

1. 说明

  前两篇分别介绍了 Hadoop 的配置方法和基本原理，本篇将介绍如何让程序借助 Hadoop 调用服务器集群中的算力。Hadoop 是 Java 语言实现的，它不仅支持 Java，还支持 C++，Python 等程序的分布计算。下面以 Python 为例介绍如何使用 Hadoop 的 MapReduce 功能。

2. 工作过程

  在原理部分介绍过，Hadoop 分为两部分，一部分是存储，一部分是运算，而各个部分又可分为主控和局部实现。这里忽略细节，主要说明运算的主控部分是如何运作的。

  从程序调用的角度看，程序员首先需要把待处理的文件复制到 HDFS 文件系统之中，然后调 Hadoop 提供的 java 程序执行分布计算，具体需要执行的 python 代码用参数的形式提供；最后在 HDFS 生成输出文件，程序员再将其拷回本地即可。

  这里的分布计算主要指 MapReduce，MapReduce 框架将输入数据分割成数据块，传给 Mapper，然后 Map 任务在各个服务器上以完全并行的方式处理，接着 MapReduce 框架对 Map 任务的输出进行排序，并将结果做为 Reduce 任务的输入，最后由 Reduce 任务输出最终的结果。请注意，在 Map 和 Reduce 中有个排序的过程，因为必须完成所有 map 后才能 reduce，这也局限了 MapReduce 的工作场景。

 MapReduce 的核心数据是<key,value>键值对，Mapper 处理完数据输出的是键值对（如果不输出，则起到了过滤的作用），框架对键值对排序（后面在具体例子中介绍排序的原因），再输出给 Reducer 继续处理。Mapper 的数量由输入文件的大小确定，Reducer 的数量由程序员指定.

3. HDFS 文件系统操作

 Hadoop 集群中的服务器处理的是 HDFS 中的数据，因此需要在本地和 HDFS 之间复制文件，常用命令如下：

1
2
3
4
5
$ hadoop fs -mkdir /tmp/input # 建立目录
$ hadoop fs -ls /tmp/ # 查看目录中文件
$ hadoop fs -copyFromLocal /tmp/test_items.txt /tmp/input/ # 复制本地文件到HDFS
$ hadoop fs -cat /tmp/input/test_items.txt # 查看文件内容
$ hadoop fs -copyToLocal /tmp/input . # 将HDFS中文件夹复制到本地

  更多命令见：http://hadoop.apache.org/docs/r1.0.4/file_system_shell.html

4. 例程

1) 实现功能

  统计文件中单词出现的次数。

2) 程序 mapper.py

  注意将程序设置为可执行权限

1
2
3
4
5
6
7
8
9
10
11
#!/usr/bin/env python
# -*- coding: utf-8 -*-

import sys

for line in sys.stdin:  # 从标准输入中按行读数据
    
    line = line.strip()  # 将行尾行首的空格去除
    words = line.split()  #按空格将句子分割成单个单词
    for word in words:
        print '%s\t%s' %(word, 1) # 输出键值对，每单词出现1次

3) 程序 reducer.py

  注意将程序设置为可执行权限

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
#!/usr/bin/env python
# -*- coding: utf-8 -*-

import sys

current_word = None  # 当前单词
current_count = 0  # 当前单词频数
word = None

for line in sys.stdin:
    words = line.strip() 
    word, count = words.split('\t')  # 按照制表符分隔单词和数量
    
    try:
        count = int(count)  # 将字符串类型的‘1’转换为整型1
    except ValueError:
        continue

    if current_word == word:  # 如果当前的单词等于读入的单词
        current_count += count  # 单词频数加1
    else:
        if current_word:  # 如果当前的单词不为空则打印其单词和频数
            print '%s\t%s' %(current_word, current_count)  
        current_count = count  # 否则将读入的单词赋值给当前单词，且更新频数
        current_word = word

if current_word == word:
    print '%s\t%s' %(current_word, current_count)

4) 在本地实验

  程序在 Hadoop 上运行前，先在本地运行，以保证其正确性，运行命令：

1
$ echo "Dear Bear River Car Car River Dear Car Bear" | ./mapper.py | sort -k 1,1 | ./reducer.py

  可以看到命令用管道的方式将 map,reduce 和数据连接到了一起，中间还有 sort 命令用于排序，排序原因从 reducer.py 程序中可以看到。也可参见下图：

 Hadoop 将三行数据分成三份，mapper.py 处理后如第三列所示，排序后变为第四列，它把同样的单词都放在一起了，使得在 reducer.py 在处理时只要判断连续单词是否相同，而无需从头到尾搜索单词。

5) Hadoop 实验

1
$ hadoop jar /home/hadoop/hadoop/share/hadoop/tools/lib/hadoop-streaming-2.7.5.jar  -files ./mapper.py,./reducer.py -mapper ./mapper.py -reducer ./reducer.py -input /tmp/input/*.txt -output /tmp/output/

  这里 py 文件被指定了两次，files 指定的是上传两个 py 文件，这里还可以上传配置文件等，后面是分别指定作为 mapper 和 reducer 的程序，然后是指定输入输出。非常简单，甚至不需要在 python 文件中加任何的支持库，只要读写输入输出却可。

5. 应用场景

 Hadoop 主要是针对海量数据处理的，试想当数据以 TB,PB 计量的时候，我们不可能用单机一次性打开所有数据。Hadoop 方式可用多台便宜 PC 组合的方式处理海量数据。

  看了一些典型的 Hadoop 应用场景，觉得现在使用 Hadoop 主要以 HDFS 加数据库的共享数据为主，更多的时候是被其它上层工具封装后调用。