用JFreeChart 来分析Cassandra/Oracle插入海量数据的性能

为了分析在插入海量数据到Cassandra集群或者Oracle时的表现，也就是插入速率，我们用java程序对插入数据的用时进行了采样，最终用JFreeChart把采样结果绘制出来了。

为了公平起见，我们做了以下处理：

1.所有的循环变量都放在了循环外面

2.对于Cassandra的replication-factor设置为1，这样插入数据不需要插入额外的备份。

3.对于Oracle我们用预编译语句，这样插入操作的执行计划可以重用。

4.所有的测试都在周末进行，这样不可能有其他人去干扰这些服务器。

5.这些机器上运行的其他进程都被我kill掉了，这样保证CPU，内存的专用性。

6.在用java代码插入Cassandra记录时候，我采用了thrift API, 因为它的效率比Hector API高。

以下是实验（分两部分，一是采样部分，二是数据分析部分）

Part 1:采样：

Cassandra的采样：

我们这里依然用循环插入50W条记录，不同的是，在循环的开始和循环每10000条记录时，我们把时间戳记录在List中，最终把这个List写入文本文件(cassandra_input_sample_data.txt)：

1. package com.charles.cassandra.demo; 
2.  
3. import java.io.File; 
4. import java.io.FileWriter; 
5.  
6. import java.util.ArrayList; 
7. import java.util.List; 
8.  
9. import org.apache.cassandra.thrift.Cassandra; 
10. import org.apache.cassandra.thrift.Column; 
11.  
12. import org.apache.cassandra.thrift.ColumnParent; 
13. import org.apache.cassandra.thrift.ConsistencyLevel; 
14.  
15. import org.apache.cassandra.thrift.TBinaryProtocol; 
16.  
17. import org.apache.thrift.protocol.TProtocol; 
18. import org.apache.thrift.transport.TFramedTransport; 
19. import org.apache.thrift.transport.TSocket; 
20. import org.apache.thrift.transport.TTransport; 
21.  
22. import com.charles.cassandra.util.CassandraOperationUtil; 
23.  
24. public class CassandraClusterStressTest 
25. { 
26.     public static void main(String[] args) 
27.         throws Exception 
28.     { 
29.         //包装好的socket 
30.         TTransport tr = new TFramedTransport(new TSocket("192.168.129.34",9160)); 
31.         TProtocol proto = new TBinaryProtocol(tr); 
32.         Cassandra.Client client = new Cassandra.Client(proto); 
33.         tr.open(); 
34.          
35.         if(!tr.isOpen()) 
36.         { 
37.             System.out.println("无法连接到服务器!"); 
38.             return; 
39.         } 
40.          
41.         System.out.println("开始压力测试，我们插入50W条数据到2节点集群中"); 
42.         System.out.println("..."); 
43.          
44.         //标记开始时间 
45.         long startTime = System.currentTimeMillis(); 
46.          
47.         client.set_keyspace("Charles_Stress_Test2");//使用Charles_Stress_Test keyspace 
48.         ColumnParent parent = new ColumnParent("student");//column family 
49.          
50.         /* 
51.          * 这里我们插入50万条数据到Student内 
52.          * 每条数据包括id和name 
53.          */ 
54.         String key_user_id = "a"; 
55.         String k; 
56.         long timestamp; 
57.         Column idColumn =null; 
58.         Column nameColumn=null; 
59.          
60.         //这个sampleData代表了每插入1W条记录到Cassandra集群的用时毫秒的数据样本 
61.         List<Integer> sampleData = new ArrayList<Integer>(51); 
62.         for(int i = 0;i < 500000;i++) 
63.         { 
64.             k = key_user_id + i;//row key 
65.             timestamp = System.currentTimeMillis();//时间戳 
66.              
67.             //每行的第一个字段（id字段） 
68.             idColumn = new Column(CassandraOperationUtil.stringToByteBuffer("id"));//字段名 
69.             idColumn.setValue(CassandraOperationUtil.stringToByteBuffer(i + ""));//字段值 
70.             idColumn.setTimestamp(timestamp);//时间戳 
71.             client.insert( 
72.                 CassandraOperationUtil.stringToByteBuffer(k),  
73.                 parent,  
74.                 idColumn,  
75.                 ConsistencyLevel.ONE); 
76.              
77.             //每行的第二个字段(name字段） 
78.             nameColumn = new Column(CassandraOperationUtil.stringToByteBuffer("name")); 
79.             nameColumn.setValue(CassandraOperationUtil.stringToByteBuffer("student" + i)); 
80.             nameColumn.setTimestamp(timestamp); 
81.             client.insert( 
82.                 CassandraOperationUtil.stringToByteBuffer(k),  
83.                 parent,  
84.                 nameColumn,  
85.                 ConsistencyLevel.ONE); 
86.              
87.             //判断是否这是起始记录（用于标记起始时间戳）和第N 万条记录（第N万条记录的时间戳） 
88.             if( (i==0) || ( (i+1)%10000==0)){ 
89.                 sampleData.add((int)(timestamp)); 
90.             } 
91.         } 
92.        
93.         //标记结束时间 
94.         long endTime = System.currentTimeMillis(); 
95.         //标记一共用时 
96.         long elapsedTime = endTime-startTime; 
97.          
98.         System.out.println("压力测试完毕,用时: "+elapsedTime+" 毫秒"); 
99.         
100.         //关闭连接 
101.         tr.close(); 
102.          
103.         //压力测试结束后，我们把所有的样本数据写入文件中等待处理 
104.         FileWriter fw = new FileWriter(new File("cassandra_insert_sample_data.txt")); 
105.         for(int j=0;j<sampleData.size();j++){ 
106.             fw.write(sampleData.get(j)+"\n"); 
107.         } 
108.         fw.flush(); 
109.         fw.close(); 
110.     } 
111.      
112.     
113. } 

