如何实现接口IDataReader以便在插入之前处理数据？

问题描述 投票：3回答：2

我有一个存储过程，它给我一个结果集，包含一个包含数百万个未处理行的列。我需要使用SqlBulkCopy将这些数据传输到另一台服务器，但问题是我不能简单地执行以下操作：

using (var con = new SqlConnection(sqlConnectionStringSource))
{
    using (var cmd = new SqlCommand("usp_GetUnprocessedData", con))
    {
        cmd.CommandType = CommandType.StoredProcedure;
        con.Open();
        using (var reader = cmd.ExecuteReader())
        {
            using (var sqlBulk = new SqlBulkCopy(sqlConnectionStringDestination))
            {
                sqlBulk.DestinationTableName = "BulkCopy";
                sqlBulk.BulkCopyTimeout = 0;
                sqlBulk.BatchSize = 200000;
                sqlBulk.WriteToServer(reader);
            }
        }
    }
}

因为根本不会处理数据。

在我的例子中，结果集的第n行如下所示：

value1_n,value2_n,value3_n

其中n只是我介绍的用于区分各行的下标。

在目的地表中，我将其命名为BulkCopy，我希望：

╔══════════╦══════════╦══════════╗
║  Field1  ║  Field2  ║  Field3  ║
╠══════════╬══════════╬══════════╣
║ Value1_1 ║ Value2_1 ║ Value3_1 ║
║ Value1_2 ║ Value2_2 ║ Value3_2 ║
║ ...      ║ ...      ║ ...      ║
║ Value1_n ║ Value2_n ║ Value3_n ║
╚══════════╩══════════╩══════════╝

我被告知通过DataReader接口的实现使用自定义IDataReader，以便在SqlBulkCopy从中复制数据之前逐行处理数据，使用EnableStreamingProperty = true确保内存中只有少量数据，但是我不知道从哪里开始。你能帮我吗？

c# sqldatareader sqlbulkcopy idatareader
2个回答
1
投票

让我们扭转这个问题。而不是找到通用解决方案，创建一个特定的问题。花了几天时间创建一个IDataReader包装器，我知道这并不是那么简单。

我们知道有多少个字段，我们不关心结果中的任何其他字段。我们可以创建一个迭代器方法来分割数据并以流方式逐个返回记录，而不是尝试正确实现IDataReader包装器。 FastMember's ObjectReader可以在任何IEnumerable上包装IDataReader接口：

class MyDTO
{
    public string Field1{get;set;}
    public string Field2{get;set;}
    public string Field3{get;set;}
}

public IEnumerable<MyDTO> ReaderToStream(IDataReader reader)
{
    while(reader.Read())
    {
        var line=reader.GetString(0);
        var fields=String.Split(",",line);
        yield return new MyDTO{Field1=fields[0];Field2=fields[1];Field3=fields[2]};
    }
}

导入方法可以更改为：

using (var con = new SqlConnection(sqlConnectionStringSource))
{
    ...
    using (var reader = cmd.ExecuteReader())
    {
        var recordStream=ReaderToStream(reader);
        using(var rd=ObjectReader(recordStream))
        using (var sqlBulk = new SqlBulkCopy(sqlConnectionStringDestination))
        {
            ...
            sqlBulk.WriteToServer(rd);
        }
    }
}

迭代器仅在SqlBulkCopy请求新记录时调用Read()，因此我们不会最终加载内存中的所有内容。

和IDataReader包装器

Resharper和Visual Studio 2019提供通过委托对包装类的调用来实现接口。在Visual Studio 2019中，这称为Implement interface through 'field_name'。

从这段代码开始：

class ReaderWrapper:IDataReader
{
    private readonly IDataReader _inner ;
    public ReaderWrapper(IDataReader inner)
    {
        _inner = inner;
    }
}

应用重构给出：

class ReaderWrapper:IDataReader
{
    private readonly IDataReader _inner ;
    public ReaderWrapper(IDataReader inner)
    {
        _inner = inner;
    }

    public object this[int i] => _inner[i];

    public object this[string name] => _inner[name];

    public int Depth => _inner.Depth;

    public bool IsClosed => _inner.IsClosed;

    public int RecordsAffected => _inner.RecordsAffected;

    public int FieldCount => _inner.FieldCount;

