C# fastest way to insert data into SQL database

Question

I am receiving (streamed) data from an external source (over Lightstreamer) into my C# application. My C# application receives data from the listener. The data from the listener are stored in a queue (ConcurrentQueue). The queue is getting cleaned every 0.5 seconds with TryDequeue into a DataTable. The DataTable will then be copy into a SQL database using SqlBulkCopy. The SQL database processes the newly data arrived from the staging table into the final table. I currently receive around 300'000 rows per day (can increae within the next weeks strongly) and my goal is to stay under 1 second from the time I receive the data until they are available in the final SQL table. Currently the maximum rows per seconds I have to process is around 50 rows.

Unfortunately, since receiving more and more data, my logic is getting slower in performance (still far under 1 second, but I wanna keep improving). The main bottleneck (so far) is the processing of the staging data (on the SQL database) into the final table. In order to improve the performance, I would like to switch the staging table into a memory-optimized table. The final table is already a memory-optimized table so they will work together fine for sure.

My questions:

Is there a way to use SqlBulkCopy (out of C#) with memory-optimized tables? (as far as I know there is no way yet)
Any suggestions for the fastest way to write the received data from my C# application into the memory-optimized staging table?

EDIT (with solution):

After the comments/answers and performance evaluations I decided to give up the bulk insert and use SQLCommand to handover a IEnumerable with my data as table-valued parameter into a native compiled stored procedure to store the data directly in my memory-optimized final table (as well as a copy into the "staging" table which now serves as archive). Performance increased significantly (even I did not consider parallelizing the inserts yet (will be at a later stage)).

Here is part of the code:

Memory-optimized user-defined table type (to handover the data from C# into SQL (stored procedure):

CREATE TYPE [Staging].[CityIndexIntradayLivePrices] AS TABLE(
    [CityIndexInstrumentID] [int] NOT NULL,
    [CityIndexTimeStamp] [bigint] NOT NULL,
    [BidPrice] [numeric](18, 8) NOT NULL,
    [AskPrice] [numeric](18, 8) NOT NULL,
    INDEX [IndexCityIndexIntradayLivePrices] NONCLUSTERED 
(
    [CityIndexInstrumentID] ASC,
    [CityIndexTimeStamp] ASC,
    [BidPrice] ASC,
    [AskPrice] ASC
)
)
WITH ( MEMORY_OPTIMIZED = ON )

Native compiled stored procedures to insert the data into final table and staging (which serves as archive in this case):

create procedure [Staging].[spProcessCityIndexIntradayLivePricesStaging]
(
    @ProcessingID int,
    @CityIndexIntradayLivePrices Staging.CityIndexIntradayLivePrices readonly
)
with native_compilation, schemabinding, execute as owner
as 
begin atomic
with (transaction isolation level=snapshot, language=N'us_english')


    -- store prices

    insert into TimeSeries.CityIndexIntradayLivePrices
    (
        ObjectID, 
        PerDateTime, 
        BidPrice, 
        AskPrice, 
        ProcessingID
    )
    select Objects.ObjectID,
    CityIndexTimeStamp,
    CityIndexIntradayLivePricesStaging.BidPrice, 
    CityIndexIntradayLivePricesStaging.AskPrice,
    @ProcessingID
    from @CityIndexIntradayLivePrices CityIndexIntradayLivePricesStaging,
    Objects.Objects
    where Objects.CityIndexInstrumentID = CityIndexIntradayLivePricesStaging.CityIndexInstrumentID


    -- store data in staging table

    insert into Staging.CityIndexIntradayLivePricesStaging
    (
        ImportProcessingID,
        CityIndexInstrumentID,
        CityIndexTimeStamp,
        BidPrice,
        AskPrice
    )
    select @ProcessingID,
    CityIndexInstrumentID,
    CityIndexTimeStamp,
    BidPrice,
    AskPrice
    from @CityIndexIntradayLivePrices


end

IEnumerable filled with the from the queue:

private static IEnumerable<SqlDataRecord> CreateSqlDataRecords()
{


    // set columns (the sequence is important as the sequence will be accordingly to the sequence of columns in the table-value parameter)

    SqlMetaData MetaDataCol1;
    SqlMetaData MetaDataCol2;
    SqlMetaData MetaDataCol3;
    SqlMetaData MetaDataCol4;

    MetaDataCol1 = new SqlMetaData("CityIndexInstrumentID", SqlDbType.Int);
    MetaDataCol2 = new SqlMetaData("CityIndexTimeStamp", SqlDbType.BigInt);
    MetaDataCol3 = new SqlMetaData("BidPrice", SqlDbType.Decimal, 18, 8); // precision 18, 8 scale
    MetaDataCol4 = new SqlMetaData("AskPrice", SqlDbType.Decimal, 18, 8); // precision 18, 8 scale


    // define sql data record with the columns

    SqlDataRecord DataRecord = new SqlDataRecord(new SqlMetaData[] { MetaDataCol1, MetaDataCol2, MetaDataCol3, MetaDataCol4 });


    // remove each price row from queue and add it to the sql data record

    LightstreamerAPI.PriceDTO PriceDTO = new LightstreamerAPI.PriceDTO();

    while (IntradayQuotesQueue.TryDequeue(out PriceDTO))
    {

        DataRecord.SetInt32(0, PriceDTO.MarketID); // city index market id
        DataRecord.SetInt64(1, Convert.ToInt64((PriceDTO.TickDate.Replace(@"\/Date(", "")).Replace(@")\/", ""))); // @ is used to avoid problem with / as escape sequence)
        DataRecord.SetDecimal(2, PriceDTO.Bid); // bid price
        DataRecord.SetDecimal(3, PriceDTO.Offer); // ask price

        yield return DataRecord;

    }


}

Handling the data every 0.5 seconds:

public static void ChildThreadIntradayQuotesHandler(Int32 CityIndexInterfaceProcessingID)
{


    try
    {

        // open new sql connection

        using (SqlConnection TimeSeriesDatabaseSQLConnection = new SqlConnection("Data Source=XXX;Initial Catalog=XXX;Integrated Security=SSPI;MultipleActiveResultSets=false"))
        {


            // open connection

            TimeSeriesDatabaseSQLConnection.Open();


            // endless loop to keep thread alive

            while(true)
            {


                // ensure queue has rows to process (otherwise no need to continue)

                if(IntradayQuotesQueue.Count > 0) 
                {


                    // define stored procedure for sql command

                    SqlCommand InsertCommand = new SqlCommand("Staging.spProcessCityIndexIntradayLivePricesStaging", TimeSeriesDatabaseSQLConnection);


                    // set command type to stored procedure

                    InsertCommand.CommandType = CommandType.StoredProcedure;


                    // define sql parameters (table-value parameter gets data from CreateSqlDataRecords())

                    SqlParameter ParameterCityIndexIntradayLivePrices = InsertCommand.Parameters.AddWithValue("@CityIndexIntradayLivePrices", CreateSqlDataRecords()); // table-valued parameter
                    SqlParameter ParameterProcessingID = InsertCommand.Parameters.AddWithValue("@ProcessingID", CityIndexInterfaceProcessingID); // processing id parameter


                    // set sql db type to structured for table-value paramter (structured = special data type for specifying structured data contained in table-valued parameters)

                    ParameterCityIndexIntradayLivePrices.SqlDbType = SqlDbType.Structured;


                    // execute stored procedure

                    InsertCommand.ExecuteNonQuery();


                }


                // wait 0.5 seconds

                Thread.Sleep(500);


            }

        }

    }
    catch (Exception e)
    {

        // handle error (standard error messages and update processing)

        ThreadErrorHandling(CityIndexInterfaceProcessingID, "ChildThreadIntradayQuotesHandler (handler stopped now)", e);

    };


}

Look at TVP (table value parameter) - you can use at as like a reverse DataReader. lennilobel.wordpress.com/2009/07/29/… — paparazzo
– paparazzo, Commented Apr 19, 2016 at 18:34

Jeroen Mostert · Accepted Answer · 2016-04-19 18:18:54Z

4

Use SQL Server 2016 (it's not RTM yet, but it's already much better than 2014 when it comes to memory-optimized tables). Then use either a memory-optimized table variable or just blast a whole lot of native stored procedure calls in a transaction, each doing one insert, depending on what's faster in your scenario (this varies). A few things to watch out for:

Doing multiple inserts in one transaction is vital to save on network roundtrips. While in-memory operations are very fast, SQL Server still needs to confirm every operation.
Depending on how you're producing data, you may find that parallelizing the inserts can speed things up (don't overdo it; you'll quickly hit the saturation point). Don't try to be very clever yourself here; leverage async/await and/or Parallel.ForEach.
If you're passing a table-valued parameter, the easiest way of doing it is to pass a DataTable as the parameter value, but this is not the most efficient way of doing it -- that would be passing an IEnumerable<SqlDataRecord>. You can use an iterator method to generate the values, so only a constant amount of memory is allocated.

You'll have to experiment a bit to find the optimal way of passing through data; this depends a lot on the size of your data and how you're getting it.

answered Apr 19, 2016 at 18:18

Jeroen Mostert

29k2 gold badges60 silver badges92 bronze badges

Sign up to request clarification or add additional context in comments.

1 Comment

Reboon Over a year ago

I am currently using SQL Server 2014 and I was able to implement my solution with it (even though I had to make some small compromises). But I am gonna consider SQL Server 2016 as soon as possible. The IEnumerable<SqlDataRecord> works great, faster than DataTable. The use of memory-optimized table-valued parameter and native compiled stored procedure reduced the database workload a lot.

Bruce Dunwiddie · Accepted Answer · 2016-04-19 20:57:38Z

1

Batch the data from the staging table to the final table in row counts less than 5k, I typically use 4k, and do not insert them in a transaction. Instead, implement programmatic transactions if needed. Staying under 5k inserted rows keeps the number of row locks from escalating into a table lock, which has to wait until everyone else gets out of the table.

answered Apr 19, 2016 at 20:57

Bruce Dunwiddie

2,9081 gold badge17 silver badges20 bronze badges

Comments

Ryan Intravia · Accepted Answer · 2016-04-19 21:00:22Z

1

Are you sure it's your logic slowing down and not the actual transactions to the database? Entity Framework for example is "sensitive", for lack of a better term, when trying to insert a ton of rows and gets pretty slow.

There's a third party library, BulkInsert, on Codeplex which I've used and it's pretty nice to do bulk inserting of data: https://efbulkinsert.codeplex.com/

You could also write your own extension method on DBContext if you EF that does this too that could be based on record count. Anything under 5000 rows use Save(), anything over that you can invoke your own bulk insert logic.

answered Apr 19, 2016 at 21:00

Ryan Intravia

4206 silver badges13 bronze badges

Collectives™ on Stack Overflow

C# fastest way to insert data into SQL database

3 Answers 3

1 Comment

Comments

Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

3 Answers 3

1 Comment

Comments

Comments

Your Answer

Sign up or log in

Post as a guest

Related