最终50W条记录插入完毕：控制台显示：

而且我们打开文本文件确定这些时间戳的样本都被记录了：

当然了，因为Cassandra的存储是基于内存的，所以我们定义了一个工具类用于转换字符串和字节数组：

1. /*  
2.  */ 
3. package com.charles.cassandra.util; 
4.  
5. import java.io.UnsupportedEncodingException; 
6. import java.nio.ByteBuffer; 
7.  
8. /** 
9.  * 
10.  * Description: 这个类提供了一些Cassandra操作的工具类 
11.  * 
12.  * @author charles.wang 
13.  * @created May 19, 2012 11:18:27 AM 
14.  *  
15.  */ 
16. public class CassandraOperationUtil { 
17.      
18.     /** 
19.      *因为在Cassandra中，信息都存在内存的，所以都是以ByteBuffer形式存储的，但是ByteBuffer对于人类来说没有String可读性强 
20.      *所以这个方法可以吧字符串转为ByteBuffer 
21.      */ 
22.     public static ByteBuffer stringToByteBuffer(String s) throws UnsupportedEncodingException{ 
23.          
24.         return ByteBuffer.wrap(s.getBytes("UTF-8")); 
25.     } 
26.      
27.     /** 
28.      *因为在Cassandra中，信息都存在内存的，所以都是以ByteBuffer形式存储的，但是ByteBuffer对于人类来说没有String可读性强 
29.      *所以对称的，这个方法吧ByteBuffer转为人类可读的字符串 
30.      */ 
31.     public static String byteBufferToString (ByteBuffer b) throws UnsupportedEncodingException{ 
32.          
33.         //先构建一个字节数组 
34.         byte[] bytes = new byte[b.remaining()]; 
35.         //吧bytebuffer里面的内容全部存入字节数组 
36.         b.get(bytes); 
37.         //然后把这些bytes转为String 
38.         return new String(bytes,"UTF-8"); 
39.     } 
40.  
41. } 

Oracle的采样：

我们这里依然用循环插入50W条记录，不同的是，在循环的开始和循环每10000条记录时，我们把时间戳记录在List中，最终把这个List写入文本文件(oracle_input_sample_data.txt)：