    public void Close() => _inner.Close();
    public void Dispose() => _inner.Dispose();
    public bool GetBoolean(int i) => _inner.GetBoolean(i);
    public byte GetByte(int i) => _inner.GetByte(i);
    public long GetBytes(int i, long fieldOffset, byte[] buffer, int bufferoffset, int length) => _inner.GetBytes(i, fieldOffset, buffer, bufferoffset, length);
    public char GetChar(int i) => _inner.GetChar(i);
    public long GetChars(int i, long fieldoffset, char[] buffer, int bufferoffset, int length) => _inner.GetChars(i, fieldoffset, buffer, bufferoffset, length);
    public IDataReader GetData(int i) => _inner.GetData(i);
    public string GetDataTypeName(int i) => _inner.GetDataTypeName(i);
    public DateTime GetDateTime(int i) => _inner.GetDateTime(i);
    public decimal GetDecimal(int i) => _inner.GetDecimal(i);
    public double GetDouble(int i) => _inner.GetDouble(i);
    public Type GetFieldType(int i) => _inner.GetFieldType(i);
    public float GetFloat(int i) => _inner.GetFloat(i);
    public Guid GetGuid(int i) => _inner.GetGuid(i);
    public short GetInt16(int i) => _inner.GetInt16(i);
    public int GetInt32(int i) => _inner.GetInt32(i);
    public long GetInt64(int i) => _inner.GetInt64(i);
    public string GetName(int i) => _inner.GetName(i);
    public int GetOrdinal(string name) => _inner.GetOrdinal(name);
    public DataTable GetSchemaTable() => _inner.GetSchemaTable();
    public string GetString(int i) => _inner.GetString(i);
    public object GetValue(int i) => _inner.GetValue(i);
    public int GetValues(object[] values) => _inner.GetValues(values);
    public bool IsDBNull(int i) => _inner.IsDBNull(i);
    public bool NextResult() => _inner.NextResult();
    public bool Read() => _inner.Read();
}

要创建拆分包装器，我们需要将Read()替换为我们自己的版本：

    private string[] _values;

    public bool Read()
    {
        var ok = _inner.Read();
        if (ok)
        {
            //It *could be null*
            if (_inner.IsDBNull(0))
            {
                //What to do? Store an empty array for now
                _values = new string[0];
            }
            var fieldValue = _inner.GetString(0);                
            _values= fieldValue.Split(',');
        }
        return ok;
    }

这会拆分CSV值并将其存储在字符串中。这说明了为什么实现包装器有点麻烦 - 我们需要处理很多事情并决定在意外情况下做什么，比如nulls，空字符串等。

之后，我们需要为SqlBulkCopy调用的方法添加自己的实现。 GetValue()被定义为，FieldCount也是如此。根据列映射类型，按名称或按顺序调用其他成员。

public int FieldCount => _values.Length;

public string GetString(int ordinal) => _values[ordinal];

public object GetValue(int ordinal)=> _values[ordinal];

//What if we have more values than expected?
public int GetValues(object[] values)
{
    if (_values.Length > 0)
    {
        Array.Copy(_values, values,_values.Length);
        return _values.Length;
    }
    return 0;
}

而现在是“有趣”的部分。怎么样GetName()？可能：

public string GetName(int ordinal) => $"Field{ordinal}";

GetOrdinal？它可以在名称映射中调用。变得棘手：

public int GetOrdinal(string name) => int.Parse(name.Substring(5));

让我们希望这有效。

我们还需要覆盖索引：

    public object this[string name] => _values[GetOrdinal(name)];

    public object this[int i] => _values[i];

我忘记了什么？ ......仍然需要处理任意值数字。需要处理空值。没有GetSchemaTable这可能意味着必须明确指定列映射，可能是序数。

一个快速和污垢的IsDbNull实现可能是：

public bool IsDBNull(int i)
{  
    //Covers the "null" case too, when `Length` is 0
    if (i>_values.Length-1)
    {
        return true;
    }
    return _inner.IsDBNull(i);
}

GetSchemaTable更难，因为我们不知道每条记录中有多少值。该表有20多列，所以在我看到它需要之前，我宁愿不编写该代码。

public DataTable GetSchemaTable() => throw new NotImplementedException();

他们说，Leave it as an excercise to the reader

PPS：默认接口实现，因为为什么不

所有这些都可能是一个很好的解决方案，其中C＃8的默认接口方法可用于创建包装的读取器特征。默认情况下，请遵循包装的内部阅读器。这将消除实现中的所有延迟calles。

interface IReaderWrapper:IDataReader
{
    //Gives access to the wrapped reader in the concrete classes
    abstract IDataReader Inner();

    override object this[int i] => Inner()[i];

    override object this[string name] => Inner()[name];

    override int Depth => Inner().Depth;

    override bool IsClosed => Inner().IsClosed;
    ...
}

class SplitterWrapper:IReaderWrapper
{

    private readonly IDataReader _inner ;
    public SplitterWrapper(IDataReader inner)
    {
        _inner = inner;
    }

    IDataReader Inner()=> _inner;

    string[] _values;
    public object this[int i] => _values[i];
    ...
}

