# [SOP]操作 TiDB 高性能写入 Flink 指南

近期遇到部分要通过 Flink 消费 Kafka，大量写入TiDB 的需求，但是写入性能不佳。通过诊断，原因大多是由于单条写入和并发不足造成。尝试优化效果不佳，难点在于各种条件下的参数组合及工作原理不确定。

经过一系列的测试，总结了一些优化实践，性能较优化前可提升10 倍以上。

**1. 实现高性能写入的必要条件：** 
批量插入 提高算子并行度，多线程并发写入 注意 ON DUPLICATE KEY UPDATE 语法，示例详见3.3

**2. 测试环境：** 

```yaml
Flink: V1.13.5
Scala: V2.12
MySQL JDBC Connector: V8.0.27
TiDB: V5.3.0
```

**3. 写入示例** 

**3.1 Flink中创建表：** 

```sql
CREATE TABLE order_item (
id BIGINT
, productId INT
, name STRING
, price DECIMAL(8,2)
, cnt INT
, PRIMARY KEY (id) NOT ENFORCED
)
WITH (
'connector' = 'jdbc',
'url' = 'jdbc:mysql://<ip>:4000/?useServerPrepStmts=true&cachePrepStmts=true&rewriteBatchedStatements=true',
'driver' = 'com.mysql.jdbc.Driver',
'table-name' = 't_order_item',
'username' = 'root',
'password' = '',
'sink.buffer-flush.max-rows' = '200',
'sink.buffer-flush.interval' = '1',
'sink.parallelism' = '200'
)
```

**创建表参数说明：** 

* PRIMARY KEY (id) NOT ENFORCED: 在插入数据的目标表中指定主键，将会自动生成 INSERT INTO …. ON DUPLICATE KEY UPDATE语句，实现 UPSERT

* JDBC连接字符串： - useServerPrepStmts=true: SQL 语句基本相同，需要避免服务器重复解析 SQL 的开销 - cachePrepStmts=true: 客户端缓存预处理语句 - rewriteBatchedStatements=true: 批量写入的关键参数，将单条 SQL 改写为多个 VALUES 值 SQL

* sink.parallelism' = '200’: 算子并行度，决定了写入 TiDB 时的线程数

* sink.buffer-flush.max-rows = '200': 每次提交的批大小，当缓冲区内行数达到该大小即触发提交

* sink.buffer-flush.interval = '1': 每次提交的时间间隔

**3.2 写入方式一：** 

```sql
val orderItemTable = tableEnv.sqlQuery("""
| SELECT
|     id
|   , productId
|   , name
|   , price
|   , cnt
| FROM
|   mock_orderitem
""".stripMargin)
orderItemTable.executeInsert("order_item")
```

Flink SQL 创建表中的配置，会以批量的方式写入，并将 INSERT INTO 自动转为 INSERT INTO …… ON DUPLICATE KEY UPDATE 语句。

**3.3 写入方式二：** 

```sql
tableEnv.executeSql(“””
| INSERT INTO
| order_item (
| id, productId, name, price, cnt
| )
| select id, productId, name, price, cnt
| FROM
| mock_orderitem)
```

以批量的方式写入，并将 INSERT INTO 自动转为 INSERT INTO …… ON DUPLICATE KEY UPDATE 语句。

**3.3 写入方式三：** 

使用最多的场景，调用JDBC方式，提供更高的灵活性，也无需在Flink中创建表(3.1)。

**需要特别注意 ON DUPLICATE KEY UPDATE 语法，必须使用**  **_field=values(field)_**  **形式，如写成 field=?，是无法转换成换量写入的。** 

```sql
val executionOptions = JdbcExecutionOptions.builder
.withBatchIntervalMs(1)
.withBatchSize(200)
.build
val connectionOptions = (new JdbcConnectionOptions.JdbcConnectionOptionsBuilder)   .withUrl("jdbc:mysql://<ip>:4000/?useServerPrepStmts=true&cachePrepStmts=true&rewriteBatchedStatements=true")
.withDriverName("com.mysql.jdbc.Driver")
.withUsername("root")
.withPassword("").build
var insertSQL = "INSERT INTO t_order_item (id, productId, name, price, cnt) values (?, ?, ?, ?, ?) ON DUPLICATE KEY UPDATE id=values(id)"
val sb: JdbcStatementBuilder[OrderItem] = new JdbcStatementBuilder[OrderItem] {
override def accept(ps: PreparedStatement, t: OrderItem): Unit = {
ps.setLong(1, t.id);
ps.setInt(2, t.productId);
ps.setString(3, t.name);
ps.setInt(4, t.price);
ps.setInt(5, t.cnt);
}
}
val mySink = JdbcSink.sink(insertSQL, sb, executionOptions, connectionOptions)
sourceDataStream.addSink(mySink).setParallelism(200)
env.execute("TiDB Batch Write Job")
```