如何实时写入优化?

Insert 语句最优写法


Insert语句标准语法如下：

1. INSERT [IGNORE]
2. [INTO] tbl_name
3. [(col_name,...)]
4. {VALUES | VALUE}

• 采用批量写入（batch insert）模式，即每次在VALUES部分添加多行数据，一般建议每次批量写入数据量大约为16KB，以提高网络和磁盘吞吐。
• 如果对一行的所有列都进行插入，则去除col_name并保证values顺序与表结构中的col_name顺序一致，以降低网络带宽耗用。
• 保持主键相对有序。AnalyticDB的insert语句要求必须提供主键，且主键可以为复合主键。当确定复合主键时，根据业务含义调整复合主键中各个列的次序，从业务层面保证插入时主键是严格递增或近似递增的，也可以提升实时写入速度。
• 增加ignore关键字。执行不带ignore关键字的insertsql，当主键冲突时，后续数据会覆盖之前插入的数据；带上ignore关键字，则主键冲突时，会保留之前插入的数据而自动忽略新数据。如果业务层没有数据覆盖的语义要求，则建议所有insertsql都加上ignore关键字，以减小覆盖数据带来的性能开销。

1. import java.util.ArrayList;
2. import java.util.List;
3. import java.util.zip.CRC32;
4. import java.util.zip.Checksum;
6. public class BatchInsertExample {
7.     // 一级分区个数
8.     private int partitionCount;
9.     // 分区列id，分区列在建表语句中序号
10.     private int partitionColumnIndex;
11.     // 总列数
12.     private int columnCount;
13.     // batch insert 条数
14.     private int batchSize;
15.     private List<List<Object[]>> partitionGroup;
16.     private String tableName;
17.     private boolean forTest = true;
19.     public BatchInsertExample(String tableName, int partitionCount, int columnCount, int partitionColumnIndex, int batchSize) {
20.         this.columnCount = columnCount;
21.         this.tableName = tableName;
22.         this.partitionColumnIndex = partitionColumnIndex;
23.         this.batchSize = batchSize;
24.         this.partitionCount = partitionCount;
25.         partitionGroup = new ArrayList<List<Object[]>>(partitionCount);
26.         for(int i = 0; i < partitionCount; i++) {
27.             List<Object[]> listi = new ArrayList<Object[]>(batchSize);
28.             partitionGroup.add(listi);
29.         }
30.     }
32.     /**
33.      * Function role, batch insert recordCount records, all records are strings except null.
34.      * @param records 的length 建议为 columnCount X batchSize。
35.      */
36.     public void insert(Object[][] records, int recordCount) {
37.         for(int i = 0; i < recordCount; i++) {
38.             int partitionNum = getPartitionNum(records);
39.             List<Object[]> batchRecord = partitionGroup.get(partitionNum);
40.             batchRecord.add(records);
41.             if (batchRecord.size() >= batchSize) {
42.                 String sql = generateInsertSql(batchRecord);
43.                 executeInsert(sql);
44.                 batchRecord.clear();
45.             }
46.         }
48.         for(int pn = 0; pn < partitionCount; pn++) {
49.             List<Object[]> batchRecord = partitionGroup.get(pn);
50.             if (batchRecord.size() > 0) {
51.                 String sql = generateInsertSql(batchRecord);
52.                 executeInsert(sql);
53.                 batchRecord.clear();
54.             }
55.         }        
56.     }
58.     /**
59.      *
60.      * @param batchRecord
61.      * @return insert sql with multiple values,
62.      * like insert into zs_test values(NULL,NULL,NULL),(7,'7',7),(20,'20',20),(48,'48',48),(57,'57',57),(60,'60',60),(95,'95',95),(97,'97',97),(100,'100',100),(102,'102',102)
63.      */
64.     private String generateInsertSql(List<Object[]> batchRecord) {
65.         StringBuffer sb = new StringBuffer("insert into ").append(tableName).append(" values");
66.         for(int i = 0; i < batchRecord.size(); i++) {
67.             if (i > 0) {
68.                 sb.append(",");
69.             }
70.             sb.append("(");
71.             Object[] objs = batchRecord.get(i);
72.             assert(objs.length == columnCount);
73.             for(int j = 0; j < objs.length; j++) {
74.                 if (j > 0) {
75.                     sb.append(",");
76.                 }
77.                 if (objs[j] == null) {
78.                     sb.append("NULL");
79.                 } else {
80.                     sb.append(objs[j].toString());
81.                 }
82.             }
83.             sb.append(")");
84.         }
85.         return sb.toString();
86.     }
88.     /**
89.      *
90.      * @param record
91.      * @return calculte partition num basing on partition column value
92.      */
93.     private int getPartitionNum(Object[] record) {
94.         Object value = record[partitionColumnIndex];
95.         long crc32 = (value == null ? getCRC32("-1") : getCRC32(value.toString()));
96.         return (int) (crc32 % this.partitionCount);
97.     }
99.     private final long getCRC32(String value) {
100.         Checksum checksum = new CRC32();
101.         byte[] bytes = value.getBytes();
102.         checksum.update(bytes, 0, bytes.length);
103.         return checksum.getValue();
104.     }
106.     /**
107.      * user should implement it with mysql protocol.
108.      * @param insertSql
109.      */
110.     public void executeInsert(String insertSql) {
111.         if (forTest) {
112.             System.out.println(insertSql);
113.         } else {
114.             throw new RuntimeException("user should implement it!!!!!");
115.         }
116.     }
118.     public static void main(String args[]) {
119.         String tableName="zs_test";
121.         int partitionCount = 12;
122.         int batchSize = 10;
123.         int columnCount = 3;
124.         int partitionColunIndex = 0;
126.         BatchInsertExample insert = new BatchInsertExample(tableName, partitionCount, columnCount, partitionColunIndex, batchSize);
128.         // buffersize should always be same with batchsize * partition-count
129.         int bufferSize = partitionCount * batchSize;
130.         int recordCount = 0;
133.         Object objs[][] = new Object[bufferSize][];
134.         int totalRecord = 1000;
135.         for(int i = 0; i < totalRecord; i++) {
136.             objs[recordCount] = i == 0 ? new Object[3]: new Object[]{i, "'" + i + "'", i};
137.             recordCount++;
138.             if (recordCount == bufferSize || (i == totalRecord - 1)) {
139.                 insert.insert(objs, recordCount);
140.                 recordCount = 0;
141.             }
142.         }
143.     }
144. }

1. optimize table <table_name>