在mysql和sqlserver (大型数据集)中插入行的最快方法
在mysql和sqlserver (大型数据集)中插入行的最快方法
提问于 2018-05-02 13:37:25
我正在寻找任何可能的改进，以便更快地运行mysql代码。
我创建了一个简单的测试winform应用程序，它创建了两个docker数据库：
一旦创建了实例，它就会在这两个实例上创建一个相似的表。然后，我使用一个存储过程尽可能快地在两个设置中插入一个Category。
以下是一个类别的定义：
public class Category
{
public int Id { get; set; }
[System.ComponentModel.DataAnnotations.StringLength(75)]
public string CategoryName { get; set; }
[System.ComponentModel.DataAnnotations.StringLength(300)]
public string Description { get; set; }
public DateTime CreationTime { get; set; }
}测试结果如下：
100k items
MySql Inserted 100000 items in 2955ms
MySql Inserted 100000 items in 2801ms
MySql Inserted 100000 items in 2706ms
MySql Inserted 100000 items in 2512ms
MySql Inserted 100000 items in 2850ms
SqlServer Inserted 100000 items in 1004ms
SqlServer Inserted 100000 items in 902ms
SqlServer Inserted 100000 items in 858ms
SqlServer Inserted 100000 items in 1421ms
SqlServer Inserted 100000 items in 905ms
600k items
MySql Inserted 600000 items in 21849ms
MySql Inserted 600000 items in 17089ms
MySql Inserted 600000 items in 16776ms
SqlServer Inserted 600000 items in 5677ms
SqlServer Inserted 600000 items in 4635ms
SqlServer Inserted 600000 items in 5474ms下面是MySql的设置
MySql存储过程：
USE `BenchmarkDb`;
DROP procedure IF EXISTS `BenchmarkDb`.`CategoriesInsertWithoutId`;
DELIMITER $
USE `BenchmarkDb`$
CREATE DEFINER=`root`@`%` PROCEDURE `CategoriesInsertWithoutId`(IN JsonPayload LONGTEXT)
BEGIN
insert into BenchmarkDb.Categories
(Category,
Description)
SELECT tt.CategoryName,tt.Description
FROM
JSON_TABLE(
JsonPayload
,"$[*]"
COLUMNS(
Id int PATH "$.Id",
CategoryName VARCHAR(75) PATH "$.CategoryName",
Description VARCHAR(300) PATH "$.Description",
CreationTime DateTime PATH "$.CreationTime"
)
) AS tt;
END$
DELIMITER ;
;使用最新的8.0NPMMySQL驱动程序。发送包含所有数据的大型Json字符串。然后，存储过程将将其转换为一个表并从中插入。
MySql c#代码：
Stopwatch stopwatch = new Stopwatch();
string JsonPayload = JsonConvert.SerializeObject(
TestingDataHelpers.GenerateTestingCategories(100000)
,new IsoDateTimeConverter() { DateTimeFormat= "yyyy-MM-dd HH:mm:ss" });
stopwatch.Start();
var parameters=new List()
{
new MySqlParameter()
{
MySqlDbType=MySqlDbType.LongText,
ParameterName="JsonPayload",
Value=JsonPayload
}
};
DataSet ResultsDataset = new DataSet();
using (var connection = new MySqlConnection("Server=localhost;Uid=root;Pwd=password1234;"))
{
using (var command = connection.CreateCommand())
{
command.CommandText = "BenchmarkDb.CategoriesInsertWithoutId";
command.CommandType = CommandType.StoredProcedure;
if (parameters != null && parameters.Count() > 0)
{
foreach (var parameter in parameters)
{
command.Parameters.Add(parameter);
}
}
using (var dataAdapter = new MySqlDataAdapter(command))
{
dataAdapter.Fill(ResultsDataset);
}
}
}
stopwatch.Stop();下面是使用结构化数据集和表值参数向存储过程发送数据的类似SqlServer代码。
SqlServer存储过程：
IF EXISTS ( SELECT *
FROM sys.objects
WHERE object_id = OBJECT_ID(N'CategoriesInsertWithoutId')
AND type IN ( N'P', N'PC' ) )
DROP PROCEDURE [dbo].[CategoriesInsertWithoutId]
IF type_id('[dbo].[CategoryType]') IS NOT NULL
DROP TYPE [dbo].[CategoryType];
CREATE TYPE CategoryType AS TABLE
( Id int,
CategoryName nvarchar(75),
Description nvarchar(300),
CreationTime DateTime);
CREATE OR ALTER PROCEDURE [dbo].[CategoriesInsertWithoutId]
@CategoriesToInsert CategoryType READONLY
AS
BEGIN
SET NOCOUNT ON;
insert into [dbo].[Categories] (Category,Description)
select c.CategoryName,c.Description from @CategoriesToInsert c
ENDSqlServer c#代码：
Stopwatch stopwatch = new Stopwatch();
var categories = SqlManagerHelpers.ToDataTable(TestingDataHelpers.GenerateTestingCategories(100000));
stopwatch.Start();
var parameters=new List()
{
new SqlParameter()
{
SqlDbType=SqlDbType.Structured,
ParameterName="@CategoriesToInsert",
Value=categories
}
};
DataSet ResultsDataset = new DataSet();
using (var connection = new SqlConnection("Data Source=.;User Id=sa;password=password1234;"))
{
using (var command = connection.CreateCommand())
{
command.CommandText = "dbo.CategoriesInsertWithoutId";
command.CommandType = CommandType.StoredProcedure;
if (parameters != null && parameters.Count() > 0)
{
foreach (var parameter in parameters)
{
command.Parameters.Add(parameter);
}
}
using (var dataAdapter = new SqlDataAdapter(command))
{
dataAdapter.Fill(ResultsDataset);
}
}
}
stopwatch.Stop();下面是上面引用的一些额外的助手类。
static class TestingDataHelpers
{
static Random rnd = new Random();
public static List GenerateTestingCategories(int NumberOfEntriesToMake)
{
List categories=new List();
for(int i = 0; i < NumberOfEntriesToMake; i++)
{
categories.Add(new Category()
{
Id = i,
CategoryName=CategoryNames[rnd.Next(CategoryNames.Count)],
Description=CategoryDescriptions[rnd.Next(CategoryDescriptions.Count)],
CreationTime=DateTime.Now
});
}
return categories;
}
#region CategoryNames
private static List CategoryNames = new List()
{
"Redacted data is redacted.. enjoy some redacted data",
"Redacted data is redacted.. enjoy some redacted data",
"Redacted data is redacted.. enjoy some redacted data",
};
#endregion
#region CategoryDescriptions
private static List CategoryDescriptions = new List()
{
"Redacted data is redacted.. enjoy some redacted data",
"Redacted data is redacted.. enjoy some redacted data",
"Redacted data is redacted.. enjoy some redacted data",
};
#endregion
}
static class SqlManagerHelpers
{
public static DataTable ToDataTable(this IList data)
{
var props = typeof(T).GetProperties().Where(pi => pi.GetCustomAttributes(typeof(SkipPropertyAttribute), true).Length == 0).ToList();
DataTable table = new DataTable();
for(int i =0;i().SingleOrDefault();
if (stringLengthAttribute != null)
{
table.Columns[i].MaxLength = stringLengthAttribute.MaximumLength;
}
}
foreach (T item in data)
{
DataRow row = table.NewRow();
foreach (var prop in props)
row[prop.Name] = prop.GetValue(item) ?? DBNull.Value;
table.Rows.Add(row);
}
return table;
}
public class SkipPropertyAttribute : Attribute
{
}
}以下是所需的数据库架构
MySql数据库和表定义
CREATE DATABASE `BenchmarkDb`;
CREATE TABLE `BenchmarkDb`.`Categories` (
`Id` INT NOT NULL AUTO_INCREMENT,
`Category` VARCHAR(75) NULL,
`Description` VARCHAR(300) NULL,
`CreationTime` DATETIME DEFAULT CURRENT_TIMESTAMP NOT NULL,
PRIMARY KEY (`id`)) ENGINE=InnoDB;SqlServer数据库和表定义
CREATE TABLE [BenchmarkDb].[dbo].[Categories] (
[Id] [int] IDENTITY(1,1) NOT NULL,
[Category] [nvarchar](75) NULL,
[Description] [nvarchar](300) NULL,
[CreationTime] DATETIME NOT NULL DEFAULT GETDATE(),
CONSTRAINT [PK_History] PRIMARY KEY CLUSTERED
(
[Id] ASC
)WITH (PAD_INDEX = OFF, STATISTICS_NORECOMPUTE = OFF, IGNORE_DUP_KEY = OFF, ALLOW_ROW_LOCKS = ON, ALLOW_PAGE_LOCKS = ON)
) ON [PRIMARY]回答 1
Code Review用户
发布于 2019-08-27 20:16:22
我对mysql一无所知，所以我忽略了你问题的那一部分。
对于Server，如果您试图使插入速度更快，则需要：
- 散装地做
- 并行地做
- 将其记录到最低限度
- 分批做
有些事情你可以帮你处理，我在下面列出，否则你就得自己写东西了。
散装作业
批量操作最终归结为尝试在单个操作中完成尽可能多的工作，以一种不影响性能的方式(事务日志记录是最常见的，但还有更多)。这个链接提到了一些好处，我在这里强调的主要好处是：
- 最小测井
- 更好的锁定(BU锁)
- 分批
- 可选触发器/约束
如果您直接从文件中插入，那么BULK INSERT就是您的朋友。它将为您处理上述几乎所有的考虑事项(除了并行性，这是BULK INSERT's控制之外的)。
但是，从C#插入更适合使用SqlBulkCopy。这使您可以在表中执行批量插入操作，并可以被配置忽略约束、触发器、标识列等。
平行插入
并行插入允许SQL一次将多行插入到表中，而不是逐行操作。这通常需要以下几点：
- 一堆
- 没有
IDENTITY列 - 桌面上正确的锁定类型
如果您没有堆(例如，有索引)，那么索引维护将禁止并行插入，并且它将按顺序执行插入。对于大规模的ETL工作负载，这是一个“暂存”数据库/表的很好的用例，它没有这些东西，因此可以获得最好的插入。布伦特·奥扎尔有一个很好的帖子，也涉及到这一点。
IDENTITY列还阻止并行插入，因为要使其正确工作，就需要保持插入的顺序。
如果表上没有正确的锁(BU锁工作，TABLOCK(X)提示(S))，那么Server必须考虑另一个会话也可以修改表，这也防止了并行性。
但是，如果您能够满足所有这些要求，那么您的操作(无论是使用上面所述的内置批量操作，还是按下面的方式滚动您自己的操作)将能够利用SQL Server的附加核心更快地运行。
最小 日志记录
最小日志记录是防止事务日志溢出的方法。有些操作比其他操作更容易回滚，或者需要比其他操作更少的事务日志空间。维护事务日志并不是免费的/廉价的，因此减少所需的事务日志也有助于性能。一般情况下，如果您要遵守规则
- 没有聚集索引的堆(即没有聚集索引的表)中的插入，使用TABLOCK提示，具有足够高的基数估计(> ~1000行)
- 插入到具有聚集索引的表中，该表没有非聚集索引，没有TABLOCK，具有足够高的基数估计值(> ~1000行)
- 向表添加索引，即使该表已经有数据。
批
最后，您可以将一大块工作分解成较小的批。如果事务日志是主要的关注点，因为每个批处理都成为自己的事务，这可能很有用。
这在一般情况下是很难实现的，而且对自己来说也可能是不愉快的。另一个主要问题是数据的正确性；如果另一个用户在未完成批处理时访问数据库，则可能会得到不一致的结果。这是一个很好的用例，用于在另一个表中完成工作，然后交换表以及快照隔离。
页面原文内容由Code Review提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接：
https://codereview.stackexchange.com/questions/193462
复制相关文章
相似问题
腾讯云开发者
