Phoenix4.6 BulkLoad OOM

待我称王封你为后i 2022-08-21 15:59 72阅读 0赞

## 1. 软件环境 ##

*  cdh5.4
 *  hbase 1.0
 *  phoenix-4.6-hbase-1.0

**【注意】**

官方提供的phoenix4.6-hbase-1.0版本并不兼容cdh5.4版本的hbase，适配方法请查看之前写的一篇文章—— [整合phoenix4.6.0-HBase-1.0到cdh5.4][phoenix4.6.0-HBase-1.0_cdh5.4]

## 2. Phoenix BulkLoad简介 ##

Phoenix 提供了一个导入海量数据的MapReduce工具 CsvBulkLoadTool，根据官方的说明，使用这个工具可以高效地往hbase导入csv文本数据，内部会使用phoenix api去处理数据，包括数据类型、salt rowkey处理、索引表同步等等。相比较使用psql来导入，效率会提高很多。

> psql实际上是采用单线程的方式来执行导入，所以效率肯定比不上使用mapreduce方式的CsvBulkLoadTool。

关于CsvBulkLoadTool这个工具类的详细介绍及使用，点击下面链接查看官网介绍：

[https://phoenix.apache.org/bulk\_dataload.html][https_phoenix.apache.org_bulk_dataload.html]

## 3. BulkLoad OOM ##

### 3.1 重现 ###

根据官网的使用说明进行较大规模数据集的测试：

【测试表】

TEST，44列，5000w行，数据文件大小大概为11G

依照官网使用方式执行：

hadoop jar phoenix-<version>-client.jar org.apache.phoenix.mapreduce.CsvBulkLoadTool --table EXAMPLE --input /data/example.csv

hadoop jar /data/phoenix-default/phoenix-4.6.0-HBase-1.0-client.jar org.apache.phoenix.mapreduce.CsvBulkLoadTool --table TEST --input /data/test.csv

发现在reduce阶段到一定进度一直抛OOM异常，直到所有重试失败，从而最终mapreduce任务失败。reduce异常日志如下所示：

2016-05-04 03:01:25,644 INFO [communication thread] org.apache.hadoop.mapred.Task: Communication exception: java.lang.OutOfMemoryError: Java heap space
        at java.lang.StringCoding.encode(StringCoding.java:338)
        at java.lang.String.getBytes(String.java:916)
        at java.io.UnixFileSystem.getBooleanAttributes0(Native Method)
        at java.io.UnixFileSystem.getBooleanAttributes(UnixFileSystem.java:242)
        at java.io.File.isDirectory(File.java:843)
        at org.apache.hadoop.yarn.util.ProcfsBasedProcessTree.getProcessList(ProcfsBasedProcessTree.java:495)
        at org.apache.hadoop.yarn.util.ProcfsBasedProcessTree.updateProcessTree(ProcfsBasedProcessTree.java:210)
        at org.apache.hadoop.mapred.Task.updateResourceCounters(Task.java:847)
        at org.apache.hadoop.mapred.Task.updateCounters(Task.java:986)
        at org.apache.hadoop.mapred.Task.access$500(Task.java:79)
        at org.apache.hadoop.mapred.Task$TaskReporter.run(Task.java:735)
        at java.lang.Thread.run(Thread.java:745)
    
    2016-05-04 03:01:25,644 FATAL [main] org.apache.hadoop.mapred.YarnChild: Error running child : java.lang.OutOfMemoryError: Java heap space
        at java.lang.StringCoding.decode(StringCoding.java:187)
        at java.lang.String.<init>(String.java:416)
        at java.lang.String.<init>(String.java:481)
        at org.apache.hadoop.io.WritableUtils.readString(WritableUtils.java:126)
        at org.apache.phoenix.mapreduce.bulkload.CsvTableRowkeyPair.readFields(CsvTableRowkeyPair.java:82)
        at org.apache.hadoop.io.serializer.WritableSerialization$WritableDeserializer.deserialize(WritableSerialization.java:71)
        at org.apache.hadoop.io.serializer.WritableSerialization$WritableDeserializer.deserialize(WritableSerialization.java:42)
        at org.apache.hadoop.mapreduce.task.ReduceContextImpl.nextKeyValue(ReduceContextImpl.java:142)
        at org.apache.hadoop.mapreduce.task.ReduceContextImpl$ValueIterator.next(ReduceContextImpl.java:239)
        at org.apache.phoenix.mapreduce.CsvToKeyValueReducer.reduce(CsvToKeyValueReducer.java:40)
        at org.apache.phoenix.mapreduce.CsvToKeyValueReducer.reduce(CsvToKeyValueReducer.java:33)
        at org.apache.hadoop.mapreduce.Reducer.run(Reducer.java:171)
        at org.apache.hadoop.mapred.ReduceTask.runNewReducer(ReduceTask.java:627)
        at org.apache.hadoop.mapred.ReduceTask.run(ReduceTask.java:389)
        at org.apache.hadoop.mapred.YarnChild$2.run(YarnChild.java:163)
        at java.security.AccessController.doPrivileged(Native Method)
        at javax.security.auth.Subject.doAs(Subject.java:415)
        at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1671)
        at org.apache.hadoop.mapred.YarnChild.main(YarnChild.java:158)