1. /*  
2.  */ 
3. package com.charles.cassandra.demo; 
4.  
5.  
6. import java.io.File; 
7. import java.io.FileWriter; 
8. import java.sql.Connection; 
9. import java.sql.Date; 
10. import java.sql.DriverManager; 
11. import java.sql.PreparedStatement; 
12. import java.sql.ResultSet; 
13. import java.sql.Statement; 
14. import java.sql.ResultSetMetaData; 
15. import java.sql.Timestamp; 
16. import java.util.ArrayList; 
17. import java.util.List; 
18.  
19.  
20. /** 
21.  * 
22.  * Description:插入50W条记录到关系数据库Oracle中 
23.  * 
24.  * @author charles.wang 
25.  * @created May 19, 2012 5:25:36 PM 
26.  *  
27.  */ 
28.  
29. public class OracleStressTest { 
30.  
31.   
32.      
33.      
34.      
35.     /** 
36.      * 既然要测负载，就尽可能减少方法调用的时间开销，所以我用了最原始的写法 
37.      * @param args 
38.      */ 
39.      
40.     public static void main(String[] args){ 
41.          
42.         String url="jdbc:oracle:thin:@192.168.129.14:15210:ora11g"; 
43.         String username="Charles_Stress_Test1"; 
44.         String password="Charles_Stress_Test1"; 
45.          
46.         String sDBDriver = "oracle.jdbc.driver.OracleDriver"; 
47.  
48.         try{ 
49.              
50.             System.out.println("开始压力测试，我们以预编译的方式插入50W条数据到Oracle中"); 
51.             System.out.println("..."); 
52.             //标记开始时间 
53.             long startTime=System.currentTimeMillis(); 
54.                      
55.                      
56.             Class.forName(sDBDriver).newInstance(); 
57.             Connection conn = DriverManager.getConnection(url,username,password); 
58.              
59.             //因为这里使用预编译语句，所以不用每次都生成新的执行计划 
60.              
61.             String rowkey=null; 
62.             String id=null; 
63.             String name=null; 
64.             Date date=null; 
65.             String statementString="insert into Student (rowkey,id,name,create_date )values(?,?,?,?)";; 
66.             
67.             PreparedStatement pstmt = conn.prepareStatement(statementString); 
68.              
69.             //这个sampleData代表了每插入1W条记录到Oracle数据库用时毫秒的数据样本 
70.             List<Integer> sampleData = new ArrayList<Integer>(51); 
71.              
72.             for(int i=0;i<500000;i++){ 
73.                  
74.                 long timestamp = System.currentTimeMillis(); 
75.                 rowkey="a"+i; 
76.                 id=""+i; 
77.                 name="student"+i; 
78.                 date= new Date(timestamp); 
79.                   
80.                 pstmt.setString(1,rowkey); 
81.                 pstmt.setString(2, id); 
82.                 pstmt.setString(3,name); 
83.                 pstmt.setDate(4, date); 
84.                 pstmt.execute(); 
85.                  
86.                 //判断是否这是起始记录（用于标记起始时间戳）和第N 万条记录（第N万条记录的时间戳） 
87.                 if( (i==0) || ( (i+1)%10000==0)){ 
88.                     sampleData.add((int)(timestamp)); 
89.                 } 
90.             } 
91.              
92.             //关闭相关连接 
93.             pstmt.close(); 
94.             conn.close(); 
95.              
96.             long endTime=System.currentTimeMillis(); 
97.             long elapsedTime=endTime-startTime; 
98.              
99.             System.out.println("压力测试完毕,用时: "+elapsedTime+" 毫秒"); 
100.              
101.             //在压力测试结束之后，我们来把样本数据写入文本文件中 
102.             FileWriter fw = new FileWriter(new File("oracle_insert_sample_data.txt")); 
103.             for(int j=0;j<sampleData.size();j++){ 
104.                 fw.write(sampleData.get(j)+"\n"); 
105.             } 
106.             fw.flush(); 
107.             fw.close(); 
108.              
109.         }catch(Exception e){ 
110.             System.out.println("数据库连接失败"); 
111.             e.printStackTrace(); 
112.         } 
113.          
114.         
115.         
116.     } 
117.      
118.  
119. } 

最终50W条记录插入完毕：控制台显示：

而且我们打开文本文件确定这些时间戳的样本都被记录了：

Part 2： 分析采样数据并且绘制比较图：

我们用JFreechart强大的图表制作能力来绘制比较图：

首先我们依然定义一个工具类 ParseDataUtil，它可以完成两件事情，一是从样本文件中读取数据，然后时间戳相减，最终把所有每1W条数据的耗时时间存入List<Integer>对象，二是它可以吧List<Integer>对象传递给JFreechart的数据模型：

