【发布时间】：2015-01-22 09:48:40
【问题描述】：
我想压缩一些包含很少或根本不读取的历史数据的大表。我首先尝试使用内置压缩（row，page，column stored，column-stored archive），但它们都不能压缩行外值（varchar(max)，nvarchar(max)）和最后最终尝试使用CLR 解决方案。
SQL Server Compressed Rowset Sample 解决方案是使用用户定义的CLR 类型压缩给定查询返回的整个行集。
例如：
CREATE TABLE Archive
(
[Date] DATETIME2 DEFAULT(GETUTCDATE())
,[Data] [dbo].[CompressedRowset]
)
INSERT INTO Archive([Data])
SELECT [dbo].[CompressQueryResults]('SELECT * FROM [dbo].[A]')
它正在工作，但我遇到了以下问题：
-
当我尝试压缩大型结果行集时，我收到以下错误：
Msg 0, Level 11, State 0, Line 0 当前命令。结果（如果有）应丢弃。
此外，以下语句有效：
SELECT [dbo].[CompressQueryResults] ('SELECT * FROM [dbo].[LargeA]')但这些不是：
INSERT INTO Archive SELECT [dbo].[CompressQueryResults] ('SELECT * FROM [dbo].[LargeA]' DECLARE @A [dbo].[CompressedRowset] SELECT @A = [dbo].[CompressQueryResults] ('SELECT * FROM [dbo].[LargeA]') -
为了压缩一个行集，
t-sql type应该映射到.net type；不幸的是，这并非适用于所有 sql 类型 - Mapping CLR Parameter Data；我已经扩展了下面的函数来处理更多的类型，但是如何处理像geography这样的类型：static SqlDbType ToSqlType(Type t){ if (t == typeof(int)){ return SqlDbType.Int; } ... if (t == typeof(Byte[])){ return SqlDbType.VarBinary; } else { throw new NotImplementedException("CLR Type " + t.Name + " Not supported for conversion"); } }
这是整个.net代码：
using System;
using System.Data;
using System.Data.SqlClient;
using System.Data.SqlTypes;
using Microsoft.SqlServer.Server;
using System.IO;
using System.Runtime.Serialization.Formatters.Binary;
using System.IO.Compression;
using System.Xml.Serialization;
using System.Xml;
[Serializable]
[Microsoft.SqlServer.Server.SqlUserDefinedType
(
Format.UserDefined
,IsByteOrdered = false
,IsFixedLength = false
,MaxByteSize = -1
)
]
public struct CompressedRowset : INullable, IBinarySerialize, IXmlSerializable
{
DataTable rowset;
public DataTable Data
{
get { return this.rowset; }
set { this.rowset = value; }
}
public override string ToString()
{
using (var sw = new StringWriter())
using (var xw = new XmlTextWriter(sw))
{
WriteXml(xw);
xw.Flush();
sw.Flush();
return sw.ToString();
}
}
public bool IsNull
{
get { return (this.rowset == null);}
}
public static CompressedRowset Null
{
get
{
CompressedRowset h = new CompressedRowset();
return h;
}
}
public static CompressedRowset Parse(SqlString s)
{
using (var sr = new StringReader(s.Value))
using (var xr = new XmlTextReader(sr))
{
var c = new CompressedRowset();
c.ReadXml(xr);
return c;
}
}
#region "Stream Wrappers"
abstract class WrapperStream : Stream
{
public override bool CanSeek
{
get { return false; }
}
public override bool CanWrite
{
get { return false; }
}
public override void Flush()
{
}
public override long Length
{
get { throw new NotImplementedException(); }
}
public override long Position
{
get
{
throw new NotImplementedException();
}
set
{
throw new NotImplementedException();
}
}
public override long Seek(long offset, SeekOrigin origin)
{
throw new NotImplementedException();
}
public override void SetLength(long value)
{
throw new NotImplementedException();
}
}
class BinaryWriterStream : WrapperStream
{
BinaryWriter br;
public BinaryWriterStream(BinaryWriter br)
{
this.br = br;
}
public override bool CanRead
{
get { return false; }
}
public override bool CanWrite
{
get { return true; }
}
public override int Read(byte[] buffer, int offset, int count)
{
throw new NotImplementedException();
}
public override void Write(byte[] buffer, int offset, int count)
{
br.Write(buffer, offset, count);
}
}
class BinaryReaderStream : WrapperStream
{
BinaryReader br;
public BinaryReaderStream(BinaryReader br)
{
this.br = br;
}
public override bool CanRead
{
get { return true; }
}
public override bool CanWrite
{
get { return false; }
}
public override int Read(byte[] buffer, int offset, int count)
{
return br.Read(buffer, offset, count);
}
public override void Write(byte[] buffer, int offset, int count)
{
throw new NotImplementedException();
}
}
#endregion
#region "IBinarySerialize"
public void Read(System.IO.BinaryReader r)
{
using (var rs = new BinaryReaderStream(r))
using (var cs = new GZipStream(rs, CompressionMode.Decompress))
{
var ser = new BinaryFormatter();
this.rowset = (DataTable)ser.Deserialize(cs);
}
}
public void Write(System.IO.BinaryWriter w)
{
if (this.IsNull)
return;
rowset.RemotingFormat = SerializationFormat.Binary;
var ser = new BinaryFormatter();
using (var binaryWriterStream = new BinaryWriterStream(w))
using (var compressionStream = new GZipStream(binaryWriterStream, CompressionMode.Compress))
{
ser.Serialize(compressionStream, rowset);
}
}
#endregion
/// <summary>
/// This procedure takes an arbitrary query, runs it and compresses the results into a varbinary(max) blob.
/// If the query has a large result set, then this procedure will use a large amount of memory to buffer the results in
/// a DataTable, and more to copy it into a compressed buffer to return.
/// </summary>
/// <param name="query"></param>
/// <param name="results"></param>
//[Microsoft.SqlServer.Server.SqlProcedure]
[SqlFunction(DataAccess = DataAccessKind.Read, SystemDataAccess = SystemDataAccessKind.Read, IsDeterministic = false, IsPrecise = false)]
public static CompressedRowset CompressQueryResults(string query)
{
//open a context connection
using (var con = new SqlConnection("Context Connection=true"))
{
con.Open();
var cmd = new SqlCommand(query, con);
var dt = new DataTable();
using (var rdr = cmd.ExecuteReader())
{
dt.Load(rdr);
}
//configure the DataTable for binary serialization
dt.RemotingFormat = SerializationFormat.Binary;
var bf = new BinaryFormatter();
var cdt = new CompressedRowset();
cdt.rowset = dt;
return cdt;
}
}
/// <summary>
/// partial Type mapping between SQL and .NET
/// </summary>
/// <param name="t"></param>
/// <returns></returns>
static SqlDbType ToSqlType(Type t)
{
if (t == typeof(int))
{
return SqlDbType.Int;
}
if (t == typeof(string))
{
return SqlDbType.NVarChar;
}
if (t == typeof(Boolean))
{
return SqlDbType.Bit;
}
if (t == typeof(decimal))
{
return SqlDbType.Decimal;
}
if (t == typeof(float))
{
return SqlDbType.Real;
}
if (t == typeof(double))
{
return SqlDbType.Float;
}
if (t == typeof(DateTime))
{
return SqlDbType.DateTime;
}
if (t == typeof(Int64))
{
return SqlDbType.BigInt;
}
if (t == typeof(Int16))
{
return SqlDbType.SmallInt;
}
if (t == typeof(byte))
{
return SqlDbType.TinyInt;
}
if ( t == typeof(Guid))
{
return SqlDbType.UniqueIdentifier;
}
//!!!!!!!!!!!!!!!!!!!
if (t == typeof(Byte[]))
{
return SqlDbType.VarBinary;
}
else
{
throw new NotImplementedException("CLR Type " + t.Name + " Not supported for conversion");
}
}
/// <summary>
/// This stored procedure takes a compressed DataTable and returns it as a resultset to the clinet
/// or into a table using exec .... into ...
/// </summary>
/// <param name="results"></param>
[Microsoft.SqlServer.Server.SqlProcedure]
public static void UnCompressRowset(CompressedRowset results)
{
if (results.IsNull)
return;
DataTable dt = results.rowset;
var fields = new SqlMetaData[dt.Columns.Count];
for (int i = 0; i < dt.Columns.Count; i++)
{
var col = dt.Columns[i];
var sqlType = ToSqlType(col.DataType);
var colName = col.ColumnName;
if (sqlType == SqlDbType.NVarChar || sqlType == SqlDbType.VarBinary)
{
fields[i] = new SqlMetaData(colName, sqlType, col.MaxLength);
}
else
{
fields[i] = new SqlMetaData(colName, sqlType);
}
}
var record = new SqlDataRecord(fields);
SqlContext.Pipe.SendResultsStart(record);
foreach (DataRow row in dt.Rows)
{
record.SetValues(row.ItemArray);
SqlContext.Pipe.SendResultsRow(record);
}
SqlContext.Pipe.SendResultsEnd();
}
public System.Xml.Schema.XmlSchema GetSchema()
{
return null;
}
public void ReadXml(System.Xml.XmlReader reader)
{
if (rowset != null)
{
throw new InvalidOperationException("rowset already read");
}
var ser = new XmlSerializer(typeof(DataTable));
rowset = (DataTable)ser.Deserialize(reader);
}
public void WriteXml(System.Xml.XmlWriter writer)
{
if (String.IsNullOrEmpty(rowset.TableName))
rowset.TableName = "Rows";
var ser = new XmlSerializer(typeof(DataTable));
ser.Serialize(writer, rowset);
}
}
这里是 SQL 对象的创建：
CREATE TYPE [dbo].[CompressedRowset]
EXTERNAL NAME [CompressedRowset].[CompressedRowset];
GO
CREATE FUNCTION [dbo].[CompressQueryResults] (@query [nvarchar](4000))
RETURNS [dbo].[CompressedRowset]
AS EXTERNAL NAME [CompressedRowset].[CompressedRowset].[CompressQueryResults];
GO
CREATE PROCEDURE [dbo].[UnCompressRowset] @results [dbo].[CompressedRowset]
AS EXTERNAL NAME [CompressedRowset].[CompressedRowset].[UnCompressRowset];
GO
【问题讨论】：
-
序列化产生相当大的输出，并且内置的 GZIpStream 压缩很差。难怪结果并不完全出色。我会尝试将 object[] 序列化为行，以至少摆脱 DataTable。
-
我猜这个例子中的想法是压缩行集以便有更多的信息来压缩（从这里更好的压缩比），对吧？另外，我找不到任何 GZip 替代方案，你能建议这样吗？
-
我无法关闭这个问题，因为它有一个开放的赏金，但我会因为这个原因关闭它太宽泛：可能的答案太多，或者好的答案太长对于这种格式。请添加详细信息以缩小答案集或隔离可以在几段中回答的问题。 所以请提出更具体的问题。我已经回答了这个庞大的、无序的代码示例、意见、担忧中的一些主题......请改进它。
-
硬盘很便宜。买一个更大的。
-
@Magnus 这根本无法解决 - 备份更大，永远在线处理更多数据和其他数据。真的很烦人，这里没有内置压缩 SQL Server 中的大对象。
标签： c# .net tsql sql-server-2012 sqlclr