Hadoop Streaming 使用及参数设置

Posted 2020-10-05 hopelee

1. MapReduce 与 HDFS 简介

　　什么是 Hadoop ？

　　Google 为自己的业务需要提出了编程模型 MapReduce 和分布式文件系统 Google File System，并发布了相关论文（可在 Google Research 的网站上获得：GFS、MapReduce）。Doug Cutting 和 Mike Cafarella 在开发搜索引擎 Nutch 时对这两篇论文进行了自己的实现，即同名的 MapReduce 和 HDFS，合起来就是 Hadoop。

　　MapReduce 的 Data Flow 如下图所示，原始数据经过 mapper 处理，再进行 partition 和 sort，到达 reducer，输出最后结果。

2. Hadoop Streaming 原理

　　Hadoop 本身是用 Java 开发的，程序也需要用 Java 编写，但是通过 Hadoop Streaming，我们可以使用任意语言来编写程序，让 Hadoop 运行。

　　Hadoop Streaming 就是通过将其他语言编写的 mapper 和 reducer 通过参数传给一个事先写好的 Java 程序（Hadoop 自带的 *-streaming.jar），这个 Java 程序会负责创建 MR 作业，另开一个进程来运行 mapper，将得到的输入通过 stdin 传给它，再将 mapper 处理后输出到 stdout 的数据交给 Hadoop，经过 partition 和 sort 之后，再另开进程运行 reducer，同样通过 stdin/stdout 得到最终结果。因此，我们只需要在其他语言编写的程序中，通过 stdin 接收数据，再将处理过的数据输出到 stdout，Hadoop Streaming 就能通过这个 Java 的 wrapper 帮我们解决中间繁琐的步骤，运行分布式程序。

　　原理上只要是能够处理 stdio 的语言都能用来写 mapper 和 reducer，也可以指定 mapper 或 reducer 为 Linux 下的程序（如 awk、grep、cat）或者按照一定格式写好的 java class。因此，mapper 和 reducer 也不必是同一类的程序。

　　1. Hadoop Streaming 的优缺点

　　　　优点：

　　　　　　1. 可以使用自己喜欢的语言来编写 MapReduce 程序（不必非得使用 Java）

　　　　　　2. 不需要像写 Java 的 MR 程序那样 import 一大堆裤，在代码里做很多配置，很多东西都抽象到了 stdio 上，代码量显著减少。

　　　　　　3. 因为没有库的依赖，调试方便，并且可以脱离 Hadoop 先在本地用管道模拟调试。

　　　　缺点：

　　　　　　1. 只能通过命令行参数来控制 MapReduce 框架，不像 Java 的程序那样可以在代码里使用 API，控制力比较弱。

　　　　　　2. 因为中间隔着一层处理，效率会比较慢。

　　　　　　3. 所以 Hadoop Streaming 比较适合做一些简单的任务，比如用 Python 写只有一两百行的脚本。如果项目比较复杂，或者需要进行比较细致的优化，使用 Streaming 就容易出现一些束手束脚的地方。

　　2. 用 Python 编写简单的 Hadoop Streaming 程序

　　　　使用 Python 编写 Hadoop Streaming 程序有几点需要注意：

　　　　　　1. 在能使用 iterator 的情况下，尽量使用 iterator，避免将 stdin 的输入大量储存在内存里，否则会严重降低性能。

　　　　　　2. Streaming 不会帮你分割 key 和 value 传进来，传进来的只是一个个字符串而已，需要你自己在代码里手动调用 split()。

　　　　　　3. 从 stdin 得到的每一行数据末尾似乎会有 ‘\n‘ ，保险起见一般都需要用 rstrip() 来去掉。

　　　　　　4. 在想获得 key-value list 而不是一个个处理 key-value pair 时，可以使用 groupby 配合 itemgetter 将 key 相同的 key-value pair 组成一个个 group，得到类似 Java 编写的 reduce 可以直接获取一个 Text 类型的 key 和一个 iterable 作为 value 的效果。注意 itemgetter 的效率比 lambda 表达式的效率要高，所以用 itemgetter 比较好。

　　　　编写 Hadoop Streaming 程序的基本模版：

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
Some description here...
"""
import sys
from operator import itemgetter
from itertools import groupby

def read_input(file):
"""Read input and split."""
    for line in file:
    yield line.rstrip().split(‘\t‘)

def main():
    data = read_input(sys.stdin)
    for key, kviter in groupby(data, itemgetter(0)):
        # some code here..

if __name__ == "__main__":
    main()    

　　如果对输入输出格式有不同于默认的控制，主要会在 read_input() 里调整。

　　3. 本地调试

　　　　本地调试用于 Hadoop Streaming 的 Python 程序的基本模式是：

$ cat <input path> | python <path to mapper script> | sort -t $‘\t‘ -k1,1 | python <path to reducer script> > <output path>

　　　　这里有几点需要注意：

　　　　　　1. Hadoop 默认按照 tab 来分割 key 和 value，以第一个分割出的部分为 key，按 key 进行排序，因此这里使用 sort -t $‘\t‘ -k1,1 来模拟。如果有其他需求，在交给 Hadoop Streaming 执行时可以通过命令行参数设置，本地调试也可以进行相应的调整，主要是调整 sort 的参数。

　　　　　　2. 如果在 Python 脚本里加上了 shebang，并且为它们添加了执行权限，也可以用类似于 ./mapper.py (会根据 shebang 自动调用指定的解释器来执行文件）来代替 python mapper.py。