Самый эффективный способ вставки строк в базу данных MySQL

Вот мой код «multiple inserts».

Вставка 100 тыс. Строк заняла вместо 40 секунд всего 3 секунды !

 public static void BulkToMySQL() { string ConnectionString = "server=192.168.1xxx"; StringBuilder sCommand = new StringBuilder("INSERT INTO User (FirstName, LastName) VALUES "); using (MySqlConnection mConnection = new MySqlConnection(ConnectionString)) { List Rows = new List(); for (int i = 0; i < 100000; i++) { Rows.Add(string.Format("('{0}','{1}')", MySqlHelper.EscapeString("test"), MySqlHelper.EscapeString("test"))); } sCommand.Append(string.Join(",", Rows)); sCommand.Append(";"); mConnection.Open(); using (MySqlCommand myCmd = new MySqlCommand(sCommand.ToString(), mConnection)) { myCmd.CommandType = CommandType.Text; myCmd.ExecuteNonQuery(); } } } 

Созданный SQL-оператор выглядит следующим образом:

 INSERT INTO User (FirstName, LastName) VALUES ('test','test'),('test','test'),... ; 

Обновление : спасибо Salman A Я добавил MySQLHelper.EscapeString чтобы избежать инъекции кода, который используется внутри, когда вы используете параметры.