此功能在VS 2019附带的C＃8编译器中不起作用，并以某种方式崩溃Sharplab.io。不知道它是否会编译或是否真的需要覆盖。

0
投票

我找到了以下codeproject：https://www.codeproject.com/script/Articles/ViewDownloads.aspx?aid=1095790。看起来您必须获取csv数据并拆分为对象。我用下面的代码修改了代码项目。有很多类型没有实现，您可能需要实现一些其他方法。还不确定结果值应该是什么类型。

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Data;
using System.Data.SqlClient;



namespace ConsoleApplication108
{
    class Program
    {
        static void Main(string[] args)
        {

        }
    }
    public class MyDataReader : IDataReader 
    {
        private SqlConnection conn { get; set; }
        private SqlCommand cmd { get; set; }
        private SqlDataReader reader { get; set; }
        private DataTable schemaTable { get; set; }

        private string data { get; set; }
        private object[] arrayData { get; set; }
        private IEnumerator<object> m_dataEnumerator { get; set; }


        public MyDataReader(string commandText, string connectionString, List<KeyValuePair<string, Type>> columns)
        {
            conn = new SqlConnection(connectionString);
            conn.Open();
            cmd = new SqlCommand(commandText, conn);
            reader = cmd.ExecuteReader();

            schemaTable = new DataTable();
            foreach(KeyValuePair<string,Type> col in columns)
            {
                schemaTable.Columns.Add(col.Key, col.Value);
            }
        }
        public Boolean NextResult()
        {
            return reader.Read();
        }
        public int RecordsAffected
        {
            get { return -1; }
        }
        public int Depth
        {
            get { return -1; }
        }
        public void Dispose()
        {
            Dispose(true);
            GC.SuppressFinalize(this);
        }
        private void Dispose(bool disposing)
        {
            if (disposing)
            {
                if (m_dataEnumerator != null)
                {
                    m_dataEnumerator.Dispose();
                    m_dataEnumerator = null;
                }
            }
        }

        public Boolean IsClosed {
            get { return reader.IsClosed; }
        }
        public Boolean Read()
        {

            if (IsClosed)
            {
                throw new ObjectDisposedException(GetType().Name);
            }
            else
            {
                arrayData = reader.GetString(0).Split(new char[] { ',' }).ToArray();
            }
            return m_dataEnumerator.MoveNext();

        }
        public DataTable GetSchemaTable()
        {
            return schemaTable;
        }
        public void Close()
        {
            Dispose();
        }


        public object this[string name]
        {
            get { throw new NotImplementedException(); }

        }

        public object this[int i]
        {
            get { return arrayData[i]; }
        }
        public int FieldCount
        {
            get { return arrayData.Length; }
        }
        public bool IsDBNull(int i)
        {
              throw new NotImplementedException();
        }
        public bool GetBoolean(int i)
        {
            throw new NotImplementedException();
        }

        public byte GetByte(int i)
        {
            throw new NotImplementedException();
        }

        public long GetBytes(int i, long fieldOffset, byte[] buffer, int bufferoffset, int length)
        {
            throw new NotImplementedException();
        }

        public char GetChar(int i)
        {
            throw new NotImplementedException();
        }

        public long GetChars(int i, long fieldoffset, char[] buffer, int bufferoffset, int length)
        {
            throw new NotImplementedException();
        }

        public IDataReader GetData(int i)
        {
            throw new NotImplementedException();
        }

        public string GetDataTypeName(int i)
        {
            throw new NotImplementedException();
        }

        public DateTime GetDateTime(int i)
        {
            throw new NotImplementedException();
        }

        public decimal GetDecimal(int i)
        {
            throw new NotImplementedException();
        }

        public double GetDouble(int i)
        {
            throw new NotImplementedException();
        }

        public Type GetFieldType(int i)
        {
            throw new NotImplementedException();
        }

        public float GetFloat(int i)
        {
            throw new NotImplementedException();
        }

        public Guid GetGuid(int i)
        {
            throw new NotImplementedException();
        }

        public short GetInt16(int i)
        {
            throw new NotImplementedException();
        }

        public int GetInt32(int i)
        {
            throw new NotImplementedException();
        }

        public long GetInt64(int i)
        {
            throw new NotImplementedException();
        }

        public string GetName(int i)
        {
            throw new NotImplementedException();
        }

        public string GetString(int i)
        {
            throw new NotImplementedException();
        }

        public int GetValues(object[] values)
        {
            values = arrayData;

            return arrayData.Length;
        }
        public int GetOrdinal(string name)
        {
            throw new NotImplementedException();
        }

        public object GetValue(int i)
        {
            return arrayData[i];
        }



    }
}
最新问题
© www.soinside.com 2019 - 2024. All rights reserved.