【Hadoop】指定分区键KeyFieldBasedPartitioner（MapReduce分层随机抽样）

项目场景：

有一个txt文件，统计的样本将是文本文件中的行。把文本分为 3 类：
i) 包含偶数个单词的行
ii) 单词数为奇数的行
iii) 包含一个或两个单词的行（将其视为一个单独的组而不是奇数或偶数组）
然后，根据算法随机选择样本。 你想怎样
随机性取决于你（确保你形成一种方式，如果你运行
多次编写代码，您最终不会选择相同的行顺序）。
输出文件应包含所有 200 个随机样本（偶数类66个，奇数类67个，其他67个）。 使用MapReduce执行。

问题描述

不能存储样本怎么进行随机抽样呢？

解决思路：

1. 尝试在reduce阶段循环多次for line in sys.stdin,第一次循环完毕获取每类的总个数n，第二次循环在for循环外面套一个while，对每一行进行判断if random.random()>=66/n(确保一类里的每一条被抽中的概率都为66/n或67/n)：失败了，reducer只能循环一次。sys.stdin 是一个一次性的数据流，一旦读取完毕，它就到达了文件的末尾。这意味着，一旦遍历完 sys.stdin，就没有更多的数据可以读取了。当 sys.stdin 中的数据被读取完毕时，它到达 EOF（文件结束标记）。在这一点上，再次尝试读取 sys.stdin 将不会得到任何数据。
2. 做类似python里shuffle的操作，然后打印前66/67行的数据成功！同一类的放到相同partitioner里，随机打乱每一类里的所有数据，在reducer抽取每类的前66/67条数据。 这里有两个部分比较困难，一是如何修改hadoop用来划分partitioner的方法；二是如何随机打乱。 首先，我搞清楚了第二个问题。在这里我利用了MapReduce过程中的sort的过程。 map过后，Hadoop会将相同的key分到一个partitioner，在partitioner内部进行字典序排序，排序后的结果会作为reducer的输入。 如何用这个排序的功能实现洗牌的效果呢？ 如果这里有一个从1-10的列表，遍历这十个数，对于每个数都随机生成一个0-1之间的随机数作为他们的index。根据index对这个列表进行排序，就可以达成随机打乱这十个数的效果。 同样的，如果我们在原本的key（也就是类别）后面加一个随机生成的数字，在sort阶段对key进行字典序排序时，就可以将同类的数据随机打乱，而这个过程是自动完成的不需要额外编写代码，也不需要使用额外的数据结构增加存储负担。

解决方案：

mapper里的输入如下：

as
was a
as a human,
shi shen me dong xi

I am a student.

有1-2个单词的一行被划分为：‘one_two_words’，其他的行里，如果单词个数为奇数个则被划分为：‘odd_words’，如果单词个数为奇数个则被划分为’even_words’。

输出包含key和value两个部分，中间用’\t’分割。

在mapper阶段，我对每一行随机生成一个(0,1)之间的随机数s，将s与类别stratum组合当作key。这样在mapper阶段就会输出字符串类型的key，它的格式大概是这样的：‘stratum \t s’。

但是我发现hadoop集群中分区partitioner的默认处理导致了hadoop集群上的最终结果与通过linux命令大不相同。在其中本地运行的代码能够返回我想要的结果，而在hadoop集群上的却不能。

这是本地运行的shell命令：

cat sampling.txt|python mapper.py|sort|python reducer.py

这是运行失败的hadoop命令：

hadoop jar $HADOOP_HOME/hadoop-streaming-3.2.3.jar -input sampling.txt -output output -mapper"python3 mapper.py"-reducer"python3 reducer.py"-file mapper.py -file reducer.py

KeyFieldBasedPartitioner

为了找到解决方案，我查看了hadoop-stream的官方文档（https://hadoop.apache.org/docs/stable/hadoop-streaming/HadoopStreaming.html#Hadoop_Partitioner_Class）。因此，在shell命令里需要对用来划分分区partitioner的那一部分key进行指定，并且指定reducer的个数。

mapred streaming  -Dstream.map.output.field.separator=’\t’  -Dstream.num.map.output.key.fields=2-Dmapreduce.partition.keypartitioner.options=-k1,1 -Dmapreduce.job.maps=3-Dmapreduce.job.reduces=3-input sampling.txt -output output -mapper"python3 mapper.py"-reducer"python3 reducer.py"-partitioner org.apache.hadoop.mapred.lib.KeyFieldBasedPartitioner -file mapper.py -file reducer.py

Here,

-D stream.map.output.field.separator=’\t’

and

-D stream.num.map.output.key.fields=2

are as explained in previous example. The two variables are used by streaming to identify the key/value pair of mapper. The Map/Reduce framework will partition the map outputs by the first field of the key using the

-D mapred.text.key.partitioner.options=-k1,1

option.

-partitioner org.apache.hadoop.mapred.lib.KeyFieldBasedPartitioner

指定了partitioner的类，KeyFieldBasedPartitioner is a library class in Hadoop and is useful for many applications.

mapper会输出这样的结果：

one_two_words    0.978976503     as
even_words    0.201848157     was a
odd_words    0.645648156     as a human,
odd_words    0.100920011     shi shen me dong xi
even_words    0.500289897     I am a student.

经过分区partitioner后，他们会被划分为3个分区：

one_two_words    0.978976503     as
---------------------------------------
odd_words    0.645648156     as a human,
odd_words    0.100920011     shi shen me dong xi
---------------------------------------
even_words    0.201848157     was a
even_words    0.891565456     I am a student.

经过在每个partitioner内部对整段key的字典序排序会变成这样，这样就实现了随机打乱，而每个reducer收到的输入也会是这样的：

one_two_words    0.978976503     as
---------------------------------------
odd_words    0.100920011     shi shen me dong xi
odd_words    0.645648156     as a human,
---------------------------------------
even_words    0.201848157     was a
even_words    0.891565456     I am a student.

在reducer里取前66/67条数据输出，这样就实现了分层随机抽样。