### 3.2 分析 ###

追踪源码发现CsvBulkLoadTool使用的reducer类是CsvToKeyValueReducer，这个类有如下说明：

/** * Reducer class for the CSVBulkLoad job. * Performs similar functionality to { @link KeyValueSortReducer} * */

而KeyValueSortReducer这个类是hbase官方提供的bulk load里面的一个reducer类。而它的说明如下：

/** * Emits sorted KeyValues. * Reads in all KeyValues from passed Iterator, sorts them, then emits * KeyValues in sorted order. If lots of columns per row, it will use lots of * memory sorting. * @see HFileOutputFormat */

KeyValueSortReducer提醒我们如果一行数据有很多列，那么会使用比较多的内存(体现在TreeSet上)进行排序。但是抛开这个提醒也不至于使得reduce过程发生OOM，因为依照mapreduce的原理，CsvToKeyValueReducer应该是将相同key里面的values进行迭代（一次reduce方法的调用处理相同key所对应的所有列数据——KeyValue），放到TreeSet里面进行排序，最后通过context写出。除非一行里面很多列，并且列数据很大很大，大到足矣撑爆reduce的最大内存（CDH5.4默认的reduce堆内存为768M），而我们的场景里面一行数据的预估大小为4KB，远不足以使得reducer发生OOM！

一开始尝试在网上检索相关资料，虽然有人也碰到了相同的问题，但是都没有准确的答复。最后，偶然机会下通过google检索到一个关于phoenix 4.7的issue，提到了这个异常出现的原因：

