Using SqlBulkCopy to Insert Massive Amounts of Data Into Windows Azure SQL Database

Recently I wrote about Inserting Large Amounts of Data Into Windows Azure SQL Database and this works well for reasonable amounts of data (5 to ~1000 inserts). Faced with a much bigger challenge which required a different approach, I started looking for new alternatives and found SqlBulkCopy.

The code from this Post is part of the Brisebois.WindowsAzure NuGet Package

To install Brisebois.WindowsAzure, run the following command in the Package Manager Console

PM> Install-Package Brisebois.WindowsAzure

Get more details about the Nuget Package.

SqlBulkCopy requires us to build a DataTable object which allows us to stream records from a DataTable to
SQL Database without being throttled. I used The Transient Fault Handling Application Block to provide resiliency against transient faults. Furthermore, locking the table during the operation will provide greater performance.

Performance will be affected by triggers and by indexes

The following code shows how to use the BulkWriter. The Dictionary is used to map columns from the DataTable to the actual table in SQL Database.The WriteWithRetries is not an async method because I usually execute this code within a Task which is already executing in parallel.

Before using the BulkWriter be sure to add a configuration setting to your Cloud Configurations for your Role or in your Web.Config or your App.Config. The configuration setting to add is “ConnectionString” containing a valid connection string for your SQL Database instance.

var bulk = new BulkWriter("EventLog", new Dictionary<string, string>
{
    {"Date", "Date"},
    {"EventId", "EventId"},
    {"Name", "Name"},
    {"Details", "Details"}
});

bulk.WriteWithRetries(table);

The table variable is a DataTable that contains  the mapped columns defined in the above Dictionary. Be sure that the column types from your DataTable match up with the column types in the SQL Database table.

BulkWriter

public class BulkWriter
{
    const int MaxRetry = 5;
    const int DelayMs = 100;

    private readonly string tableName;
    private readonly Dictionary<string, string> tableMap;
    private readonly string connString;

    public BulkWriter(string tableName,
                                Dictionary<string, string> tableMap)
    {
        this.tableName = tableName;
        this.tableMap = tableMap;

        // get your connection string
        connString = CloudConfigurationManager
            .GetSetting("ConnectionString");
    }

    public void WriteWithRetries(DataTable datatable)
    {
        TryWrite(datatable);
    }

    private void TryWrite(DataTable datatable)
    {
        var policy = MakeRetryPolicy();
        try
        {
            policy.ExecuteAction(() => Write(datatable));
        }
        catch (Exception ex)
        {

            //TODO: Add logging logic

            Trace.TraceError(ex.ToString());
            throw;
        }
    }

    private void Write(DataTable datatable)
    {
        // connect to SQL
        using (var connection =
            new SqlConnection(connString))
        {
            var bulkCopy = MakeSqlBulkCopy(connection);

            // set the destination table name
            connection.Open();

            using (var dataTableReader = new DataTableReader(datatable))
            {
                bulkCopy.WriteToServer(dataTableReader);
            }

            connection.Close();
        }
    }

    private RetryPolicy<SqlAzureTransientErrorDetectionStrategy> MakeRetryPolicy()
    {
        var fromMilliseconds = TimeSpan.FromMilliseconds(DelayMs);
        var policy = new RetryPolicy<SqlAzureTransientErrorDetectionStrategy>
            (MaxRetry, fromMilliseconds);
        return policy;
    }

    private SqlBulkCopy MakeSqlBulkCopy(SqlConnection connection)
    {
        var bulkCopy =
            new SqlBulkCopy
                (
                connection,
                SqlBulkCopyOptions.TableLock |
                SqlBulkCopyOptions.FireTriggers |
                SqlBulkCopyOptions.UseInternalTransaction,
                null
                )
            {
                DestinationTableName = tableName,
                EnableStreaming = true
            };

        tableMap
            .ToList()
            .ForEach(kp =>
            {
                bulkCopy
            .ColumnMappings
            .Add(kp.Key, kp.Value);
            });
        return bulkCopy;
    }
}