Я сделал небольшой тест, используя три вещи: MySqlDataAdapter, транзакции и UpdateBatchSize. Это примерно в 30 раз быстрее, чем ваш первый пример. Mysql работает на отдельной коробке, поэтому есть латентность. Для пакетной обработки может потребоваться некоторая настройка. Код следует:

 string ConnectionString = "server=xxx;Uid=xxx;Pwd=xxx;Database=xxx"; string Command = "INSERT INTO User2 (FirstName, LastName ) VALUES (@FirstName, @LastName);"; using (var mConnection = new MySqlConnection(ConnectionString)) { mConnection.Open(); MySqlTransaction transaction = mConnection.BeginTransaction(); //Obtain a dataset, obviously a "select *" is not the best way... var mySqlDataAdapterSelect = new MySqlDataAdapter("select * from User2", mConnection); var ds = new DataSet(); mySqlDataAdapterSelect.Fill(ds, "User2"); var mySqlDataAdapter = new MySqlDataAdapter(); mySqlDataAdapter.InsertCommand = new MySqlCommand(Command, mConnection); mySqlDataAdapter.InsertCommand.Parameters.Add("@FirstName", MySqlDbType.VarChar, 32, "FirstName"); mySqlDataAdapter.InsertCommand.Parameters.Add("@LastName", MySqlDbType.VarChar, 32, "LastName"); mySqlDataAdapter.InsertCommand.UpdatedRowSource = UpdateRowSource.None; var stopwatch = new Stopwatch(); stopwatch.Start(); for (int i = 0; i < 50000; i++) { DataRow row = ds.Tables["User2"].NewRow(); row["FirstName"] = "1234"; row["LastName"] = "1234"; ds.Tables["User2"].Rows.Add(row); } mySqlDataAdapter.UpdateBatchSize = 100; mySqlDataAdapter.Update(ds, "User2"); transaction.Commit(); stopwatch.Stop(); Debug.WriteLine(" inserts took " + stopwatch.ElapsedMilliseconds + "ms"); } } 

Выполнить команду в Transaction и повторно использовать один и тот же экземпляр команды для каждой итерации. Для дальнейшей оптимизации производительности отправьте 100 запросов в одну команду. Переход на параллельное выполнение может дать лучшую производительность ( Parallel.For ), но убедитесь, что каждый параллельный цикл получает свой собственный экземпляр MySqlCommand .

 public static void CSVToMySQL() { string ConnectionString = "server=192.168.1xxx"; string Command = "INSERT INTO User (FirstName, LastName ) VALUES (@FirstName, @LastName);"; using (MySqlConnection mConnection = new MySqlConnection(ConnectionString)) { mConnection.Open(); using (MySqlTransaction trans = mConnection.BeginTransaction()) { using (MySqlCommand myCmd = new MySqlCommand(Command, mConnection, trans)) { myCmd.CommandType = CommandType.Text; for (int i = 0; i <= 99999; i++) { //inserting 100k items myCmd.Parameters.Clear(); myCmd.Parameters.AddWithValue("@FirstName", "test"); myCmd.Parameters.AddWithValue("@LastName", "test"); myCmd.ExecuteNonQuery(); } trans.Commit(); } } } } 

Если « Add из AddWithValue не AddWithValue строки, вы должны сделать это заранее, чтобы избежать ошибок SQL и синтаксических ошибок.

Стройте INSERT всего за 1000 строк за раз. Это должно работать в 10 раз быстрее, чем вы начали (1 строка за INSERT ). Выполнение всего 100K одновременно является рискованным и, возможно, более медленным. Рискованно, потому что вы можете сбить некоторые ограничения (размер пакета и т. Д.); медленнее из-за необходимости создания огромного журнала ROLLBACK . COMMIT после каждой партии или использовать autocommit=1 .

Этот способ может быть не быстрее, чем метод stringbuilder, но он параметризуется:

 /// 
 /// Bulk insert some data, uses parameters /// 
 /// The Table Name /// Holds list of data to insert /// executes the insert after batch lines /// Progress reporting public void BulkInsert(string table, MySQLBulkInsertData inserts, int batchSize = 100, IProgress progress = null) { if (inserts.Count <= 0) throw new ArgumentException("Nothing to Insert"); string insertcmd = string.Format("INSERT INTO `{0}` ({1}) VALUES ", table, inserts.Fields.Select(p => p.FieldName).ToCSV()); StringBuilder sb = new StringBuilder(); using (MySqlConnection conn = new MySqlConnection(ConnectionString)) using (MySqlCommand sqlExecCommand = conn.CreateCommand()) { conn.Open(); sb.AppendLine(insertcmd); for (int i = 0; i < inserts.Count; i++) { sb.AppendLine(ToParameterCSV(inserts.Fields, i)); for (int j = 0; j < inserts[i].Count(); j++) { sqlExecCommand.Parameters.AddWithValue(string.Format("{0}{1}",inserts.Fields[j].FieldName,i), inserts[i][j]); } //commit if we are on the batch sizeor the last item if (i > 0 && (i%batchSize == 0 || i == inserts.Count - 1)) { sb.Append(";"); sqlExecCommand.CommandText = sb.ToString(); sqlExecCommand.ExecuteNonQuery(); //reset the stringBuilder sb.Clear(); sb.AppendLine(insertcmd); if (progress != null) { progress.Report((double)i/inserts.Count); } } else { sb.Append(","); } } } } 

Это использует вспомогательные classы, как показано ниже:

 /// 
 /// Helper class to builk insert data into a table /// 
 public struct MySQLFieldDefinition { public MySQLFieldDefinition(string field, MySqlDbType type) : this() { FieldName = field; ParameterType = type; } public string FieldName { get; private set; } public MySqlDbType ParameterType { get; private set; } } /// ///You need to ensure the fieldnames are in the same order as the object[] array /// public class MySQLBulkInsertData : List { public MySQLBulkInsertData(params MySQLFieldDefinition[] fieldnames) { Fields = fieldnames; } public MySQLFieldDefinition[] Fields { get; private set; } } 

И этот вспомогательный метод:

  /// 
 /// Return a CSV string of the values in the list /// 
 ///  ///  private string ToParameterCSV(IEnumerable p, int row) { string csv = p.Aggregate(string.Empty, (current, i) => string.IsNullOrEmpty(current) ? string.Format("@{0}{1}",i.FieldName, row) : string.Format("{0},@{2}{1}", current, row, i.FieldName)); return string.Format("({0})", csv); } 

Может быть, не супер элегантный, но он работает хорошо. Мне требуется отслеживание прогресса, поэтому для меня это необходимо, не стесняйтесь удалять эту часть.

Это приведет к выполнению команд SQL, похожих на ваш желаемый результат.

EDIT: ToCSV:

  /// 
 /// Return a CSV string of the values in the list /// 
 ///  ///  ///  ///  ///  public static string ToCSV(this IEnumerable intValues, string separator = ",", string encloser = "") { string result = String.Empty; foreach (T value in intValues) { result = String.IsNullOrEmpty(result) ? string.Format("{1}{0}{1}", value, encloser) : String.Format("{0}{1}{3}{2}{3}", result, separator, value, encloser); } return result; } 

Как говорит Стефан Штайгер, Bulk Insert подходит для ваших ситуаций.

Еще один трюк – это использование промежуточных таблиц, поэтому вместо того, чтобы писать непосредственно в производственную таблицу, вы будете писать на сцену (имеющую ту же структуру). Написав всю информацию, вы просто меняете таблицы. С помощью ap apache вы избегаете блокировки таблиц для вставки (их можно также использовать для обновления и удаления), и этот шаблон в значительной степени используется с MySQL в некоторых проектах.

Кроме того, отключение клавиш таблицы может ускорить вставку, но также может возникнуть ряд проблем при их включении (только для механизма MyISAM).

Добавлено :

Допустим, у вас есть столовая Products :

• Код товара
• Наименование товара
• Цена продукта

Для постановки целей вы создаете промежуточную таблицу под названием ProductsStaging с тем же набором столбцов.

Вся ваша операция, которую вы выполняете на промежуточной таблице:

 UpdateStagingTable(); SwapTables(); UpdateStagingTable(); 

потому что после свопинга ваша промежуточная таблица не имеет новых данных, вы снова вызываете тот же метод. В SwapTables() вы выполняете один оператор SQL:

 RENAME TABLE Products TO ProductsTemp, ProductsStaging TO Products, ProductsTemp TO ProductsStagin; 

Скорость манипулирования данными зависит от движка MySql (например, InnoDB, MyISAM и т. Д.), Поэтому вы можете ускорить вставку, изменив механизм.

Одним из способов ускорения было бы объединение всех вложений в транзакцию ONE (код SQL-Server):

 using (SqlConnection connection = new SqlConnection(CloudConfigurationManager.GetSetting("Sql.ConnectionString"))) { conn.Open(); SqlTransaction transaction = conn.BeginTransaction(); try { foreach (string commandString in dbOperations) { SqlCommand cmd = new SqlCommand(commandString, conn, transaction); cmd.ExecuteNonQuery(); } transaction.Commit(); } // Here the execution is committed to the DB catch (Exception) { transaction.Rollback(); throw; } conn.Close(); } 

Другой способ – загрузить CSV-файл в datatable и использовать функцию пакетной обработки DataAdapter

  DataTable dtInsertRows = GetDataTable(); SqlConnection connection = new SqlConnection(connectionString); SqlCommand command = new SqlCommand("sp_BatchInsert", connection); command.CommandType = CommandType.StoredProcedure; command.UpdatedRowSource = UpdateRowSource.None; // Set the Parameter with appropriate Source Column Name command.Parameters.Add("@PersonId", SqlDbType.Int, 4, dtInsertRows.Columns[0].ColumnName); command.Parameters.Add("@PersonName", SqlDbType.VarChar, 100, dtInsertRows.Columns[1].ColumnName); SqlDataAdapter adpt = new SqlDataAdapter(); adpt.InsertCommand = command; // Specify the number of records to be Inserted/Updated in one go. Default is 1. adpt.UpdateBatchSize = 2; connection.Open(); int recordsInserted = adpt.Update(dtInsertRows); connection.Close(); 

Здесь вы найдете хороший пример.

Или вы можете использовать class MySQL BulkLoader C #:

 var bl = new MySqlBulkLoader(connection); bl.TableName = "mytable"; bl.FieldTerminator = ","; bl.LineTerminator = "\r\n"; bl.FileName = "myfileformytable.csv"; bl.NumberOfLinesToSkip = 1; var inserted = bl.Load(); Debug.Print(inserted + " rows inserted."); 

Если вы делаете несколько вложений в одной команде, вы все равно можете сжать один или два раза, используя StringBuilder вместо строки.

Мое предложение – идея, а не пример или решение. Что делать, если вы не используете INSERT, но передаете данные как несколько параметров (не обязательно все 100K одновременно, вы можете использовать пакеты из 1K, например), чтобы ЗАПОМНЕННАЯ ПРОЦЕДУРА, которая сама делает INSERT.

Я столкнулся с подобной проблемой при работе с EF-MySQL. Вставки EF были слишком медленными и, следовательно, использовали подход, упомянутый fubo . Начнем с того, что производительность значительно улучшилась (~ 20 тыс. Записей были вставлены через ~ 10 секунд), но ухудшилось по мере роста таблицы, причем в таблице было записано ~ 1 М записей, вставка занимает ~ 250 секунд.

Наконец понял вопрос! PK таблицы имеет тип GUID (UUID – char (36)). Поскольку UUID не могут индексировать последовательно, и каждая вставка требует, чтобы индексы были перестроены, она замедлилась.

Исправление заключалось в том, чтобы заменить PK на bigint (или int) и установить его как столбец идентификатора. Это улучшило производительность, вставки занимали в среднем ~ 12 секунд с ~ 2M + записями в таблице!

Думал, что я поделюсь этим нахождением здесь, на случай, если кто-то застрянет в подобной проблеме!