MongoDB和HBase的性能分析测试对比_moongodb和hbas

性能测试：SequoiaDB vs. MongoDB vs. Cassandra vs. HBase

在本篇测试报告中，我们使用Yahoo!发布的标准YCSB测试规则，对MongoDB、SequoiaDB、Cassandra、HBase进行对比，并尝试给出每种不同产品所适用的应用场景。在测试配置中，我们尽可能对全部产品做到高可用配置，而在一致性级别上则使用最终一致性。

在测试中我们会对两种类型的NoSQL数据库做横向对比，包括Document-Oriented文档类数据库、以及Big-Table宽表类数据库。由于每种类型的数据库具有很多自己独特的特性，我们不能将每种特性一一表现在该测评结果中。本测试主要针对数据库在不同任务类型下的性能指标进行，且仅依赖YCSB所提供的标准测试流程。

本测试将详细列出测试的物理环境以及配置信息，以便于读者能够使用自己的环境独立验证结果。

测试概要

1.测试产品

本测试主要对比两种类型的NoSQL数据库，包括四款不同的产品：

• MongoDB（文档类，V2.6.1）
• SequoiaDB（文档类，V1.8）
• HBase（宽表类，V0.94.6-CDH 4.5.0）
• Cassandra（宽表类，V1.1.12）

其中MongoDB作为当前市场占有率最高的数据库，可能是众多读者所关心的产品，提供丰富的数据库功能，号称是最接近关系型数据库的NoSQL产品；而SequoiaDB由前IBM DB2团队的研发人员创建，据称在性能和功能上能够与MongoDB做正面抗衡，同样提供很多MongoDB所提供的功能（例如分片、多索引等特性）。

HBase则是Hadoop框架的一员，也已经被广大企业和互联网用户所接受，我们使用的版本0.94.6是跟随CDH 4.5.0安装包的版本；而Cassandra则是与HBase类似的产品，由Facebook研发并进行开源，同样拥有广大的用户市场。

我们的测试使用由Yahoo!研究院发布的Yahoo Cloud Serving Benchmark （YCSB）基准测试，并将接口对各自产品的最新版进行了修改和适配。我们在正文后的附录中也提供了SequoiaDB的YCSB测试接口。

需要重新强调的是，每种不同的产品都有各自的应用场景。YCSB测试尽管是Yahoo!研究院提供的测试框架，但是在很多场景下并不能完全发挥出每个产品各自的特点。在本测试中，我们尝试使用YCSB框架给出最为客观的评估结果。如果对于该测试结果或配置存在疑问，我们欢迎广大读者根据自身需要重新调整，并将结果开放以供参考。

2. 测试场景

YCSB测试框架提供了丰富的场景配置机制，允许用户根据需要选择需要导入的数据量和增删改查之间相应的比例。在本测试中，我们导入一亿条数据，并对如下场景进行对比。

对于数据导入的场景，我们对单条记录插入和批量插入两个场景进行了区分。对于一些数据库来说，默认配置会在客户端将一批记录打包并统一发送给服务器，对于这类产品，尽管其接口为单条记录操作，我们依然将其归类为批量记录导入模式。

写入和查询的数据模拟典型日志记录的长度，具有以下特性：

其中，SequoiaDB与MongoDB的分片均配置为一主两从；HBase所在的HDFS设置复制份数为3；Cassandra建表时使用参数replication_factor=2。

一致性级别上，我们使用最弱的最终一致性，读写的write concern均设置为1。

3. 测试环境

本测试中，测试环境总共包含4台Dell R520物理机作为数据存储。生成数据的YCSB程序与数据库运行在同一物理环境。

注：如使用独立服务器进行YCSB的数据生成，会导致千兆网瓶颈。

整个集群的拓扑结构如图1所示：

图1：测试集群拓扑

服务器环境。本测试数据库服务器使用4台Dell R520物理机环境，每台物理机配置如下：

4. 测试方法

本测试使用YCSB标准，基于四台物理机执行。对于每种不同产品的测试流程如下：

• 安装软件
• 基于四节点部署集群，配置时尽可能基于以下准则：

• 高可用配置
• 最终一致性
• 功能与单节点环境保持一致
• 充分利用硬件资源

• 在四台物理机中部署YCSB集群，向本地集群写入读取数据
• 进行数据操作时通过YCSB产生记录的统计数量
• 根据结果生成Excel文件
• 针对其他场景重复以上步骤

并发性方面则基于以下规则：

• 单条记录插入每服务器24条线程
• 批量记录插入每服务器8条线程
• 其他所有操作每服务器36条线程

测试结果

一、 场景1：单条记录导入

图2：单条记录导入场景

在单条记录导入场景中，SequoiaDB与MongoDB使用insert方法，writeConcern设置为Normal；HBase则设置客户端缓冲区为2KB。而在错误检验方式上，由于是单条记录插入，所以MongoDB必须在每次操作后检测返回值是否成功，因此不可以使用异步插入方式。

在图2的结果中可以看到，单条记录导入操作SequoiaDB最高，总吞吐量可以达到每秒钟近7万。而HBase与Cassandra则比较接近，在5-6万之间。MongoDB在该场景中表现较差，总吞吐量不到每秒1万。

在该场景中，YCSB在4台服务器上各启动24条线程，总共并发量为96线程。

二、 场景2：批量记录导入

图3：批量记录导入场景

批量记录导入场景的结果见图3。在该场景中，SequoiaDB与MongoDB使用各自提供的bulk insert方法；HBase则设置client buffer为4MB；Cassandra不提供批量数据导入方式。

在该测试中，批量导入数据为每批次3000条记录，每节点启动8条线程，总数32线程。

测试结果显示，SequoiaDB可以达到每秒钟近19万的导入速度，而MongoDB则与单线程导入的性能接近（1万左右），HBase也没有本质提升。

三、 场景3：单纯查询

图4：单纯查询场景

图4显示单纯随机查询的场景。在该场景中MongoDB表现最为突出，整体吞吐量达到每秒钟8万以上。SequoiaDB和Cassandra类似，大约为MongoDB的一半，在4万至5万之间徘徊。而HBase表现最差，未达到每秒1万的指标。

