大家好,又见面了,我是全栈君,今天给大家准备了Idea注册码。
Partition中国人意味着分区,意义的碎片,这个阶段也是整个MapReduce该过程的第三阶段。在Map返回任务,是使key分到通过一定的分区算法。分到固定的区域中。给不同的Reduce做处理,达到负载均衡的目的。
他的运行过程事实上就是发生在上篇文章提到的collect的过程阶段,当输入的key调用了用户的map函数时。中间结果就会被分区了。虽说这个过程看似不是非常重要,可是也有值得学习的地方。Hadoop默认的算法是HashPartitioner,就是依据key的hashcode取摸运算,非常easy的。
/** Partition keys by their {@link Object#hashCode()}. */ public class HashPartitioner<K2, V2> implements Partitioner<K2, V2> { public void configure(JobConf job) {} /** Use {@link Object#hashCode()} to partition. */ public int getPartition(K2 key, V2 value, int numReduceTasks) { return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks; } }
可是这尽管能保证了key的随机分布。但不能保证全局有序的实现,由于有些需求须要不同的分区呈现出阶段性的分布,第一个区全部key小于第二区间。相同第二区间要小于第三区间,而你的随机算法仅仅是局部有序,在区间内时有序的。可是存在第一区间的key会大于第二区间的。因此。这里出现了一个叫
TotalOrderPartitioner的类,这也是本次学习的重点。
先看看关系Partition的相关类结构。
可见。TotalOrderPartitioner还是挺复杂的。
TotalOrderPartitioner的作用就是保证全局有序,对于key的划分,他划分了几个key的抽样点。作为key的划分点,比【2,4,6,8】,4个key抽样点。把区间划成了5份,假设某个key的值为5,他的区间为4-6。所以在第三区间,也就是说,这个类的作用就是环绕给定的划分点,寻找他的区间号,就代表任务的完毕,至于你中间用的是二分搜索。还是其它的什么算法。都由你说了算。
好的。首先第一步,从配置文件里得到划分点,他事实上是存在于一个叫partition.file的文件里,配置中仅仅保留了路径,
public void configure(JobConf job) { try { //获得partition file String parts = getPartitionFile(job); final Path partFile = new Path(parts); final FileSystem fs = (DEFAULT_PATH.equals(parts)) ? FileSystem.getLocal(job) // assume in DistributedCache : partFile.getFileSystem(job); Class<K> keyClass = (Class<K>)job.getMapOutputKeyClass(); //从partition中读出Spilts分区点 K[] splitPoints = readPartitions(fs, partFile, keyClass, job); ....
spiltPoints在后面会起着关键的作用。
然后開始关键的操作了,假设你的key值类型不是BinaryComparable二进制比較类型的话。比方能直接比較值的数字类型,就直接用二分算法。创建二分搜索节点。传入自己的比較器实现:
.... RawComparator<K> comparator = (RawComparator<K>) job.getOutputKeyComparator(); for (int i = 0; i < splitPoints.length - 1; ++i) { if (comparator.compare(splitPoints[i], splitPoints[i+1]) >= 0) { throw new IOException("Split points are out of order"); } } boolean natOrder = job.getBoolean("total.order.partitioner.natural.order", true); //推断是否为BinaryComparable类型,假设是,建立Trie树 if (natOrder && BinaryComparable.class.isAssignableFrom(keyClass)) { partitions = buildTrie((BinaryComparable[])splitPoints, 0, splitPoints.length, new byte[0], job.getInt("total.order.partitioner.max.trie.depth", 2)); } else { //假设是不是则建立构建BinarySearchNode,用二分查找。用自己构建的比較器 partitions = new BinarySearchNode(splitPoints, comparator); }
继续往里点,里面的获取分区号的算法。直接用的是二分搜索查找:
/** * For types that are not {@link org.apache.hadoop.io.BinaryComparable} or * where disabled by <tt>total.order.partitioner.natural.order</tt>, * search the partition keyset with a binary search. */ class BinarySearchNode implements Node<K> { //比較的内容节点 private final K[] splitPoints; //比較器 private final RawComparator<K> comparator; BinarySearchNode(K[] splitPoints, RawComparator<K> comparator) { this.splitPoints = splitPoints; this.comparator = comparator; } /** * 通过自己传入的比較器方法进行二分查找 */ public int findPartition(K key) { final int pos = Arrays.binarySearch(splitPoints, key, comparator) + 1; return (pos < 0) ?-pos : pos; } }
可是假设key的类型假设是
BinaryComparable二进制比較类型的呢(你能够就理解为字符串类型),则要依赖TrieTree的创建了。
里面分为2种节点,InnerTrieNode和LeafTrieNode,都继承了TrieNode , LeafTrieNode为叶子节点,最底层保存的是分区点刚刚说过的splitPoints。InnerTrieNode就是在叶子节点上面的节点。这个TrieTree的原理就是从上往下扫描节点,最后到叶子节点,返回分区号
。
有种二分搜索树的感觉。
每一个inner节点保留255个字节点,代表着255个字符
/** * An inner trie node that contains 256 children based on the next * character. */ class InnerTrieNode extends TrieNode { private TrieNode[] child = new TrieNode[256]; InnerTrieNode(int level) { super(level); } ...
所以最后的图线类似以下这样,这里仅仅显示出了A-Z 26个字母,事实上应该有255个:
能够想象这个树全然展开还是很大的,所以这是标准的空间换时间的算法实现,所以创建TrieTree的过程应该是递归的过程,直到到达最深的深度。此时应该创建的Leaf叶子节点,至此,树创建完成,看代码实现:
private TrieNode buildTrie(BinaryComparable[] splits, int lower, int upper, byte[] prefix, int maxDepth) { final int depth = prefix.length; if (depth >= maxDepth || lower == upper) { //深度抵达最大的时候,应创建叶子节点了 return new LeafTrieNode(depth, splits, lower, upper); } InnerTrieNode result = new InnerTrieNode(depth); byte[] trial = Arrays.copyOf(prefix, prefix.length + 1); // append an extra byte on to the prefix int currentBound = lower; //每一个父节点拥有着255个子节点 for(int ch = 0; ch < 255; ++ch) { trial[depth] = (byte) (ch + 1); lower = currentBound; while (currentBound < upper) { if (splits[currentBound].compareTo(trial, 0, trial.length) >= 0) { break; } currentBound += 1; } trial[depth] = (byte) ch; //result.child为首节点。递归创建子节点 result.child[0xFF & ch] = buildTrie(splits, lower, currentBound, trial, maxDepth); } // pick up the rest trial[depth] = 127; result.child[255] = buildTrie(splits, currentBound, upper, trial, maxDepth); return result; }
以上的步骤还仅仅是初始化的过程,并不是key查找获取partition分区的操作。构建过程的的流程图例如以下:
接下来的步骤就是关键的输入key,进而查找分区的过程了。非二进制比較类型的情况非常easy。直接通过自己的插入的比較器,二分搜索就可以知道结果。我们看看TrieTree实现的字符串类型的查找分区怎样实现。从以上构建的过程。我们知道,他是一层层的逐层查找过程,比方你要找,aad这个字符。你当然首先得第一个节点找a,然后再往这个节点的第一个子节点就是字符a在查找,最后找到叶子节点,在叶子节点的查找,Hadoop还是用了二分查找,这时由于本身的划分数据不是非常多,不须要排序直接查找就可以。
以下看看代码的实现,首先是innner节点,但字符的查找:
.... /** * 非叶子的节点的查询 */ public int findPartition(BinaryComparable key) { //获取当前的深度 int level = getLevel(); if (key.getLength() <= level) { return child[0].findPartition(key); } //从key在此位置相应的字符child開始继续搜寻下一个,key.getBytes()[level]为第level位置的字符 return child[0xFF & key.getBytes()[level]].findPartition(key); }
假设抵达了最后一层的LeafTrieNode。调用的是他自己的方法:
.... //在叶子节点,进行二分查找分区号 public int findPartition(BinaryComparable key) { final int pos = Arrays.binarySearch(splitPoints, lower, upper, key) + 1; return (pos < 0) ? -pos : pos; }
终于返回的也是分区号。也就完毕了这个分区算法的终于实现了。标准的空间换时间算法,可是要保证此算法的高效性,对于划分点的採集就显得很重要了。得要保证有一定的代表性。才干保证分区间的有序。
在Hadoop中提供了3个採集的类:
SplitSampler:对前n个记录进行採样
RandomSampler:遍历全部数据,随机採样
IntervalSampler:固定间隔採样
小小的partition算法也蕴藏着非常多奇异的算法,MapReduce该代码确实是一个难得的好消息啊。
版权声明:本文博客原创文章,博客,未经同意,不得转载。
发布者:全栈程序员-用户IM,转载请注明出处:https://javaforall.cn/117184.html原文链接:https://javaforall.cn
【正版授权,激活自己账号】: Jetbrains全家桶Ide使用,1年售后保障,每天仅需1毛
【官方授权 正版激活】: 官方授权 正版激活 支持Jetbrains家族下所有IDE 使用个人JB账号...