菜菜呀，咱们业务 BJKJ 有个表数据需要做迁移，现在有多少数据？
更多精彩内容请看 web 前端中文站
http://www.lisa33xiaoq.net 可按 Ctrl + D 进行收藏
大约 21 亿吧，2017 年以前的数据没有业务意义了，给你半天时间把这个事搞定，绩效给
2. 无水平或者垂直切分，但是采用了分区表。分区表策略是按时间降序分的区，将近 30 个分区。正因为分区表的原因，系统才保证了在性能不是太差的情况下坚持至今。
3. 此表除聚集索引之外，无其他索引，无主键（主键其实是利用索引来快速查重的）。所以在频繁插入新数据的情况下，索引调整所耗费的性能比较低。
数据迁移工作包括三个个步骤：
1.? 从源数据表查询出要迁移的数据
2.? 把数据插入新表
3.? 把旧表的数据删除
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 条数据，大约需要的时间为(单位：分)
1.? 在 9 亿数据前查询必须命中索引，就算是非聚集索引菜菜也不推荐，首推聚集索引。
2.??如果你了解索引的原理，你应该明白，不停的插入新数据的时候，索引在不停的更新，调整，以保持树的平衡等特性。尤其是聚集索引影响甚大，因为还需要移动实际的数据。
提取以上两点共同的要素，那就是聚集索引。相应的解决方案也就应运而生：
1.??按照聚集索引分页查询数据
2.? 批量插入数据迎合聚集索引，即：按照聚集索引的顺序批量插入。
3. 按照聚集索引顺序批量删除
1.? 一个表的聚集索引的顺序就是实际数据文件的顺序，映射到磁盘上，本质上位于同一个磁道上，所以操作的时候磁盘的磁头不必跳跃着去操作。
???DateTime?dtMax?=?DateTime.Parse("2017.1.1");
???var?allClassId?=?DBProxy.GeSourcetLstClassId(dtMax)?.OrderBy(s=>s);???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;
}
}
?1915560 / 60=31926 条/秒
?cpu 情况（不高）：
1. 源数据库和目标数据库硬盘为 ssd，并且分别为不同的服务器
2. 迁移程序和数据库在同一个局域网，保障数据传输时候带宽不会成为瓶颈
3. 合理的设置 SqlBulkCopy 参数
4. 菜菜的场景大多数场景下每次批量插入的数据量达不到设置的值，因为有的 class_id 对应的数据量就几十条，甚至几条而已，打开关闭数据库连接也是需要耗时的
5. 单纯的批量添加或者批量删除操作
【注：本文源自网络文章资源，由站长整理发布】