[https://issues.apache.org/jira/browse/PHOENIX-2649][https_issues.apache.org_jira_browse_PHOENIX-2649]

根据里面的描述信息，CsvToKeyValueReducer在reduce阶段会OOM的原因在于对应传入的key，即CsvTableRowkeyPair在map之后并没有被正确地处理，从而引起所有的CsvTableRowkeyPair都被分配到单个reduce调用。**注意，是一次reduce调用传入了所有map处理后的结果。**比如上面我们导入5000W数据(44 columns per row)，那么传进单个reduce方法的values总共有**（5000W \* 44）**个，所以，在TreeSet不断添加元素的过程就发生了OOM异常，从而导致mapreduce任务失败。

最终追溯到的错误根源在于，CsvTableRowkeyPair里的静态内部类Comparator没有正确处理CsvTableRowkeyPair里面的tableName以及 rowkey的与其他CsvTableRowkeyPair的tableName和rowkey的比较。Comparator的compare方法结果会影响到到reduce的inputKey（即CsvTableRowkeyPair）的分布结果。**在phoenix4.6版本里面，对于不同的CsvTableRowkeyPair（tableName + rowkey），Comparator的compare方法总是返回0（表示相等），从而导致所有的map结果都分配到一个reducer的一次reduce调用上进行处理。**

使用这个单元测试进行测试：

public class CsvTableRowKeyPairTest { 
    
        @Test
        public void testRowKeyPair() throws IOException {
    
            testsRowsKeys("first", "aa", "first", "aa", 0);
            testsRowsKeys("first", "aa", "first", "ab", -1);
            testsRowsKeys("second", "aa", "first", "aa", 1);
            testsRowsKeys("first", "aa", "first", "aaa", -1);
            testsRowsKeys("first","bb", "first", "aaaa", 1);
    
        }
    
        private void testsRowsKeys(String aTable, String akey, 
                String bTable, String bkey, int expectedSignum) throws IOException {  
    
            ImmutableBytesWritable arowkey = new ImmutableBytesWritable(Bytes.toBytes(akey));
            CsvTableRowkeyPair pair1 = new CsvTableRowkeyPair(aTable, arowkey);
    
            ImmutableBytesWritable browkey = new ImmutableBytesWritable(Bytes.toBytes(bkey));
            CsvTableRowkeyPair pair2 = new CsvTableRowkeyPair(bTable, browkey);
    
            CsvTableRowkeyPair.Comparator comparator = new CsvTableRowkeyPair.Comparator();
    
            try( ByteArrayOutputStream baosA = new ByteArrayOutputStream();
                     ByteArrayOutputStream baosB = new ByteArrayOutputStream()) {
                Assert.assertEquals(expectedSignum , signum(pair1.compareTo(pair2)));
                pair1.write(new DataOutputStream(baosA));
                pair2.write(new DataOutputStream(baosB));
                Assert.assertEquals(expectedSignum , signum(comparator.compare(baosA.toByteArray(), 0, baosA.size(), baosB.toByteArray(), 0, baosB.size())));
                Assert.assertEquals(expectedSignum, -signum(comparator.compare(baosB.toByteArray(), 0, baosB.size(), baosA.toByteArray(), 0, baosA.size())));
            }
        }
    
        private int signum(int i) {
            return i > 0 ? 1: (i == 0 ? 0: -1);
        }
    
    }

**调试发现确实在tablename或rowkey不相等的情况下，如testsRowsKeys(“first”, “aa”, “first”, “ab”, -1)，compare的结果也为0（0代表相等）。**

### 3.3 解决方案 ###

幸运的是在phoenix4.7版本已经修复了这个bug。因此在4.6版本里面我们只需要根据4.7版本来修改即可修复这个bug。修改后的Comparator类如下所示：

/** Comparator optimized for <code>CsvTableRowkeyPair</code>. */
        public static class Comparator extends WritableComparator { 
    
    // private BytesWritable.Comparator comparator = new BytesWritable.Comparator();
    
            public Comparator() {
                super(CsvTableRowkeyPair.class);
            }
    
            @Override
            public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {
                try {
    // int vintL1 = WritableUtils.decodeVIntSize(b1[s1]);
    // int vintL2 = WritableUtils.decodeVIntSize(b2[s2]);
    // int strL1 = readVInt(b1, s1);
    // int strL2 = readVInt(b2, s2);
    // int cmp = compareBytes(b1, s1 + vintL1, strL1, b2, s2 + vintL2, strL2);
    // if (cmp != 0) { 
    // return cmp;
    // }
    // int vintL3 = WritableUtils.decodeVIntSize(b1[s1 + vintL1 + strL1]);
    // int vintL4 = WritableUtils.decodeVIntSize(b2[s2 + vintL2 + strL2]);
    // int strL3 = readVInt(b1, s1 + vintL1 + strL1);
    // int strL4 = readVInt(b2, s2 + vintL2 + strL2);
    // return comparator.compare(b1, s1 + vintL1 + strL1 + vintL3, strL3, b2, s2
    // + vintL2 + strL2 + vintL4, strL4);
    
                    // Compare table names
                    int strL1 = readInt(b1, s1);
                    int strL2 = readInt(b2, s2);
                    int cmp = compareBytes(b1, s1 + Bytes.SIZEOF_INT, strL1, b2, s2 + Bytes.SIZEOF_INT, strL2);
                    if (cmp != 0) {
                        return cmp;
                    }
                    // Compare row keys
                    int strL3 = readInt(b1, s1 + Bytes.SIZEOF_INT + strL1);
                    int strL4 = readInt(b2, s2 + Bytes.SIZEOF_INT + strL2);
                    int i = compareBytes(b1, s1 + Bytes.SIZEOF_INT * 2 + strL1, strL3, b2, s2
                            + Bytes.SIZEOF_INT * 2 + strL2, strL4);
                    return i;
    
                } catch(Exception ex) {
                    throw new IllegalArgumentException(ex);
                }
            }
        }
    
        static { 
            WritableComparator.define(CsvTableRowkeyPair.class, new Comparator());
        }

修改以后，重新使用maven命令打包并替换集群上 $\{PHOENIX\}/bin下对应的`phoenix-<version>-client.jar`即可。再次执行mapreduce，没有发生OOM异常，问题解决。

[phoenix4.6.0-HBase-1.0_cdh5.4]: http://blog.csdn.net/d6619309/article/details/50358309
[https_phoenix.apache.org_bulk_dataload.html]: https://phoenix.apache.org/bulk_dataload.html
[https_issues.apache.org_jira_browse_PHOENIX-2649]: https://issues.apache.org/jira/browse/PHOENIX-2649