Phoenix4.6 BulkLoad OOM

Phoenix官方提供了一个导入海量数据的MapReduce工具 CsvBulkLoadTool,根据官方的说明,使用这个工具可以高效地往hbase导入csv文本数据,内部会使用phoenix api去处理数据,包括数据类型、salt rowkey处理、索引表同步等等。相比较使用psql来导入,效率会提高很多。但是在使用过程中发现导入几百万数据会在reduce阶段抛OOM异常从而导致MapReduce任务失败。这篇文章将介绍如何解决BulkLoad产生OOM的问题。

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

2. Phoenix BulkLoad简介

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

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

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

phoenix bulk load

3. BulkLoad OOM

3.1 重现

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

【测试表】

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

依照官网使用方式执行:

1
hadoop jar phoenix-<version>-client.jar org.apache.phoenix.mapreduce.CsvBulkLoadTool --table EXAMPLE --input /data/example.csv
1
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异常日志如下所示:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
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,这个类有如下说明:

1
2
3
4
5
/**
* Reducer class for the CSVBulkLoad job.
* Performs similar functionality to {@link KeyValueSortReducer}
*
*/

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

1
2
3
4
5
6
7
/**
* 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,提到了这个异常出现的原因:

issue : 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调用上进行处理。

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
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类如下所示:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
/** 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异常,问题解决。

坚持原创技术分享,您的支持将鼓励我继续创作!