该场景每台物理服务器使用36条客户端线程，总数144条线程。

四、 场景4：查询导入平衡

图5：查询导入平衡场景

该场景主要模拟50%的插入和50%的查询业务（图5）。其中插入业务使用单条记录插入。

最终的结果显示，SequoiaDB的整体表现最优，平均达到每秒钟超过14000TPS，而MongoDB/HBase/Cassandra则比较接近，各自不到10000TPS。

五、 场景5：更新为主

图6：更新为主场景

如图6所示，更新为主场景模拟95%更新与5%查询的场景。该场景中，SequoiaDB表现最优，结果介于5万到6万之间每秒。

而MongoDB表现相对较弱，大约在5千每秒左右的数量级。

六、 场景6：查询为主

图7：查询为主场景

在查询为主的场景中，模拟95%查询+5%更新。在该测试中，SequoiaDB与Cassandra的性能接近单纯查询的场景，而更新操作对MongoDB的损耗相对较大，使其性能仅不到3万每秒。

HBase在随机读为主的场景下相对较慢。

七、 场景7：查询最新

图8：查询最新场景

查询最新场景为95%读+5%插入，并且读取的数据尽可能是刚刚写入的数据。

从图8中可以看出，SequoiaDB对于刚刚写入至内存中便读取的场景性能最佳，达到近4万每秒。

而MongoDB和Cassandra则相比场景6有明显下降，HBase依然性能较低。

结论

从第三部分的各个场景对比中可以看出，SequoiaDB数据库在数据插入场景中表现最为突出，甚至超过本身以插入性能著称的Cassandra，混合读写场景下性能也可圈可点。而业界普及率最高的MongoDB则在单纯读取性能上最为抢眼，远超其他。

HBase与Cassandra虽然在写入性能上远高于MongoDB，但是和SequoiaDB相比仍然逊色一筹；而在主键随机读操作方面，Cassandra的新版本和之前的版本比起来性能大幅度上升，基本做到和MongoDB处于同一水平线，而HBase则远不能和其他产品相比。

当然，这些比较也仅仅局限于YCSB所做的测试，而文档类数据库能够提供的二级索引等机制并非是YCSB所测试的。因此，文档类数据库能够提供比宽表类数据库更多的应用场景。

如此看来，对于宽表类数据库来说，如果在其最有优势的主场都败给了文档类数据库，这是否意味着，HBase和Cassandra最大的优势已经不再，文档类数据库会在各个领域的性能表现超越宽表呢？

附录A：配置信息

1. MongoDB

MongoDB的分片分布如图9，不同颜色代表不同的分片，我们采用的是多个副本的分片

MongoDB的部署脚本如下(deploy.sh)：

 

 

[xml]view plaincopy

 

