首页 > 代码库 > 用python + hadoop streaming 编写分布式程序(三) -- 自定义功能

用python + hadoop streaming 编写分布式程序(三) -- 自定义功能

又是期末又是实训TA的事耽搁了好久……先把写好的放上博客吧

前文:

用python + hadoop streaming 编写分布式程序(一) -- 原理介绍,样例程序与本地调试

用python + hadoop streaming 编写分布式程序(二) -- 在集群上运行与监控

使用额外的文件

假如你跑的job除了输入以外还需要一些额外的文件(side data),有两种选择:

  1. 大文件

    所谓的大文件就是大小大于设置的local.cache.size的文件,默认是10GB。这个时候可以用-file来分发。除此之外代码本身也可以用file来分发。

    格式:假如我要加多一个sideData.txt给python脚本用:

    $HADOOP_HOME/bin/hadoop  jar $HADOOP_HOME/hadoop-streaming.jar     -input iputDir     -output outputDir     -mapper mapper.py     -file mapper.py     -reducer reduer.py     -file reducer.py     -file sideDate.txt

    在python脚本里,只要把这个文件当成自己同一目录下的本地文件来打开就可以了。比如:

    f = open("sideData.txt")

    注意这个file是只读的,不可以写。

  2. 小文件

    如果是比较小的文件,想要提高读写速度可以将它放在distributed cache里(也就是每台机器都有自己的一份copy,不需要网络IO就可以拿到数据)。这里要用到的参数是-cachefile,写法和用法上一个一样,就是-file改成-cachefile而已。

控制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。

上面分割出key之后, KeyFieldBasedPartitioner还需要知道你想要用key里的哪部分作为partition的依据。它进行配置的过程可以看源代码来理解。

假设在上一步我们通过使用

-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换成其他各种组合也是一样。

实例说明一下,就是这样的:

  1. mapper的输出是

    11.12.1.211.14.2.311.11.4.111.12.1.111.14.2.2
  2. 指定前4个元素做key,key里的前两个元素做partition key,分成3个partition的话,就会被分成

    11.11.4.1-----------11.12.1.211.12.1.1-----------11.14.2.3 
    11.14.2.2
  3. 下一步reducer会对自己得到的每个partition内进行排序,结果就是

    11.11.4.1-----------11.12.1.111.12.1.2-----------11.14.2.211.14.2.3 

命令格式大约就是长这样

$HADOOP_HOME/bin/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     

注意-D参数放在前面,指定用KeyFieldBasedPartitioner的-partitioner要放在下面。

控制comparator与自定义排序

上面说到mapper的输出被partition到各个reducer之后,会有一步排序。这个排序的标准也是可以通过设置comparator控制的。和上面一样,要先设置分割出key用的分隔符、key的范围,key内部分割用的分隔符

-D stream.map.output.field.separator=. -D stream.num.map.output.key.fields=4 -D map.output.key.field.separator=. \

这里要控制的就是key内部的哪些元素用来做排序依据,是排字典序还是数字序,倒序还是正序。用来控制的参数是mapred.text.key.comparator.options,接受的值格式类似于unix sort。比如我要按第二个元素的数字序(默认字典序)+倒序来排元素的话,就用

-D mapred.text.key.comparator.options=-k2,2nr

n表示数字序,r表示倒序。这样一来

11.12.1.211.14.2.311.11.4.111.12.1.111.14.2.2

就会被排成

11.14.2.311.14.2.211.12.1.211.12.1.111.11.4.1