MapReduce data flow:
dataflow
Hadoop Streaming:
Hadoop本身是用Java开发的，程序也需要用Java编写，但是通过Hadoop Streaming，我们可以使用任意语言来编写程序，让Hadoop运行。
streaming
Hadoop Streaming的优缺点:
优点
1.可以使用自己喜欢的语言来编写MapReduce程序（换句话说，不必写Java XD）
2.不需要像写Java的MR程序那样import一大堆库，在代码里做一大堆配置，很多东西都抽象到了stdio上，代码量显著减少
3.因为没有库的依赖，调试方便，并且可以脱离Hadoop先在本地用管道模拟调试
缺点
1.只能通过命令行参数来控制MapReduce框架，不像Java的程序那样可以在代码里使用API，控制力比较弱，有些东西鞭长莫及
2.因为中间隔着一层处理，效率会比较慢
3.所以Hadoop Streaming比较适合做一些简单的任务，比如用python写只有一两百行的脚本。如果项目比较复杂，或者需要进行比较细致的优化，使用Streaming就容易出现一些束手束脚的地方。
使用python编写Hadoop Streaming程序有几点需要注意：
1.在能使用iterator的情况下，尽量使用iterator，避免将stdin的输入大量储存在内存里，否则会严重降低性能
2.streaming不会帮你分割key和value传进来，传进来的只是一个个字符串而已，需要你自己在代码里手动调用split()
3.从stdin得到的每一行数据末尾似乎会有\n，保险起见一般都需要使用rstrip()或者strip()来去掉
4.在想获得K-V list而不是一个个处理key-value pair时，可以使用groupby配合itemgetter将key相同的k-v pair组成一个个group，得到类似Java编写的reduce可以直接获取一个Text类型的key和一个iterable作为value的效果。注意itemgetter的效率比lambda表达式要高，所以如果需求不是很复杂的话，尽量用itemgetter比较好。
本地调试：
$ cat <input path> | python <path to mapper script> | sort -t $'\t' -k1,1 | python <path to reducer script> > <output path>
Tips：
Hadoop默认按照tab来分割key和value，以第一个分割出的部分为key，按key进行排序，因此这里使用
sort -t $'\t' -k1,1
在集群上运行与监控:
为了更好地模拟集群环境，我们可以在mapred-site.xml中增设reducer和mapper的最大数目（默认为2，实际可用数目大约是CPU核数-1）
首先需要知道用于streaming的java程序在哪里。
hadoop jar /usr/local/hadoop/share/hadoop/tools/lib/hadoop-streaming-2.7.2.jar \
用Hadoop Streaming执行python程序的一般步骤是：
1.将输入文件放到HDFS上，建议使用copyFromLocal而不是put命令，参见Difference between hadoop fs -put and hadoop fs -copyFromLocal
一般可以新建一个文件夹用于存放输入文件，假设叫input
$ hadoop fs -mkdir input
$ hadoop fs -ls
查看目录，可以看到出现了一个/user/hadoop/input文件夹。/user/hadoop是默认的用户文件夹，相当于本地文件系统中的/home/hadoop
再使用
$ hadoop fs -copyFromLocal <PATH TO LOCAL FILE(S)> input/
将本地文件放到input文件夹下
2.开始MR作业
hadoop jar /usr/local/hadoop/share/hadoop/tools/lib/hadoop-streaming-2.7.2.jar \
-file mapper.py \
-mapper mapper.py \
-file reducer.py \
-reducer reducer.py \
-input $1 \
-output $2 \
一般来说要检查运行状况，都是去jobtracker的webUI。如果在master上，用浏览器访问http://localhost:50030 即可 (如果你在配置hadoop的时候修改了mapred-site.xml的mapred.job.tracker.http.address，请访问对应的其他地址）
在webUI里你可以看到running jobs, completed jobs和retired jobs。点击Jobid下的超链接，可以看到对应job的执行状况。进去后如果看到Failed/Killed Task Attempts下非空，你可以点进对应的超链接，找到对应的log去进行debug。
成功执行完这个任务之后，你用output参数在HDFS上指定的输出文件夹里就会多出几个文件
一个空白文件_SUCCESS，表明job运行成功，这个文件可以让其他程序只要查看一下HDFS就能判断这次job是否成功运行，从而进行相关处理。
一个_logs文件夹，顾名思义里面放着任务日志
part-00000, .... part-xxxxx文件，有多少个reducer后面的数字就会有多大，对应每个reducer的输出结果。
如何串联多趟MR
如果你有多次任务要执行，下一步需要用上一步的任务做输入，解决办法其实很简单。假设上一步在HDFS的输出文件夹是output1，那么在下一步的运行命令中，指明
-input output1/part-*
即指定上一次的所有输出为本次任务的输入即可。注意这里假设你不需要对上一步的输出做额外处理
控制partitioner
partitioning指的是数据经过mapper处理后，被分发到reducer上的过程。partitioner控制的，就是“怎样的mapper输出会被分发到哪一个reducer上”。Hadoop有几个自带的partitioner。默认的是HashPartitioner，也就是把第一个tab前的key做hash之后用于分配partition。写Hadoop Streaming程序是可以选择其他partitioner的，你可以选择自带的其他几种里的一种，也可以自己写一个继承Partitioner的java类然后编译成jar，在运行参数里指定为你用的partitioner。官方自带的partitioner里最常用的是KeyFieldBasedPartitioner。它会按照key的一部分来做partition，而不是用整个key来做partition。在学会用KeyFieldBasedPartitioner之前，必然要先学怎么控制key-value的分割。分割key的步骤可以分为两步，用python来描述一下大约是
fields = output.split(seperator)
key = fields[:numKeyfields]
1.选择用什么符号来分割key，也就是选择seperator
map.output.key.field.separator可以指定用于分隔key的符号。比如指定为一点的话，就要加上参数
-D stream.map.output.field.separator=.
假设你的mapper输出是
11.22.33.44
这时会先看准[11, 22, 33, 44]这里的其中一个或几个作为key
2.选择key的范围，也就是选择numKeyfields
控制key的范围的参数是这个，假设我要设置被分割出的前2个元素为key：
-D stream.num.map.output.key.fields=2
那么key就是上面的 1122。值得注意的是假如这个数字设置到覆盖整个输出，在这个例子里是4的话，那么整一行都会变成key。
假设在上一步我们通过使用
-D stream.map.output.field.separator=. \
-D stream.num.map.output.key.fields=4 \
将11.22.33.44的整个字符串都设置成了key，下一步就是在这个key的内部再进行一次分割。map.output.key.field.separator可以用来设置第二次分割用的分割符，mapred.text.key.partitioner.options可以接受参数来划分被分割出来的partition key，比如：
-D map.output.key.field.separator=. \
-D mapred.text.key.partitioner.options=-k1,2 \
指的就是在key的内部里，将第1到第2个被点分割的元素作为partition key，这个例子里也就是1122。这里的值-ki,j表示从i到j个元素（inclusive）会作为partition key。如果终点省略不写，像-ki的话，那么i和i之后的元素都会作为partition key。
partition key相同的输出会保证分到同一个reducer上，也就是所有11.22.xx.xx的输出都会到同一个partitioner，11.22换成其他各种组合也是一样。
命令格式大约就是长这样:
hadoop jar $HADOOP_HOME/hadoop-streaming.jar \
-D stream.map.output.field.separator=. \
-D stream.num.map.output.key.fields=4 \
-D map.output.key.field.separator=. \
-D mapred.text.key.partitioner.options=-k1,2 \
-input inputDir \
-output outputDir \
-mapper mapper.py -file mapper.py \
-reducer reducer.py -file reducer.py \
-partitioner org.apache.hadoop.mapred.lib.KeyFieldBasedPartitioner
控制comparator与自定义排序
用来控制的参数是mapred.text.key.comparator.options，接受的值格式类似于unix sort。比如我要按第二个元素的数字序（默认字典序）+倒序来排元素的话，就用
-D mapred.text.key.comparator.options=-k2,2nr
n表示数字序，r表示倒序。这样一来
11.12.1.2
11.14.2.3
11.11.4.1
11.12.1.1
11.14.2.2
就会被排成
11.14.2.3
11.14.2.2
11.12.1.2
11.12.1.1
11.11.4.1
网友评论