1. /*  
2.  */ 
3. package com.charles.parsedata.util; 
4.  
5. import java.io.BufferedReader; 
6. import java.io.File; 
7. import java.io.FileInputStream; 
8. import java.io.FileNotFoundException; 
9. import java.io.FileReader; 
10. import java.io.IOException; 
11. import java.io.InputStreamReader; 
12. import java.util.ArrayList; 
13. import java.util.List; 
14.  
15. import org.jfree.data.category.DefaultCategoryDataset; 
16.  
17. /** 
18.  *  
19.  * Description: 
20.  *  
21.  * @author charles.wang 
22.  * @created May 21, 2012 8:45:28 AM 
23.  *  
24.  */ 
25. public class ParseDataUtil { 
26.  
27.     /** 
28.      * 这个方法用于添加指定的分析来的数据作为JFreechart显示的数据集 
29.      *  
30.      * @param ds 
31.      *            JFreechart的数据集对象 
32.      * @param datas 
33.      *            从压力测试采样并且经过加工后的数据 
34.      * @param seriesName 
35.      *            曲线的名称 
36.      */ 
37.     public static void addDataToDataset(DefaultCategoryDataset ds, List<Integer> datas, String seriesName) { 
38.         // 对于数据集合的检查 
39.         if (datas.size() <= 0) 
40.             return; 
41.  
42.         // type表示横轴的每个坐标点 
43.         Integer value = 0; 
44.         String type = null; 
45.  
46.         // 用循环依次添加 
47.         for (int i = 1; i <= datas.size(); i++) { 
48.             // 获取每个样本数据中的横坐标纵坐标 
49.             type = i + ""; 
50.             value = datas.get(i - 1); 
51.             ds.addValue(value, seriesName, type); 
52.         } 
53.  
54.     } 
55.  
56.     /** 
57.      * 这个方法用于从样本数据中构建最终传入到JFreechart绘制的数据 
58.      *  
59.      * @param fileName 
60.      * @param numOfRecords 
61.      * @return 
62.      */ 
63.     public static List<Integer> buildSampleDataListFromFile(String fileName, int numOfRecords) { 
64.  
65.         // 判断参数 
66.         if (numOfRecords <= 0) 
67.             return null; 
68.  
69.         try { 
70.  
71.             // 创建一个rawSampleData 作为采样数据的List 
72.             List<Integer> rawSampleData = new ArrayList<Integer>(numOfRecords); 
73.  
74.             // 打开一个到指定文件的输入流 
75.             FileInputStream fis = new FileInputStream(fileName); 
76.             InputStreamReader isr = new InputStreamReader(fis); 
77.             BufferedReader br = new BufferedReader(isr); 
78.             if (br == null) { 
79.                 System.out.println("样本文件不存在！"); 
80.                 return null; 
81.             } 
82.             // 依次读入 
83.             for (int i = 0; i < numOfRecords; i++) { 
84.                 String rawRecord = br.readLine(); 
85.                 rawSampleData.add(Integer.parseInt(rawRecord)); 
86.             } 
87.  
88.             // 读完了关闭输入流 
89.             br.close(); 
90.             isr.close(); 
91.             fis.close(); 
92.  
93.             // 现在我们把rawSampleData转为真正可以被JFreeChart显示的SampleData 
94.             // 这里SampleData的每个数据都是用时，所以是当前时间戳-第一条记录的时间戳 
95.             int sampleDataSize = rawSampleData.size() - 1; 
96.             List<Integer> sampleData = new ArrayList<Integer>(sampleDataSize); 
97.             // 设置起始时间戳，以后每一个时间戳都要减去这个起始时间戳 
98.             Integer baseTimeStamp = rawSampleData.get(0); 
99.             // System.out.println("baseTimeStamp: "+baseTimeStamp); 
100.             // System.out.println("sampleDataSize: "+sampleData.size()); 
101.  
102.             // System.out.println("hello"); 
103.             for (int j = 0; j < sampleDataSize; j++) { 
104.                 int time = rawSampleData.get(j + 1) - baseTimeStamp; 
105.                 System.out.println(time); 
106.                 sampleData.add(time); 
107.             } 
108.  
109.             return sampleData; 
110.  
111.         } catch (Exception ex) { 
112.             ex.printStackTrace(); 
113.             return null; 
114.         } 
115.  
116.     } 
117.  
118. } 

然后我们有个最终执行画图的类，这个类吧从原始数据分析后的数据显示在图表上，并且作为对比，吧Oracle和Cassandra集群的数据显示在同一张表上：