1. #!/bin/bash
2. sshperftest-1"mongod--configsvr--logpath/data/disk1/mongodb-cfg/configsvr.log--dbpath/data/disk1/mongodb-cfg/--port37019--journal--fork"
3. sshperftest-2"mongod--configsvr--logpath/data/disk1/mongodb-cfg/configsvr.log--dbpath/data/disk1/mongodb-cfg/--port37019--journal--fork"
4. sshperftest-3"mongod--configsvr--logpath/data/disk1/mongodb-cfg/configsvr.log--dbpath/data/disk1/mongodb-cfg/--port37019--journal--fork"
5.  
6. sshperftest-1"mongos--configdbperftest-0:37019,perftest-1:37019,perftest-2:37019--logpath/data/disk1/mongodb-data/mongos.log--fork"
7. sshperftest-2"mongos--configdbperftest-0:37019,perftest-1:37019,perftest-2:37019--logpath/data/disk1/mongodb-data/mongos.log--fork"
8. sshperftest-3"mongos--configdbperftest-0:37019,perftest-1:37019,perftest-2:37019--logpath/data/disk1/mongodb-data/mongos.log--fork"
9. sshperftest-4"mongos--configdbperftest-0:37019,perftest-1:37019,perftest-2:37019--logpath/data/disk1/mongodb-data/mongos.log--fork"
10.  
11.  
12. hostno=0
13. hosts=("perftest-1""perftest-2""perftest-3""perftest-4")
14. disknos=(1111)
15. port=37017
16. for((i=0;i<8;++i))
17. do
18. for((j=0;j<3;++j))
19. do
20. ssh${hosts[$hostno]}"mongod--replSetdg$i--logpath/data/disk${disknos[$hostno]}/mongodb-data/mongd.log--dbpath/data/disk${disknos[$hostno]}/mongodb-data/--logappend--quiet--port$port--fork"
21.  
22. letdisknos[$hostno]=${disknos[$hostno]}+1
23. lethostno=hostno+1
24. if[$hostno-eq${#hosts[@]}];then
25. lethostno=0
26. fi
27. done
28. letport=port+10
29.  
30. done

 

MongoDB的分片添加脚本如下(addshard.js)：

 

 

[java]view plaincopy

 

1. varport=37017
2. varhosts=["perftest-1","perftest-2","perftest-3","perftest-4"];
3. varhostid=0;
4. for(i=0;i<8;++i)
5. {
6. varconf=newObject();
7. conf._id='dg'+i;
8. conf.members=newArray();
9. for(j=0;j<3;++j)
10. {
11. varmember=newObject();
12. member._id=j;
13. member.host=hosts[hostid]+":"+port;
14. conf.members[j]=member;
15. hostid=hostid+1;
16. if(hostid==hosts.length)
17. {
18. hostid=0;
19. }
20. }
21. vardb=connect(conf.members[0].host+"/test");
22. rs.initiate(conf);
23. rs.conf();
24. port=port+10
25. vardb2=connect(conf.members[0].host+'/test');
26. sh.addShard('dg'+i+'/'+conf.members[0].host)
27. }

 

MongoDB的集合创建脚本如下(createcl.sh)：

 

 

[js]view plaincopy

 

1. mongo<<EOF
2. sh.stopBalancer();
3. useycsb;
4. db.dropDatabase();
5. useadmin;
6. db.runCommand({enableSharding:"ycsb"});
7. useadmin;
8. db.runCommand({shardcollection:"ycsb.usertable",key:{_id:'hashed'},numInitialChunks:4096});
9. exit
10. EOF

 

所有的writeConcern都为normal。

2. SequoiaDB

SequoiaDB的数据组分布情况如图10，其中不同颜色代表不同的分片。

SequoiaDB的部署脚本如下(deploy.js)：

 

 

[java]view plaincopy

 

1. try
2. {
3. vardb=newSdb();
4. db.createCataRG('perftest-1',11820,'/opt/sequoiadb/database/cata/11820');
5. db.getRG(1).createNode('perftest-2',11820,'/opt/sequoiadb/database/cata/11820');
6. db.getRG(1).createNode('perftest-3',11820,'/opt/sequoiadb/database/cata/11820');
7. db.getRG(1).getNode('perftest-2',11820).start();
8. db.getRG(1).getNode('perftest-3',11820).start();
9.  
10. //groupnumberis8
11. varport=11830;
12. varhostid=0;
13. vardiskno=1;
14. vardiskids=[1,1,1,1];
15. for(i=0;i<8;++i)
16. {
17. db.createRG('dg'+i);
18. //3nodesofpergroup
19. for(j=0;j<3;++j)
20. {
21. db.getRG('dg'+i).createNode('perftest-'+(hostid+1),port,'/data/disk'+diskids[hostid]+'/sequoiadb/database/data'+port);
22. diskids[hostid]+=1;
23. hostid+=1;
24. if(hostid>3)
25. {
26. hostid=0;
27. }
28. }
29. db.getRG('dg'+i).start();
30. port+=10;
31. }
32. }catch(e)
33. {
34. throwe;
35. }

 

创建分区集合的脚本如下(createcl.js)：

 

 

[js]view plaincopy

 

1. try
2. {
3. vardb=newSdb()
4. db.dropCS('ycsb')
5. }catch(e)
6. {
7. if(e!=-34)
8. {
9. throw"dropcsfailure"+e;
10. }
11. }
12.  
13. try
14. {
15. db.createCS('ycsb')
16. db.ycsb.createCL('usertable',{ShardingType:'hash',ShardingKey:{_id:1},EnsureShardingIndex:false})
17. varsnapshot8=db.snapshot(8,{Name:'ycsb.usertable'}).toArray();
18. varobj=eval("("+snapshot8[0]+")");
19. varsrcgroup=obj["CataInfo"][0]["GroupName"];
20. varpartition=obj["Partition"];
21.  
22. vargroupnames=newArray()
23. vargroups=db.list(7).toArray();
24. for(i=0;i<groups.length;++i)
25. {
26. vargroup=eval("("+groups[i]+")");
27. if(group["GroupName"]=="SYSCatalogGroup")
28. {
29. continue;
30. }
31. groupnames.push(group["GroupName"]);
32. }
33.  
34. varremainderpart=partition%groupnames.length;
35. varpart=(partition-remainderpart)/groupnames.length
36. for(i=0;i<groupnames.length;++i)
37. {
38. if(groupnames[i]==srcgroup)
39. {
40. continue;
41. }
42. println("splitingfrom"+srcgroup+"to"+groupnames[i]+"........");
43. db.ycsb.usertable.split(srcgroup,groupnames[i],{Partition:(i*part)},{Partition:(i+1)*part});
44. if(remainderpart>1)
45. {
46. db.ycsb.usertable.split(srcgroup,groupnames[i],{Partition:endpart},{Partition:(endpart+1)})
47. endpart+=1;
48. remainderpart-=1;
49. }
50. }
51. }catch(e)
52. {
53. throwe;
54. }

 

3. HBase

HBase的数据分布情况如图11：

图11：HBase部署架构图

创始表语句使用：

create 'usertable', 'cf', {SPLITS => ['user1', 'user2', 'user3', 'user4', 'user5', 'user6', 'user7', 'user8', 'user9' ]}

5.4Cassandra

图12是一个Cassandra四节点集群。我们采用使用二十四块硬盘同时处理数据和提交日志。

图12：Cassandra部署架构图

与测试的其他数据库不同，Cassandra 在配置中使用环形拓扑，节点需要被明确地视为“种子”节点（这有助于它们加入到环中）。在配置时，必须指定哪些令牌将映射到哪些实例。

我们使用了https://raw.github.com/riptano/ComboAMI/2.2/tokentoolv2.py提供的令牌生成工具来创建节点配置。

 

 

[js]view plaincopy

 

1. $./tokentoolv2.py4
2. {
3. "0":{
4. "0":0,
5. "1":42535295865117307932921825928971026432,
6. "2":85070591730234615865843651857942052864,
7. "3":127605887595351923798765477786913079296
8. }
9. }

 

Cassandra 的一致性级别可以调节。每次读取和写入都可以明确地说明该操作需要什么级别的数据库一致性。由于这是一个基准测试项目，因此我们使用了最弱和最快的一致性级别（ONE）来进行读取和写入。

对于所有数据库，我们使用的复制因子都是 2。其他主要设置为：

以下内容为 conf/cassandra.yaml 的设置：

 

 

[js]view plaincopy

 

1. cluster_name:'Test'
2. initial_token:0
3.  
4. hinted_handoff_enabled:true
5. max_hint_window_in_ms:3600000#onehour
6. hinted_handoff_throttle_delay_in_ms:1
7. authenticator:org.apache.cassandra.auth.AllowAllAuthenticator
8. authority:org.apache.cassandra.auth.AllowAllAuthority
9. partitioner:org.apache.cassandra.dht.RandomPartitioner
10. data_file_directories:
11. -/data/disk1/cassandra-data
12. -/data/disk2/cassandra-data
13. -/data/disk3/cassandra-data
14. -/data/disk4/cassandra-data
15. -/data/disk5/cassandra-data
16. -/data/disk6/cassandra-data
17.  
18. commitlog_directory:
19. /data/disk1/cassandra-log/,/data/disk2/cassandra-log/,/data/disk3/cassandra-log/,/data/disk4/cassandra-log/
20. ,/data/disk5/cassandra-log/,/data/disk6/cassandra-log/
21.  
22. saved_caches_directory:/data/disk1/apache-cassandra/saved_caches
23. commitlog_sync:periodic
24. commitlog_sync_period_in_ms:10000
25.  
26. seed_provider:
27. -class_name:org.apache.cassandra.locator.SimpleSeedProvider
28. parameters:
29. -seeds:"192.168.30.62,192.168.30.64,192.168.30.65,192.168.30.67"
30.  
31.  
32. flush_largest_memtables_at:0.75
33.  
34. reduce_cache_sizes_at:0.85
35. reduce_cache_capacity_to:0.6
36.  
37. concurrent_reads:48
38. concurrent_writes:48
39.  
40. memtable_flush_queue_size:4
41.  
42. sliced_buffer_size_in_kb:64
43.  
44. storage_port:7000
45.  
46. ssl_storage_port:7001
47.  
48. listen_address:192.168.30.62
49.  
50. rpc_address:0.0.0.0
51. rpc_port:9160
52.  
53. rpc_keepalive:true
54.  
55. rpc_server_type:sync
56. thrift_framed_transport_size_in_mb:15
57.  
58. thrift_max_message_length_in_mb:16
59.  
60. incremental_backups:false
61.  
62. snapshot_before_compaction:false
63. column_index_size_in_kb:64
64.  
65. in_memory_compaction_limit_in_mb:64
66.  
67. multithreaded_compaction:false
68.  
69. compaction_throughput_mb_per_sec:16
70.  
71. compaction_preheat_key_cache:true
72.  
73. rpc_timeout_in_ms:10000
74. endpoint_snitch:org.apache.cassandra.locator.SimpleSnitch
75.  
76. dynamic_snitch_update_interval_in_ms:100
77.  
78. dynamic_snitch_reset_interval_in_ms:600000
79. dynamic_snitch_badness_threshold:0.1
80. request_scheduler:org.apache.cassandra.scheduler.NoScheduler
81.  
82. index_interval:128
83.  
84. encryption_options:
85. internode_encryption:none
86. keystore:conf/.keystore
87. keystore_password:cassandra
88. truststore:conf/.truststore
89. truststore_password:cassandra

 

使用以下命令对数据库进行初始化：

 

 

[js]view plaincopy

 

1. CREATEKEYSPACEusertable
2. WITHplacement_strategy='org.apache.cassandra.locator.SimpleStrategy'ANDstrategy_options={replication_factor:2};
3. useusertable;
4. CREATECOLUMNFAMILYdataWITHcomparator=UTF8Type
5. ANDkey_validation_class=UTF8Type
6.  

 

附录B：YCSB调整

一、 驱动调整

1. MongoDB

• 默认没有采用连接池的形式实现，调整为连接池形式
• 默认不支持批量插入，增加支持批量插入
• 默认不支持选择查询接口，增加支持选择查询接口
• 默认不支持选择readpreference,增加支持选择readpreference
• 为适应2.12.1版本的driver作了些调整

详细调整如下:

 

 

[js]view plaincopy

 

1. voidinit()
2. {
3. ...
4. Propertiesprops=getProperties();
5. Stringurl=props.getProperty("mongodb.url",
6. "mongodb://localhost:27017");
7. database=props.getProperty("mongodb.database","ycsb");
8. StringwriteConcernType=props.getProperty("mongodb.writeConcern",
9. "safe").toLowerCase();
10. //finalStringmaxConnections=props.getProperty(
11. //"mongodb.maxconnections","100");
12. insertmode=props.getProperty("mongodb.insertmode","single");
13. readpreference=props.getProperty("mongodb.readpreference",
14. "primary");
15. bulknumber=Integer.parseInt(props.getProperty("mongodb.bulknumber",
16. "5000"));
17. finalStringfind=props.getProperty("mongodb.usefindinterface",
18. "false");
19. if(replWriteNum!=1){
20. writeConcern=newWriteConcern(replWriteNum);
21. }
22.  
23. try{
24. //stripoutprefixsinceJavadriverdoesn'tcurrentlysupport
25. //standardconnectionformatURLyet
26. //<ahref="http://www.mongodb.org/display/DOCS/Connections">http://www.mongodb.org/display/DOCS/Connections</a>
27. /*if(url.startsWith("mongodb://")){
28. url=url.substring(10);
29. }*/
30. //needtoappenddbtourl.
31. url+="/"+database;
32. System.out.println("newdatabaseurl="+url);
33. MongoClientURIuri=newMongoClientURI(url);
34. mongo=newMongoClient(uri);
35. mongo.setReadPreference(ReadPreference.valueOf(readpreference));
36. mongo.setWriteConcern(writeConcern);
37. System.out.println("mongoconnectioncreatedwith"+url);
38. }
39. catch(Exceptione1){
40. System.err
41. .println("CouldnotinitializeMongoDBconnectionpoolforLoader:"
42. +e1.toString());
43. e1.printStackTrace();
44. return;
45. }
46. }
47. publicintinsert(Stringtable,Stringkey,
48. HashMap<String,ByteIterator>values){
49. com.mongodb.DBdb=null;
50. try{
51. db=mongo.getDB(database);
52. if(!outputclientflag){
53. CommandResultcommandResult=db.command("buildInfo");
54. if(commandResult.containsField("tokumxVersion")){
55. System.out.println("tokumx");
56. }
57. else{
58. System.out.println("mongodb");
59. }
60. outputclientflag=true;
61. }
62. db.requestStart();
63.  
64. DBCollectioncollection=db.getCollection(table);
65. DBObjectr=newBasicDBObject().append("_id",key);
66. for(Stringk:values.keySet()){
67. r.put(k,values.get(k).toArray());
68. }
69. //WriteResultres=null;
70. if(insertmode.equals("bulk")){
71. objs.add(r);
72.  
73. //bulkwrite.insert(r);
74. if(objs.size()==bulknumber){
75. //res=
76. collection.insert(objs);
77. objs.clear();
78. //return0;
79. }
80. }else{
81. //res=
82. collection.insert(r);
83. }
84.  
85. //returnres.getN()==replWriteNum?0:1;
86. return0;
87.  
88. }
89. catch(Exceptione){
90. e.printStackTrace();
91. return1;
92. }
93. finally{
94. if(db!=null){
95. db.requestDone();
96. }
97. }
98. }
99.  
100. publicintread(Stringtable,Stringkey,Set<String>fields,
101. HashMap<String,ByteIterator>result){
102. com.mongodb.DBdb=null;
103. DBCursorcursor=null;
104. try{
105. db=mongo.getDB(database);
106. db.requestStart();
107. //getCollection(table);
108. DBCollectioncollection=db.getCollection(table);
109. DBObjectq=newBasicDBObject().append("_id",key);
110. DBObjectfieldsToReturn=newBasicDBObject();
111.  
112. DBObjectqueryResult=null;
113. //DBCursorcursor=null;
114. if(fields!=null){
115. Iterator<String>iter=fields.iterator();
116. while(iter.hasNext()){
117. fieldsToReturn.put(iter.next(),INCLUDE);
118. }
119. if(findone){
120. queryResult=collection.findOne(q,fieldsToReturn);
121. }
122. else{
123. cursor=collection.find(q,fieldsToReturn);
124. }
125. }
126. else{
127. if(findone){
128. queryResult=collection.findOne(q);
129. }
130. else{
131. cursor=collection.find(q).setReadPreference(ReadPreference.secondaryPreferred());
132. }
133. }
134.  
135. if(cursor!=null&&cursor.hasNext()){
136. queryResult=cursor.next();
137. }
138.  
139. if(queryResult!=null){
140. result.putAll(queryResult.toMap());
141. }
142. returnqueryResult!=null?0:1;
143. }
144. catch(Exceptione){
145. System.err.println(e.toString());
146. return1;
147. }
148. finally{
149. if(db!=null){
150. db.requestDone();
151. }
152. if(cursor!=null){
153. cursor.close();
154. }
155. }
156. }

 

2. HBase

• 支持通过参数控制WriteBufferSize
• 适应驱动做微调

详细如下：

 

 

[java]view plaincopy

 

1. publicvoidinit()throwsDBException
2. {
3. if((getProperties().getProperty("debug")!=null)&&
4. (getProperties().getProperty("debug").compareTo("true")==0))
5. {
6. _debug=true;
7. }
8.  
9. _columnFamily=getProperties().getProperty("columnfamily");
10. if(_columnFamily==null)
11. {
12. System.err.println("Error,mustspecifyacolumnfamilyforHBasetable");
13. thrownewDBException("Nocolumnfamilyspecified");
14. }
15. _columnFamilyBytes=Bytes.toBytes(_columnFamily);
16.  
17. clientbufsize=Integer.parseInt(getProperties().getProperty("clientbuffersize"));
18. }
19.  
20.  
21. publicvoidgetHTable(Stringtable)throwsIOException
22. {
23. synchronized(tableLock){
24. _hTable=newHTable(config,table);
25. //2suggestionsfrom<ahref="http://ryantwopointoh.blogspot.com/2009/01/performance-of-hbase-importing.html">http://ryantwopointoh.blogspot.com/2009/01/performance-of-hbase-importing.html</a>
26. _hTable.setAutoFlush(false);
27. _hTable.setWriteBufferSize(clientbufsize*1024);
28. //returnhTable;
29. }
30.  
31. }

 

二、 统计数据收集

从原有的Measurements派生出ExcelMeasurementsExporter用于将生成的统计数据导出到excel文件中，ExcelMeasurementsExporter调用jxl.jar开源库实现。

统计数据由Overalloperresult、Overallresult,Periodresult这几个类存储，为了保存统计数据原来的Measurements，StatusThread都相应作了些调整。

三、 预热

增加如下xml配置文件

 

 

[xml]view plaincopy

 

1. <?xmlversion="1.0"encoding="utf-8"?>
2. <Test>
3. <load>
4. <transactionname="bulkload_concurrent"/>
5. <!--transactionname="bulkload"/-->
6. <transactionname="singleload_concurrent"/>
7. <!--transactionname="singleload"/-->
8. </load>
9. <run>
10. <transactionname="readonly_concurrent"/>
11. <transactionname="readheavy_concurrent"/>
12. <transactionname="updateheavy_concurrent"/>
13. <transactionname="insertheavy_concurrent"/>
14. <transactionname="readlastest_concurrent"/>
15. </run>
16. </Test>

 

我们增加了如下python脚本用于连续运行：

 

 

[py]view plaincopy

 

1. #!/usr/bin/python
2. #!/bin/envpython
3.  
4. importos
5. importsys
6. importsubprocess
7. fromxml.etreeimportElementTree
8. importycsb
9. fromycsbimport(DATABASES,COMMANDS,BASE_URL,
10. get_ycsb_home,find_jars)
11.  
12. defgetloadtrans(workloadpath,root):
13. load=[]
14. lst_node=root.find("load")
15. fornodeinlst_node:
16. load.append(workloadpath+node.attrib['name'])
17. returnload
18.  
19. defgetruntrans(workloadpath,root):
20. run=[]
21. lst_node=root.find("run")
22. fornodeinlst_node:
23. run.append(workloadpath+node.attrib['name'])
24. returnrun
25.  
26. defushelp():
27. print"Usage:%sdatabase"%sys.argv[0]
28. print"nDatabases:"
29. fordbinsorted(DATABASES.keys()):
30. print"%s%s"%(db.ljust(13),BASE_URL+db.split("-")[0])
31. sys.exit(1)
32.  
33. defrunscene(trans,cmd,db_classname,pervscene):
34. curscene=""
35. foritemintrans:
36. curscene=os.path.basename(item)
37. command=COMMANDS[cmd]["command"]
38. options=["-s","-P",item]
39. ycsb_command=["java","-cp",":".join(find_jars(ycsb_home,database)),
40. COMMANDS[cmd]["main"],"-db",db_classname]+options
41. ifcommand:
42. ycsb_command.append(command)
43. #print"".join(ycsb_command)
44. subprocess.call(ycsb_command)
45. pervscene=curscene
46. returnpervscene
47.  
48. iflen(sys.argv)<2:
49. ushelp()
50. ifsys.argv[1]notinDATABASES:
51. print"ERROR:Database'%s'notfound"%sys.argv[1]
52. ushelp()
53.  
54. os.chdir('..')
55. conffile=os.getcwd()+"/conf/workload.xml"
56. resultfile=os.getcwd()+"/result/report_output.xls"
57. resultdir=os.getcwd()+"/result/"
58. workloadsdir=os.getcwd()+"/workloads/"
59.  
60. ifFalse==os.path.exists(conffile):
61. printconffile+"notexist";
62. sys.exit(1)
63.  
64. root=ElementTree.parse(conffile)
65. loadtrans=getloadtrans(workloadsdir,root)
66. runtrans=getruntrans(workloadsdir,root)
67.  
68. os.chdir('bin')
69. ycsb_home=get_ycsb_home()
70. database=sys.argv[1]
71. db_classname=DATABASES[database]
72.  
73. runscene(loadtrans,"load",db_classname,pervscene)
74. runscene(runtrans,"run",db_classname,pervscene)
75.  
76. importtime
77. curtime=time.strftime("%Y-%m-%d_%H_%M_%S",time.localtime(time.time()))
78. newreportfile=resultdir+"report_"+curtime+".xls"
79. os.rename(resultfile,newreportfile)

 

为尽量保证后续的查询、更新操作是基于前续的load操作，以保证缓存的高命中率。

四、 数据类型

本次测试的数据皆为字符串类型：

• fieldcount=10
• fieldlength=10
• key字段由单词"user"后面加上64位的Fowler-Noll-Vo哈希值构成
• key大小为23字节
• 所有字段的值采用zipfian算法生成

附录C：SequoiaDB接口

 

 

[java]view plaincopy

 

1. packagecom.yahoo.ycsb.db;
2. importcom.yahoo.ycsb.ByteArrayByteIterator;
3. importcom.yahoo.ycsb.ByteIterator;
4. importcom.yahoo.ycsb.DB;
5. importcom.yahoo.ycsb.DBException;
6.  
7. importjava.util.ArrayList;
8. importjava.util.HashMap;
9. importjava.util.Iterator;
10. importjava.util.Map;
11. importjava.util.Properties;
12. importjava.util.Set;
13. importjava.util.Vector;
14. importjava.util.List;
15. importjava.util.concurrent.atomic.AtomicInteger;
16.  
17. importorg.bson.BSONObject;
18. importorg.bson.BasicBSONObject;
19. importorg.bson.types.BasicBSONList;
20.  
21. importcom.sequoiadb.base.SequoiadbOption;
22. importcom.sequoiadb.base.SequoiadbDatasource;
23. importcom.sequoiadb.base.CollectionSpace;
24. importcom.sequoiadb.base.DBCollection;
25. importcom.sequoiadb.base.DBCursor;
26. importcom.sequoiadb.base.Sequoiadb;
27.  
28.  
29. publicclassSequoiaDBClientextendsDB{
30.  
31. /**Usedtoincludeafieldinaresponse.*/
32. protectedstaticfinalIntegerINCLUDE=Integer.valueOf(1);
33.  
34. /**Thekeyfieldname*/
35. //privatestaticfinalStringKEY_FILED="key";
36. privatestaticfinalStringKEY_FILED="_id";
37. privatestaticfinalStringDEFAULT_INSERTMODE="single";
38.  
39. /**
40. *Countthenumberoftimesinitializedtoteardownonthelast
41. *{@link#cleanup()}.
42. */
43. privatestaticfinalAtomicIntegerinitCount=newAtomicInteger(0);
44. privatestaticintbulknum=0;
45.  
46. /**Sequoiadbinstance.*/
47. privatestaticSequoiadbDatasourcesdbpools=null;
48. privateSequoiadbsdb=null;
49.  
50. /**CollectionSpaceinstance.*/
51. privateCollectionSpacecs=null;
52. privateDBCollectioncl=null;
53. privatestaticStringkeyfield=null;
54. privatestaticStringinsertmode=null;
55. privateList<BSONObject>objs=newArrayList<BSONObject>();
56. //privateintcallnumber=0;
57. privatestaticStringspacename="ycsb";
58.  
59. //privateDBCollectioncollection=null;
60.  
61. /**
62. *InitializeanystateforthisDB.CalledonceperDBinstance;thereis
63. *oneDBinstanceperclientthread.
64. */
65. publicvoidinit()throwsDBException{
66. initCount.incrementAndGet();
67. synchronized(INCLUDE){
68. if(sdb!=null){
69. return;
70. }
71.  
72. try{
73. if(sdbpools!=null){
74. sdb=sdbpools.getConnection();
75. cs=sdb.getCollectionSpace(spacename);
76. return;
77. }
78. }catch(Exceptione){
79. e.printStackTrace();
80. return;
81. }
82.  
83. //initializesequoiadbdriver
84. Propertiesprops=getProperties();
85. Stringhost=props.getProperty("sequoiadb.host","localhost");
86. Stringport=props.getProperty("sequoiadb.port","11810");
87. keyfield=props.getProperty("sequoiadb.keyfield","_id");
88. intmaxConnectionnum=Integer.parseInt(props.getProperty("sequoiadb.maxConnectionnum","100"));
89. intmaxidleconnnum=Integer.parseInt(props.getProperty("sequoiadb.maxConnectionnum","10"));
90. intperiod=Integer.parseInt(props.getProperty("sequoiadb.maxConnectionnum","300"));
91.  
92. //String
93. spacename=props.getProperty("sequoiadb.space",spacename);
94. insertmode=props.getProperty("sequoiadb.insertmode",DEFAULT_INSERTMODE);
95. bulknum=Integer.parseInt(props.getProperty("sequoiadb.bulknumber","5000"));
96.  
97. try{
98. SequoiadbOptionsdbOption=newSequoiadbOption();
99. sdbOption.setMaxConnectionNum(maxConnectionnum);
100. sdbOption.setMaxIdeNum(maxidleconnnum);
101. sdbOption.setRecheckCyclePeriod(period*1000);
102.  
103. sdbpools=newSequoiadbDatasource(host+":"+port,"","",sdbOption);
104. //needtoappenddbtourl.
105. //sdb=newSequoiadb(host,Integer.parseInt(port),"","");
106. sdb=sdbpools.getConnection();
107. if(!sdb.isCollectionSpaceExist(spacename)){
108. cs=sdb.createCollectionSpace(spacename);
109. }else{
110. cs=sdb.getCollectionSpace(spacename);
111. }
112. System.out.println("sequoiadbconnectioncreatedwith"+host
113. +":"+port);
114.  
115. }catch(Exceptione1){
116. System.err
117. .println("CouldnotinitializeSequoiadbconnectionpoolforLoader:"
118. +e1.toString());
119. e1.printStackTrace();
120. thrownewDBException(e1.toString());
121. }
122. }
123. }
124.  
125. /**
126. *CleanupanystateforthisDB.CalledonceperDBinstance;thereisone
127. *DBinstanceperclientthread.
128. */
129. publicvoidcleanup()throwsDBException{
130. initCount.decrementAndGet();
131. try{
132. if(0!=objs.size()){
133. cl.bulkInsert(objs,DBCollection.FLG_INSERT_CONTONDUP);
134. }
135. sdbpools.close(sdb);
136. }catch(Exceptione1){
137. System.err.println("CouldnotcloseSequoiadbconnectionpool:"
138. +e1.toString());
139. e1.printStackTrace();
140. return;
141. }
142. }
143. privateList<String>getAllDataGroup(){
144. //获取数据组的数量
145. List<String>groups=newArrayList<String>();
146. DBCursorcursor=sdb.getList(Sequoiadb.SDB_LIST_GROUPS,null,null,null);
147. while(cursor.hasNext()){
148. BSONObjectobj=cursor.getNext();
149. Stringgroupname=(String)obj.get("GroupName");
150. if(!groupname.equals("SYSCatalogGroup")){
151. groups.add(groupname);
152. }
153. }
154. returngroups;
155. }
156.  
157. privateintgetPartition(Stringspacename,Stringtablename){
158. //获取源数据组
159. BSONObjectcondition=newBasicBSONObject();
160. condition.put("Name",spacename+"."+tablename);
161. DBCursorcr=sdb.getSnapshot(Sequoiadb.SDB_SNAP_CATALOG,condition,null,null);
162. intPartition=0;
163. while(cr.hasNext()){
164. BSONObjectobj=cr.getNext();
165. Partition=((Integer)obj.get("Partition")).intValue();
166. }
167. returnPartition;
168. }
169.  
170. privateStringgetSrcDataGroup(Stringspacename,Stringtablename){
171. //获取源数据组
172. BSONObjectcondition=newBasicBSONObject();
173. condition.put("Name",spacename+"."+tablename);
174. DBCursorcr=sdb.getSnapshot(Sequoiadb.SDB_SNAP_CATALOG,condition,null,null);
175. Stringsrcgroup="";
176. while(cr.hasNext()){
177. BSONObjectobj=cr.getNext();
178. BasicBSONListcatainfo=(BasicBSONList)obj.get("CataInfo");
179. srcgroup=(String)((BSONObject)catainfo.get(0)).get("GroupName");
180. }
181.  
182. returnsrcgroup;
183. }
184.  
185. privatevoidsplitCollection(DBCollectioncl,Stringspacename,Stringtablename){
186. //获取数据组的数量
187. List<String>groups=getAllDataGroup();
188. intPartition=getPartition(spacename,tablename);
189. Stringsrcgroup=getSrcDataGroup(spacename,tablename);
190.  
191. intpart=Partition/groups.size();
192. intremainder=Partition%groups.size();
193. intstartpart=Partition-remainder;
194. for(inti=0;i<groups.size();++i){
195. //切分
196. BSONObjectstart=newBasicBSONObject();
197. start.put("Partition",i*part);
198. BSONObjectend=newBasicBSONObject();
199. end.put("Partition",(i+1)*part);
200. if(!groups.get(i).equals(srcgroup)){
201. cl.split(srcgroup,groups.get(i),start,end);
202.  
203. if(0!=remainder){
204. BSONObjectremainderstart=newBasicBSONObject();
205. remainderstart.put("Partition",startpart);
206. BSONObjectremainderend=newBasicBSONObject();
207. remainderend.put("Partition",startpart+1);
208. cl.split(srcgroup,groups.get(i),remainderstart,remainderend);
209. --remainder;
210. }
211. }
212. }
213. }
214.  
215. privatevoidcreateCollection(Stringtable)throwsDBException{
216. BSONObjectoptions=newBasicBSONObject();
217. BSONObjectsubobj=newBasicBSONObject();
218. subobj.put(KEY_FILED,1);
219. options.put("ShardingKey",subobj);
220. options.put("ShardingType","hash");
221. options.put("EnsureShardingIndex",false);
222.  
223. cl=cs.createCollection(table,options);
224. splitCollection(cl,spacename,table);
225.  
226. if(0!=keyfield.compareTo("_id")){
227. cl.createIndex("index",
228. "{"+keyfield+":1}",true,true);
229. }
230. }
231.  
232. privateDBCollectiongetCollection(Stringtable){
233. if(sdb==null){
234. try{
235. sdb=sdbpools.getConnection();
236. }catch(Exceptione)
237. {
238. e.printStackTrace();
239. returnnull;
240. }
241. }
242.  
243. if(cs==null){
244. try{
245. cs=sdb.getCollectionSpace(spacename);
246. }catch(Exceptione)
247. {
248. e.printStackTrace();
249. returnnull;
250. }
251. }
252.  
253. if(cl==null){
254. try{
255. booleanbExist=cs.isCollectionExist(table);
256. if(!bExist){
257. synchronized(INCLUDE){
258. if(cs.isCollectionExist(table)){
259. cl=cs.getCollection(table);
260. }else{
261. createCollection(table);
262. }
263. }
264. }else{
265. cl=cs.getCollection(table);
266. }
267. }catch(Exceptione)
268. {
269. e.printStackTrace();
270. cl=null;
271. sdbpools.close(sdb);
272. sdb=null;
273. returngetCollection(table);
274. }
275. }
276. returncl;
277. }
278.  
279. @Override
280. publicintread(Stringtable,Stringkey,Set<String>fields,
281. HashMap<String,ByteIterator>result){
282. DBCursorcursor=null;
283. DBCollectioncollection=null;
284. try{
285. collection=getCollection(table);
286. if(collection==null){
287. System.out.println("Failedtogetcollection"+table);
288. }
289.  
290. BSONObjectquery=newBasicBSONObject().append(keyfield,key);
291. BSONObjectfieldsToReturn=null;
292. if(fields!=null){
293. fieldsToReturn=newBasicBSONObject();
294. Iterator<String>iter=fields.iterator();
295. while(iter.hasNext()){
296. fieldsToReturn.put(iter.next(),"");
297. }
298. }
299.  
300. cursor=collection.query(query,fieldsToReturn,null,null);
301. if(cursor!=null&&cursor.hasNext()){
302. HashMap<String,ByteIterator>resultMap=newHashMap<String,ByteIterator>();
303.  
304. fillMap(resultMap,cursor.getNext());
305. result.putAll(resultMap);
306. return0;
307. }else{
308. return1;
309. }
310. }catch(Exceptione){
311. System.err.println(e.toString());
312. e.printStackTrace();
313. return1;
314. }finally{
315. if(cursor!=null){
316. cursor.close();
317. }
318. }
319. }
320.  
321. /**
322. *TODO-Finish
323. *
324. *@paramresultMap
325. *@paramobj
326. */
327. @SuppressWarnings("unchecked")
328. protectedvoidfillMap(HashMap<String,ByteIterator>resultMap,
329. BSONObjectobj){
330. Map<String,Object>objMap=obj.toMap();
331. for(Map.Entry<String,Object>entry:objMap.entrySet()){
332. if(entry.getValue()instanceofbyte[]){
333. resultMap.put(entry.getKey(),newByteArrayByteIterator(
334. (byte[])entry.getValue()));
335. }
336. }
337. }
338.  
339. @Override
340. publicintscan(Stringtable,Stringstartkey,intrecordcount,
341. Set<String>fields,Vector<HashMap<String,ByteIterator>>result){
342. DBCursorcursor=null;
343. try{
344. DBCollectioncollection=getCollection(table);
345.  
346. BSONObjectscanRange=newBasicBSONObject().append("$gte",
347. startkey);
348. BSONObjectquery=newBasicBSONObject().append(keyfield,
349. scanRange);
350. BSONObjectfieldsToReturn=null;
351. if(fields!=null){
352. fieldsToReturn=newBasicBSONObject();
353. Iterator<String>iter=fields.iterator();
354. while(iter.hasNext()){
355. fieldsToReturn.put(iter.next(),"");
356. }
357. }
358.  
359. cursor=collection.query(query,fieldsToReturn,null,null,0,
360. recordcount);
361. while(cursor.hasNext()){
362. //toMap()returnsaMap,butresult.add()expectsa
363. //Map<String,String>.Hence,thesuppresswarnings.
364. HashMap<String,ByteIterator>resultMap=newHashMap<String,ByteIterator>();
365. BSONObjectobj=cursor.getNext();
366. fillMap(resultMap,obj);
367. result.add(resultMap);
368. }
369. return0;
370. }catch(Exceptione){
371. System.err.println(e.toString());
372. e.printStackTrace();
373. return1;
374. }finally{
375. if(cursor!=null){
376. cursor.close();
377. }
378. }
379. }
380.  
381. @Override
382. publicintupdate(Stringtable,Stringkey,
383. HashMap<String,ByteIterator>values){
384. try{
385. DBCollectioncollection=getCollection(table);
386. if(collection==null){
387. System.out.println("Failedtogetcollection"+table);
388. return-1;
389. }
390.  
391. BSONObjectquery=newBasicBSONObject().append(keyfield,key);
392. BSONObjectupdate=newBasicBSONObject();
393. BSONObjectfieldsToSet=newBasicBSONObject();
394.  
395. Iterator<String>keys=values.keySet().iterator();
396. while(keys.hasNext()){
397. StringtmpKey=keys.next();
398. fieldsToSet.put(tmpKey,values.get(tmpKey).toArray());
399. }
400.  
401. update.put("$set",fieldsToSet);
402. collection.update(query,update,null);
403. return0;
404. }catch(Exceptione){
405. System.err.println(e.toString());
406. e.printStackTrace();
407. return1;
408. }
409. }
410.  
411. @Override
412. publicintinsert(Stringtable,Stringkey,
413. HashMap<String,ByteIterator>values){
414. try{
415. DBCollectioncollection=getCollection(table);
416. BSONObjectrecord=newBasicBSONObject().append(keyfield,key);
417. for(Stringk:values.keySet()){
418. record.put(k,values.get(k).toArray());
419. }
420.  
421. if(insertmode.equals(DEFAULT_INSERTMODE)){
422. collection.insert(record);
423. }
424. else{
425. if(objs.size()!=bulknum){
426. objs.add(record);
427. }
428.  
429. if(objs.size()==bulknum){
430. collection.bulkInsert(objs,DBCollection.FLG_INSERT_CONTONDUP);
431. objs.clear();
432. }
433. }
434. return0;
435. }catch(Exceptione){
436. System.err.println(e.toString());
437. e.printStackTrace();
438. return1;
439. }
440. }
441.  
442. @Override
443. publicintdelete(Stringtable,Stringkey){
444. try{
445. DBCollectioncollection=getCollection(table);
446. BSONObjectrecord=newBasicBSONObject().append(keyfield,key);
447. collection.delete(record);
448. return0;
449. }catch(Exceptione){
450. System.err.println(e.toString());
451. e.printStackTrace();
452. return1;
453. }
454. }
455.  
456. publicstaticvoidmain(String[]args){
457.  
458. Propertiesprops=newProperties();
459. props.setProperty("sequoiadb.host","192.168.30.63");
460. props.setProperty("sequoiadb.port","11810");
461. props.setProperty("sequoiadb.space","test");
462.  
463. SequoiaDBClientclient=newSequoiaDBClient();
464. client.setProperties(props);
465.  
466. try{
467. client.init();
468. client.getCollection("usertable");
469. }catch(DBExceptione){
470. e.printStackTrace();
471. }
472. }
473. }
474.