百萬級資料遷移方案測評小記
阿新 • • 發佈：2020-06-29
### 前言
最近公司在使用 `ABP` 重構之前的老專案，資料庫也由 `SQL SERVER` 切換到了 `MySql`。吐槽一下，之前的產品使用的是 `Windows Server 2008` , `SqlServer 2008R2`, `.Net Framework 4.5`，現在開始擁抱 `.net core`。回到正題。目前單表有 `10w+`，`100w+` 資料不等，等會都測試一下。資料庫切換，以及資料庫表結構變化，不可以避免的需要進行資料遷移。而遷移方案也並不是很多，下面是我嘗試使用的兩種方案進行測試。
### 多執行緒批量寫入
```
private static async Task BatchInsertTestUsers(List testUsers)
{
var prefix =
"INSERT INTO users (Id,Name,Age) VALUES";
using (IDbConnection conn = new MySqlConnection(DataMigrationConfig.MySqlConstr))
{
var sqlText = new StringBuilder();
sqlText.Append(prefix);
foreach (var testUser in testUsers)
{
sqlText.AppendFormat(
$"({testUser.Id},'{testUser.Name}', {testUser.Age}),");
}
var insertSql = sqlText.ToString().Substring(0, sqlText.ToString().LastIndexOf(','));
await conn.ExecuteAsync(insertSql);
}
}
```
* `BatchInsertTestUsers` 將傳入的集合，拼接成 `SQL` 並執行。
```
public static Task RunMultiTasks(List users)
{
var tasks = new List();
var pageSize = 10000;
var writeCount = (users.Count() / pageSize) + 2;
for (var i = 1; i < writeCount; i++)
{
var skipCount = (i - 1) * pageSize;
var batchInsertList = users.Skip(skipCount).Take(pageSize).ToList();
var task = Task.Run(() => { BatchInsertTestUsers(batchInsertList); });
tasks.Add(task);
}
var sw = new Stopwatch();
sw.Start();
Task.WaitAll(tasks.ToArray());
sw.Stop();
Console.WriteLine($"多執行緒批量插入用時：{sw.ElapsedMilliseconds} ms");
return Task.FromResult(0);
}
```
* `RunMultiTasks` 將資料分批，一次性插入 `1w` 條。
### MySqlBulkLoader 方案
瞭解到 `MySqlBulkLoader` 是因為 `SqlServer` 的 `Sqlbulkcopy`。`MySqlBulkLoader` 並不支援集合的匯入，需要先將資料匯出為 `.csv` 格式，然後讀取 `.csv` 資料匯入。
```
public static async Task Export(string filePath, List items)
{
IExporter exporter = new CsvExporter();
await exporter.Export(filePath, items);
}
```
* 這裡資料匯出使用國人開源的 [dotnetcore/Magicodes.IE](https://github.com/dotnetcore/Magicodes.IE) 我這個匯出程式碼，應該就懂了吧！操作簡潔！！！
```
public static void Load(string filePath, string tableName)
{
using MySqlConnection conn = new MySqlConnection(DataMigrationConfig.MySqlConstr);
var bulk = new MySqlBulkLoader(conn)
{
FieldTerminator = ",",
FieldQuotationCharacter = '"',
EscapeCharacter = '"',
LineTerminator = "\r\n",
FileName = filePath,
Local = true,
NumberOfLinesToSkip = 1,
TableName = tableName,
CharacterSet = "utf8mb4",
};
bulk.Load();
}
```
* 這裡因為資料庫並不在自己本機上，所以設定了 `Local = true` 讀取本地檔案，進行匯入。
### 測試說明
* 這個測試是在我本地測試的，資料庫是跑在內網部署的一臺機器上的 `Docker` 容器內，用的是機械硬碟。如果您的使用的是 `SSD` 硬碟，效果會更佳。
* 這裡測試主要是插入簡單的使用者資料，定義如下：
```
public class TestUser
{
public int Id { get; set; }
public string Name { get; set; }
public int Age { get; set; }
}
```
* 分別測試`1w`，`10w`,`100w`條資料插入的效能，以及開啟索引以及關閉索引的影響
* 測試執行程式碼如下：
```
class Program
{
static async Task Main(string[] args)
{
var testData = DataGen.Run(100 * 10000);
await RunMultiTasks(testData);
await RunMySqlLoaderTask(testData);
}
public static async Task RunMultiTasks(List users)
{
await DataMigrateTask.RunMultiTasks(users);
}
public static async Task RunMySqlLoaderTask(List users)
{
var fileName = "users";
var filePath = Directory.GetCurrentDirectory() + "\\" + fileName + ".csv";
await DataMigrateTask.Export(filePath, users);
var sw = new Stopwatch();
sw.Start();
DataMigrateTask.Load(filePath, "users");
sw.Stop();
Console.WriteLine($"MySqlBulkLoader 用時：{sw.ElapsedMilliseconds} ms");
}
}
```
### 測試結果
說了那麼多，這裡才是最重點。
| 方案 | 1w | 10w | 100w |
| ---- | ---- | ---- | ---- |
| RunMultiTasks | 367ms | 3548ms | 91263ms |
| RunMySqlLoaderTask | 2031ms | 1597ms | 13105ms |
| RunMultiTasks（關閉索引） | 233ms | 3230ms | 67040ms |
| RunMySqlLoaderTask （關閉索引） | 1785ms | 1367ms | 12456ms |
### 最後
以上的測試僅供參考，上面的簡單測試一下，資料量大的時候 `MySqlLoaderTask` 優勢是明顯的，對於小於 `1w` 資料量的可以採用多執行緒批量插入效果更好。有興趣的小夥伴的可以自己下載程式碼玩玩。如有更好的
方案，不吝賜教。
* 程式碼地址：[DataMigrationTest](https://github.com/timy7/DataMigrationTest)
### 坑
* `MySqlLoader` 匯入 `null` 資料使用 `NULL`，而不是`mysql`文件上說的 `\N`