1. /*  
2.  */ 
3. package com.charles.parsedata; 
4.  
5. import java.util.ArrayList; 
6. import java.util.List; 
7.  
8. import javax.swing.JPanel; 
9.  
10. import org.jfree.chart.ChartFactory; 
11. import org.jfree.chart.ChartPanel; 
12. import org.jfree.chart.JFreeChart; 
13. import org.jfree.chart.axis.NumberAxis; 
14. import org.jfree.chart.plot.CategoryPlot; 
15. import org.jfree.chart.plot.PlotOrientation; 
16. import org.jfree.data.category.DefaultCategoryDataset; 
17. import org.jfree.ui.ApplicationFrame; 
18. import org.jfree.ui.RefineryUtilities; 
19.  
20. import com.charles.parsedata.util.ParseDataUtil; 
21.  
22.  
23. /** 
24.  * 
25.  * Description: 用JFreechart来分析插入数据 
26.  * 
27.  * @author charles.wang 
28.  * @created May 21, 2012 8:38:27 AM 
29.  *  
30.  */ 
31. public class InsertDataStressTestDataParser extends ApplicationFrame{ 
32.  
33.  
34.  
35.  
36.  
37. public InsertDataStressTestDataParser(String s) { 
38.     super(s); 
39.     setContentPane(createDemoLine()); 
40.    }  
41.  
42.    public static void main(String[] args)  { 
43.     InsertDataStressTestDataParser fjc = new InsertDataStressTestDataParser("Cassandra&Oracle插入数据对比图"); 
44.     fjc.pack(); 
45.     RefineryUtilities.centerFrameOnScreen(fjc); 
46.     fjc.setVisible(true);  
47.  
48.    }  
49.  
50.    // 生成显示图表的面板 
51.    public static JPanel createDemoLine(){ 
52.     JFreeChart jfreechart = createChart(createDataset()); 
53.     return new ChartPanel(jfreechart); 
54.    }  
55.  
56.    // 生成图表主对象JFreeChart 
57.    public static JFreeChart createChart(DefaultCategoryDataset linedataset) { 
58.     //定义图表对象 
59.     JFreeChart chart = ChartFactory.createLineChart("Cassandra和Oracle插入数据对比图", // chart title 
60.       "记录数(单位：万条）", // 横轴标签 
61.       "用时(毫秒)", // 纵轴标签 
62.       linedataset, // 传入的数据集 
63.       PlotOrientation.VERTICAL, // 方向 
64.       true, // bool变量表示是否要加入图例(legend) 
65.       true, // 工具集 
66.       false // 是否添加url 
67.       ); 
68.     CategoryPlot plot = chart.getCategoryPlot(); 
69.     // 范围轴线 
70.     NumberAxis rangeAxis = (NumberAxis) plot.getRangeAxis(); 
71.     rangeAxis.setStandardTickUnits(NumberAxis.createIntegerTickUnits()); 
72.     rangeAxis.setAutoRangeIncludesZero(true); 
73.     rangeAxis.setUpperMargin(0.20); 
74.     rangeAxis.setLabelAngle(Math.PI / 2.0);  
75.  
76.     return chart; 
77.    }  
78.  
79.    //生成数据 
80.    public static DefaultCategoryDataset createDataset() { 
81.     DefaultCategoryDataset ds = new DefaultCategoryDataset(); 
82.  
83.      
84.     List<Integer> data1 = ParseDataUtil.buildSampleDataListFromFile("cassandra_insert_sample_data.txt",51); 
85.     List<Integer> data2 = ParseDataUtil.buildSampleDataListFromFile("oracle_insert_sample_data.txt",51); 
86.     ParseDataUtil.addDataToDataset(ds, data1, "Cassandra插入数据所用时间分布图"); 
87.     ParseDataUtil.addDataToDataset(ds, data2, "Oracle插入数据所用时间分布图"); 
88.  
89.  
90.     return ds; 
91.    }  
92.     
93.     
94. } 

最终对比图如下：

结论：

所以我们这里很清楚的看到：

(1) 无论是Cassandra集群还是Oracle，其插入操作用时都是线性的，也就是它的平均插入速率基本是恒速。

(2) 在低配置服务器上，Cassandra集群的插入数据操作耗时要高于Oracle关系数据库。