我有一个具有以下结构的表:
CREATE TABLE [dbo].[case_waveform_data] (
[case_id] INT NOT NULL,
[channel_index] INT NOT NULL,
[seconds_between_points] REAL NOT NULL,
[last_time_stamp] DATETIME NOT NULL,
[value_array] VARBINARY (MAX) NULL
);
这个表将包含数百万条记录。我想从一个数据库中按case_id和channel_index读取数据,然后通过解压value_array数据并将它们链接在一起来组合5分钟的数据,压缩该流,然后将组合块添加到另一个数据库中。
我拥有的代码对大约100k条记录以上的数据运行良好。之后,我会遇到随机错误,例如系统.data中的内存不足,CRC与压缩/未压缩数据不符,压缩数据中存在无效字符等。如果我超过100k,这些错误将随机发生。
我曾经使用linq循环遍历记录,但后来直接使用SqlDataReader。为了导入记录,我使用SqlBulkCopy,但我发现即使将该部分注释掉,我仍会遇到错误。似乎,如果我将每个组合记录作为插入写入文件,代码将完成,但是如果我开始收集组合记录到列表中,然后将其传递给SqlBulkCopy进行插入,我将获得随机错误。大多数情况下,它是reader.Read()(或使用linq时的foreach(var record in records)行)上的内存不足错误。进程本身的内存工作集、私有和提交约为80MB。
我做错了什么?是否有更好的方法来实现这一点?如果我使用写出的文件,它将达到约300MB,我可以加载那么大的文件吗?
这是整个函数。它已经重写了大约20次,所以可能有一些奇怪的代码:
using (LiveORDataContext dc = new LiveORDataContext(LiveORDataManager.ConnectionString))
{
dc.Log = Console.Out;
dc.ObjectTrackingEnabled = false;
Stopwatch sw = Stopwatch.StartNew();
int recordcount = 0;
// Increase the timeout to 10 minutes for really big cases
dc.CommandTimeout = 600;
//Dictionary<int, int> channelindexes = dc.case_waveform_datas.Where(d => d.case_id == livecaseid).GroupBy(d => d.channel_index).ToDictionary(d => d.Key, d => d.Count());
// get a distict list of all the channel indexes we need to import for this case
List<int> channelindexes = (from wd in dc.case_waveform_datas
where wd.case_id == livecaseid
group wd by wd.channel_index into grp
select grp.Key)
.ToList();
// Loop through each channel's data for the case, combine it and compress it
foreach (int channel in channelindexes)
{
List<case_waveform_data> wavedatalist = new List<case_waveform_data>();
int warehouserecordcount = 0;
float secondsbetweenpoints = float.NaN;
DateTime lastaddedrecordtime = DateTime.MinValue;
DateTime previoustime = DateTime.MinValue;
List<float> wfpoints = new List<float>();
string queryString = String.Format("SELECT case_id, channel_index, last_time_stamp, seconds_between_points, " +
"value_array FROM case_waveform_data " +
"WHERE case_id = {0} and channel_index = {1} " +
"ORDER BY last_time_stamp",
livecaseid, channel);
using (SqlConnection connection = new SqlConnection(LiveORDataManager.ConnectionString))
{
SqlCommand command = new SqlCommand(queryString, connection);
connection.Open();
SqlDataReader reader = command.ExecuteReader();
// Call Read before accessing data.
while (reader.Read()) // Currently fails here
{
var item = new
{
case_id = reader.GetInt32(0),
channel_index = reader.GetInt32(1),
last_time_stamp = reader.GetDateTime(2),
seconds_between_points = reader.GetFloat(3),
value_array = (byte[])reader["value_array"]
};
//var wdlist = from wfd in dc.case_waveform_datas
// where wfd.case_id == livecaseid && wfd.channel_index == channel
// orderby wfd.last_time_stamp
// select new
// {
// wfd.case_id,
// wfd.channel_index,
// wfd.last_time_stamp,
// wfd.seconds_between_points,
// wfd.value_array
// };
// Loop through each channel and create floating point arrays that are larger than
// per second groups.
//foreach (var item in wdlist)
//{
// Get a record count for the info log
recordcount++;
if (float.IsNaN(secondsbetweenpoints))
{
secondsbetweenpoints = item.seconds_between_points > 0.0f
? item.seconds_between_points
: 0.002f;
} // assume .002 as a default if this is not set
if (lastaddedrecordtime == DateTime.MinValue)
{
lastaddedrecordtime = item.last_time_stamp;
}
if (previoustime == DateTime.MinValue)
{
previoustime = item.last_time_stamp;
}
if ((secondsbetweenpoints != item.seconds_between_points && item.seconds_between_points > 0.0f) ||
item.last_time_stamp > lastaddedrecordtime.AddMinutes(5))
{
// The seconds between points has changed so gzip the array of
// floats and insert the record.
var ms = new MemoryStream();
using (var gZipStream = new GZipStream(ms, CompressionMode.Compress))
{
new BinaryFormatter().Serialize(gZipStream, wfpoints.ToArray());
}
// add the new combined record to a list that will be bulk inserted every 1000 records
wavedatalist.Add(
//dcwarehouse.case_waveform_datas.InsertOnSubmit(
new case_waveform_data
{
case_id = warehousecaseid,
channel_index = channel,
seconds_between_points = secondsbetweenpoints,
last_time_stamp = previoustime,
value_array = ms.ToArray()
});
if (writeFile) { writer.WriteLine("(@caseid, {0}, {1}, '{2}', 0x{3}),", channel, secondsbetweenpoints, previoustime, BitConverter.ToString(ms.ToArray()).Replace("-", string.Empty)); }
ms.Close();
wfpoints.Clear();
secondsbetweenpoints = item.seconds_between_points;
lastaddedrecordtime = item.last_time_stamp;
// To keep memory down submit the changes to the warehouse database more often
// than after the whole channel's data has been prepared. This handles cases
// that have run for multiple days
warehouserecordcount++;
if (warehouserecordcount > 300)
{
BulkInsertAll(wavedatalist);
wavedatalist.Clear();
warehouserecordcount = 0;
Console.WriteLine("Recordcount: {0}", recordcount);
}
}
// Decompress the float values and append them
var ms1 = new MemoryStream(item.value_array);
using (var gZipStream = new GZipStream(ms1, CompressionMode.Decompress))
{
// Decompress the float array
float[] wd = (float[])new BinaryFormatter().Deserialize(gZipStream);
// determine the timestamp of the first float given the timestamp of the last float,
// the number of elements and the seconds between floats
var listfirsttimestamp =
item.last_time_stamp.AddSeconds((wd.Length - 1) * secondsbetweenpoints * -1);
// if the last time of the previous list + the seconds between is still
// less than the new list's first time then add in NaNs
while (previoustime.AddSeconds(secondsbetweenpoints) < listfirsttimestamp)
{
wfpoints.Add(float.NaN);
previoustime = previoustime.AddSeconds(secondsbetweenpoints);
}
// now append the list
wfpoints.AddRange(wd);
}
ms1.Close();
previoustime = item.last_time_stamp;
//}
}
// Call Close when done reading.
reader.Close();
}
// If there are any points left for the channel add them here
if (wfpoints.Any())
{
var ms = new MemoryStream();
using (var gZipStream = new GZipStream(ms, CompressionMode.Compress))
{
new BinaryFormatter().Serialize(gZipStream, wfpoints.ToArray());
}
wavedatalist.Add(
new case_waveform_data
{
case_id = warehousecaseid,
channel_index = channel,
seconds_between_points = secondsbetweenpoints,
last_time_stamp = previoustime,
value_array = ms.ToArray()
});
if (writeFile) { writer.WriteLine("(@caseid, {0}, {1}, '{2}', 0x{3}),", channel, secondsbetweenpoints, previoustime, BitConverter.ToString(ms.ToArray()).Replace("-", string.Empty)); }
ms.Close();
}
if (wavedatalist.Count > 0)
{
BulkInsertAll(wavedatalist);
wavedatalist.Clear();
}
Console.WriteLine("Recordcount: {0}", recordcount);
}
sw.Stop();
logger.Info("Livecase: [{0}], Warehouse Caseid: [{1}], Recordcount: [{2}]. Waveform data import took [{3}ms]",
livecaseid, warehousecaseid, recordcount, sw.ElapsedMilliseconds);
}
if (writeFile)
{
writer.Close();
}
编辑: 以下是其中一个错误。它发生在这一行:
var item = new
{
case_id = reader.GetInt32(0),
channel_index = reader.GetInt32(1),
last_time_stamp = reader.GetDateTime(2),
seconds_between_points = reader.GetFloat(3),
value_array = (byte[])reader["value_array"]
};
以下是堆栈跟踪信息:
System.InvalidOperationException - Internal connection fatal error.
at System.Data.SqlClient.TdsParserStateObject.TryProcessHeader()
at System.Data.SqlClient.TdsParserStateObject.TryPrepareBuffer()
at System.Data.SqlClient.TdsParserStateObject.TryReadByteArray(Byte[] buff, Int32 offset, Int32 len, Int32& totalRead)
at System.Data.SqlClient.TdsParserStateObject.TryReadPlpBytes(Byte[]& buff, Int32 offst, Int32 len, Int32& totalBytesRead)
at System.Data.SqlClient.TdsParser.TryReadSqlValue(SqlBuffer value, SqlMetaDataPriv md, Int32 length, TdsParserStateObject stateObj)
at System.Data.SqlClient.SqlDataReader.TryReadColumnInternal(Int32 i, Boolean readHeaderOnly)
at System.Data.SqlClient.SqlDataReader.TryReadColumn(Int32 i, Boolean setTimeout, Boolean allowPartiallyReadColumn)
at System.Data.SqlClient.SqlDataReader.GetValueInternal(Int32 i)
at System.Data.SqlClient.SqlDataReader.GetValue(Int32 i)
at System.Data.SqlClient.SqlDataReader.get_Item(String name)
at LiveOR.Data.AccessLayer.LiveORDataManager.ImportWaveformDataLiveToWarehouse(Int32 livecaseid, Int32 warehousecaseid, String backupfilepath) in c:\SRC\LiveOR\LiveOR.Data\LiveORDataManager.cs:line 2416
at VisionSupport.Scheduler.Start() in c:\SRC\LiveOR\VisionSupport\Scheduler.cs:line 90
以上代码出现了OutOfMemoryException异常。以下是堆栈跟踪信息:
at System.Data.SqlClient.TdsParserStateObject.TryReadPlpBytes(Byte[]& buff, Int32 offst, Int32 len, Int32& totalBytesRead)
at System.Data.SqlClient.TdsParser.TryReadSqlValue(SqlBuffer value, SqlMetaDataPriv md, Int32 length, TdsParserStateObject stateObj)
at System.Data.SqlClient.SqlDataReader.TryReadColumnInternal(Int32 i, Boolean readHeaderOnly)
at System.Data.SqlClient.SqlDataReader.TryReadColumn(Int32 i, Boolean setTimeout, Boolean allowPartiallyReadColumn)
at System.Data.SqlClient.SqlDataReader.GetValueInternal(Int32 i)
at System.Data.SqlClient.SqlDataReader.GetValue(Int32 i)
at System.Data.SqlClient.SqlDataReader.get_Item(String name)
at LiveOR.Data.AccessLayer.LiveORDataManager.ImportWaveformDataLiveToWarehouse(Int32 livecaseid, Int32 warehousecaseid, String backupfilepath) in c:\SRC\LiveOR\LiveOR.Data\LiveORDataManager.cs:line 2419
编辑2:
这里是另一个随机的例子。我只需重新运行相同的代码即可获得这些。
行:
float[] wd = (float[])new BinaryFormatter().Deserialize(gZipStream);
异常:
SerializationException: Binary stream '75' does not contain a valid BinaryHeader. Possible causes are invalid stream or object version change between serialization and deserialization.
堆栈跟踪:
at System.Runtime.Serialization.Formatters.Binary.__BinaryParser.Run()
at System.Runtime.Serialization.Formatters.Binary.ObjectReader.Deserialize(HeaderHandler handler, __BinaryParser serParser, Boolean fCheck, Boolean isCrossAppDomain, IMethodCallMessage methodCallMessage)
at System.Runtime.Serialization.Formatters.Binary.BinaryFormatter.Deserialize(Stream serializationStream, HeaderHandler handler, Boolean fCheck, Boolean isCrossAppDomain, IMethodCallMessage methodCallMessage)
at System.Runtime.Serialization.Formatters.Binary.BinaryFormatter.Deserialize(Stream serializationStream)
at LiveOR.Data.AccessLayer.LiveORDataManager.ImportWaveformDataLiveToWarehouse(Int32 livecaseid, Int32 warehousecaseid, String backupfilepath) in c:\SRC\LiveOR\LiveOR.Data\LiveORDataManager.cs:line 2516