菜菜呀，咱们业务 BJKJ 有个表数据需要做迁移
更多精彩内容请看 web 前端中文站
http://www.lisa33xiaoq.net 可按 Ctrl + D 进行收藏
现在有多少数据？
大约 21 亿吧，2017 年以前的数据没有业务意义了，给你半天时间把这个事搞定，绩效给你 A
有绩效奖金吗？
钱的事你去问 X 总，我当家不管钱
………..
经过几分钟的排查，数据库情况如下：
至于业务，不是太复杂。经过相关人员咨询，大约 40%的请求为单条 Insert，大约 60%的请求为按 class_id 和 in_time（倒序）分页获取数据。Select 请求全部命中聚集索引，所以性能非常高。这也是聚集索引之所以这样设计的目的。
由于单表数据量已经超过 21 亿，并且 2017 年以前的数据几乎不影响业务，所以决定把 2017 年以前（不包括 2017 年）的数据迁移到新表，仅供以后特殊业务查询使用。经过查询大约有 9 亿数据量。
数据迁移工作包括三个个步骤：
1. 从源数据表查询出要迁移的数据
2. 把数据插入新表
3. 把旧表的数据删除
传统做法
这里申明一点，就算是传统的做法也需要分页获取源数据，因为你的内存一次性装载不下 9 亿条数据。
SQL 语句类似于
SELECT * FROM (
SELECT *,ROW_NUMBER() OVER(ORDER BY class_id,in_time) p FROM tablexx WHERE in_time <'2017.1.1'
) t WHERE t.p BETWEEN 1 AND 100
如果你的数据量不大，以上方法完全没有问题，但是在 9 亿这个数字前面，以上方法显得心有余而力不足。一个字：慢，太慢，非常慢。
可以大体算一下，假如每秒可以迁移 1000 条数据，大约需要的时间为(单位：分)
大约需要 10 天^ V ^
以上的传统做法弊端在哪里呢？
1. 在 9 亿数据前查询必须命中索引，就算是非聚集索引菜菜也不推荐，首推聚集索引。
2. 如果你了解索引的原理，你应该明白，不停的插入新数据的时候，索引在不停的更新，调整，以保持树的平衡等特性。尤其是聚集索引影响甚大，因为还需要移动实际的数据。
提取以上两点共同的要素，那就是聚集索引。相应的解决方案也就应运而生：
1. 按照聚集索引分页查询数据
2. 批量插入数据迎合聚集索引，即：按照聚集索引的顺序批量插入。
3. 按照聚集索引顺序批量删除
由于做了表分区，如果有一种方式把 2017 年以前的分区直接在磁盘物理层面从当前表剥离，然后挂载到另外一个表，可算是神级操作。有谁能指导一下菜菜，感激不尽
1. 第一步：由于聚集索引需要 class_id ，所以宁可花 2-4 秒时间把要操作的 class_id 查询出来(ORM 为 dapper)，并且升序排列
DateTime dtMax = DateTime.Parse("2017.1.1");
var allClassId = DBProxy.GeSourcetLstClassId(dtMax)?.OrderBy(s=>s);2. 按照第一步 class_id 列表顺序查询数据，每个 class_id 分页获取，然后插入目标表，全部完成然后删除源表相应 class_id 的数据。(全部命中聚集索引)
D int pageIndex = 1; //页码
int pageCount = 20000;//每页的数据条数
DataTable tempData =null;
int successCount = 0;
foreach (var classId in allClassId)
{
tempData = null;
pageIndex = 1;
while (true)
{
int startIndex = (pageIndex - 1) * pageCount+1;
int endIndex = pageIndex * pageCount;
tempData = DBProxy.GetSourceDataByClassIdTable(dtMax, classId, startIndex, endIndex);
if (tempData == null || tempData.Rows.Count==0)
{
//最后一页无数据了，删除源数据源数据然后跳出
DBProxy.DeleteSourceClassData(dtMax, classId);
break;
}
else
{
DBProxy.AddTargetData(tempData);
}
pageIndex++;
}
successCount++;
Console.WriteLine($"班级:{classId} 完成,已经完成：{successCount}个");
}DBProxy 完整代码：
class DBProxy
{
//获取要迁移的数据所有班级 id
public static IEnumerable<int> GeSourcetLstClassId(DateTime dtMax)
{
var connection = Config.GetConnection(Config.SourceDBStr);
string Sql = @"SELECT class_id FROM tablexx WHERE in_time <@dtMax GROUP BY class_id ";
using (connection)
{
return connection.Query<int>(Sql, new { dtMax = dtMax }, commandType: System.Data.CommandType.Text);
}
}
public static DataTable GetSourceDataByClassIdTable(DateTime dtMax, int classId, int startIndex, int endIndex)
{
var connection = Config.GetConnection(Config.SourceDBStr);
string Sql = @" SELECT * FROM (
SELECT *,ROW_NUMBER() OVER(ORDER BY in_time desc) p FROM tablexx WHERE in_time <@dtMax AND class_id=@classId
) t WHERE t.p BETWEEN @startIndex AND @endIndex ";
using (connection)
{
DataTable table = new DataTable("MyTable");
var reader = connection.ExecuteReader(Sql, new { dtMax = dtMax, classId = classId, startIndex = startIndex, endIndex = endIndex }, commandType: System.Data.CommandType.Text);
table.Load(reader);
reader.Dispose();
return table;
}
}
public static int DeleteSourceClassData(DateTime dtMax, int classId)
{
var connection = Config.GetConnection(Config.SourceDBStr);
string Sql = @" delete from tablexx WHERE in_time <@dtMax AND class_id=@classId ";
using (connection)
{
return connection.Execute(Sql, new { dtMax = dtMax, classId = classId }, commandType: System.Data.CommandType.Text);
}
}
//SqlBulkCopy 批量添加数据
public static int AddTargetData(DataTable data)
{
var connection = Config.GetConnection(Config.TargetDBStr);
using (var sbc = new SqlBulkCopy(connection))
{
sbc.DestinationTableName = "tablexx_2017";
sbc.ColumnMappings.Add("class_id", "class_id");
sbc.ColumnMappings.Add("in_time", "in_time");
.
.
.
using (connection)
{
connection.Open();
sbc.WriteToServer(data);
}
}
return 1;
}
}运行报告:
程序本机运行，开 vpn 连接远程 DB 服务器，运行 1 分钟，迁移的数据数据量为 1915560，每秒约 3 万条数据
1915560 / 60=31926 条/秒
cpu 情况（不高）：
磁盘队列情况（不高）：
在以下情况下速度还将提高
1. 源数据库和目标数据库硬盘为 ssd，并且分别为不同的服务器
2. 迁移程序和数据库在同一个局域网，保障数据传输时候带宽不会成为瓶颈
3. 合理的设置 SqlBulkCopy 参数
4. 菜菜的场景大多数场景下每次批量插入的数据量达不到设置的值，因为有的 class_id 对应的数据量就几十条，甚至几条而已，打开关闭数据库连接也是需要耗时的
5. 单纯的批量添加或者批量删除操作
【注：本文源自网络文章资源，由站长整理发布】
