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

Ответ 1

Вот мой код "multiple inserts".

Вставка строк 100k заняла вместо 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<string> Rows = new List<string>();
        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, чтобы избежать инъекции кода, которая внутренне используется при использовании параметров.

Ответ 2

Я сделал небольшой тест, используя три вещи: 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");
    }
}

Ответ 3

Выполнить команду в 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();
            }
        }
    }
}

Ответ 4

Если Add of AddWithValue не удаляет строки, вы должны сделать это заранее, чтобы избежать ошибок SQL-инъекций и синтаксиса.

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

Ответ 5

Один из способов ускорения - обернуть все вставки в транзакцию 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();

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

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

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 вместо строки.

Ответ 6

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

/// <summary>
    /// Bulk insert some data, uses parameters
    /// </summary>
    /// <param name="table">The Table Name</param>
    /// <param name="inserts">Holds list of data to insert</param>
    /// <param name="batchSize">executes the insert after batch lines</param>
    /// <param name="progress">Progress reporting</param>
    public void BulkInsert(string table, MySQLBulkInsertData inserts, int batchSize = 100, IProgress<double> 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(",");
                }
            }
        }
    }

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

/// <summary>
/// Helper class to builk insert data into a table
/// </summary>
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<object[]>
{
    public MySQLBulkInsertData(params MySQLFieldDefinition[] fieldnames)
    {
        Fields = fieldnames;
    }

    public MySQLFieldDefinition[] Fields { get; private set; }
}

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

    /// <summary>
    /// Return a CSV string of the values in the list
    /// </summary>
    /// <returns></returns>
    /// <exception cref="ArgumentNullException"></exception>
    private string ToParameterCSV(IEnumerable<MySQLFieldDefinition> 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:

        /// <summary>
    /// Return a CSV string of the values in the list
    /// </summary>
    /// <param name="intValues"></param>
    /// <param name="separator"></param>
    /// <param name="encloser"></param>
    /// <returns></returns>
    /// <exception cref="ArgumentNullException"></exception>
    public static string ToCSV<T>(this IEnumerable<T> 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;
    }

Ответ 7

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

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

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

Добавлено

Скажем, у вас есть таблица Products:

• ProductID
• ProductName
• ProductPrice

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

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

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

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

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

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

Ответ 8

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

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

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

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

Ответ 9

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

Ответ 10

Массовая операция была бы хорошим способом привести к этому. То, что читает ваши свойства, а затем создает массовый запрос для вас...

Есть репозиторий github, который содержит оба полезных метода: BulkInsert и BulkUpdate с использованием MySql и EF6+.

BulkUpdate/BulkInsert в основном считывает все свойства из вашей общей сущности, а затем создает массовый запрос для вас.

Ps: Это было разработано для моих нужд, и проект открыт для тех, кто хочет улучшить его или изменить его на лучшее решение, которое будет полезно сообществу.

PS2: Если это не решает проблему, попробуйте внести изменения в проект, чтобы улучшить и достичь того, чего вы хотите, по крайней мере, это хорошее начало.

Пожалуйста, посмотрите здесь

Ответ 11

Я нашел способ избежать использования файла для массовой вставки. В этом соединителе был реализатор загрузки из потока. Таким образом, загрузка может быть сделана что-то подобное

  public void InsertData(string table, List<string> columns, List<List<object>> data) {

  using (var con = OpenConnection() as MySqlConnection) {
    var bulk = new MySqlBulkLoader(con);
    using (var stream = new MemoryStream()) {
      bulk.SourceStream = stream;
      bulk.TableName = table;
      bulk.FieldTerminator = ";";
      var writer = new StreamWriter(stream);

      foreach (var d in data)
        writer.WriteLine(string.Join(";", d));

      writer.Flush();
      stream.Position = 0;
      bulk.Load();
    }
  }
}

Ответ 12

Массовая вставка MySQL (как SqlBulkCopy из MS SQL)

Это сделает свое дело:

    public void Start(string tableName, List<ClsLink> linkList)
    {            
        DataTable table = new DataTable();

        // Getting datatable layout from database
        table = GetDataTableLayout(tableName);

        // Pupulate datatable
        foreach (ClsLink link in linkList)
        {
            DataRow row = table.NewRow();                
            //row["LinkURL"] = link.LinkURL;
            //row["CreateDate"] = link.CreateDate;
            //row["Titel"] = link.Titel;
            table.Rows.Add(row);
        }

        // Insert Datatable in to MySQL
        BulkInsertMySQL(table, tableName);
        // Enjoy
    } 


    public DataTable GetDataTableLayout(string tableName)
    {
        DataTable table = new DataTable();

        using (MySqlConnection connection = new MySqlConnection(connectionString))
        {
            connection.Open();
            // Select * is not a good thing, but in this cases is is very usefull to make the code dynamic/reusable 
            // We get the tabel layout for our DataTable
            string query = $"SELECT * FROM " + tableName + " limit 0";
            using (MySqlDataAdapter adapter = new MySqlDataAdapter(query, connection))
            {
                adapter.Fill(table);
            };
        }

        return table;
    }

    public void BulkInsertMySQL(DataTable table, string tableName)
    {
        using (MySqlConnection connection = new MySqlConnection(connectionString))
        {
            connection.Open();

            using (MySqlTransaction tran = connection.BeginTransaction(IsolationLevel.Serializable))
            {
                using (MySqlCommand cmd = new MySqlCommand())
                {
                    cmd.Connection = connection;
                    cmd.Transaction = tran;
                    cmd.CommandText = $"SELECT * FROM " + tableName + " limit 0";

                    using (MySqlDataAdapter adapter = new MySqlDataAdapter(cmd))
                    {
                        adapter.UpdateBatchSize = 10000;
                        using (MySqlCommandBuilder cb = new MySqlCommandBuilder(adapter))
                        {
                            cb.SetAllValues = true;
                            adapter.Update(table);
                            tran.Commit();
                        }
                    };
                }
            }
        }
    }