C# 使用 DataReader 和 CSV writer 批量/ block 导出 SQL Server 大表-6ren

C# 使用 DataReader 和 CSV writer 批量/ block 导出 SQL Server 大表

转载作者：行者123 更新时间：2023-12-05 01:09:19

我开发了一个 CSV 批处理写入器。但与 BCP 相比，这个过程似乎相当缓慢。我唯一的要求是将没有标识或主键列的大表导出到多个小型 CSV 文件中，并用相应的批处理 id 命名。

BCP 的问题是它只会写入一个大文件。

我当前的流程是:读取数据并使用 CSV 写入器写入内存流我不断检查内存流是否大于特定的批处理大小，然后我会将内存流异步复制到文本文件中。

在没有内存不足异常的情况下，我可以导出 250MB 的批量文件

但与 BCP 导出相比，此过程需要多 5 倍的时间。

有没有比我正在做的更好的方法来实现批量导出到 CSV。

请指教。

最佳答案

我想到了几个选项:

使用 FETCH/OFFSET

如果源查询能够轻松地在 SQL Server 中进行批处理(例如，您可以关闭的聚集索引)，则 FETCH 和 OFFSET 基本上是免费的。

如果表是堆，则 FETCH/OFFSET 不是一个真正的选项，但您可能会考虑添加聚集索引，因为没有太多反对这样做的好论据(尽管对于 100 GB 表这样做会很昂贵:)

bcp "SELECT * FROM DemoTable ORDER BY ClusteredKey OFFSET 0 ROWS FETCH NEXT 20000 ROWS ONLY" queryout fetch1.csv -S Server -U sa -P Password -w
bcp "SELECT * FROM DemoTable ORDER BY ClusteredKey OFFSET 20000 ROWS FETCH NEXT 20000 ROWS ONLY" queryout fetch2.csv -S Server -U sa -P Password -w
bcp "SELECT * FROM DemoTable ORDER BY ClusteredKey OFFSET 40000 ROWS FETCH NEXT 20000 ROWS ONLY" queryout fetch3.csv -S Server -U sa -P Password -w
bcp "SELECT * FROM DemoTable ORDER BY ClusteredKey OFFSET 60000 ROWS FETCH NEXT 20000 ROWS ONLY" queryout fetch4.csv -S Server -U sa -P Password -w

使用 SqlDataReader

根据使用约 1.2 GB 表的测量结果，C# CSV SQL 导出(下图)的简单实现在同一表和系统上实现了 BCP 75% 的性能。 (它还具有 handling the CSV format correctly 关于嵌入式逗号、引号和 CRLF 的好处)。

static void Main(string[] args)
{
    var con = new SqlConnection(@"Server=(local);Database=Demo;User Id=sa;Password=bar;");
    con.Open();

    var sqr = new SqlCommand("SELECT * FROM dbo.Table", con);

    using (var reader = sqr.ExecuteReader())
    using (var tw = File.CreateText("out.csv"))
    {
        while (reader.Read())
        {
            for (int i = 0; i < reader.FieldCount; i++)
            {
                if (i != 0)
                {
                    tw.Write(',');
                }

                var val = FormatValue(reader[i]);
                if (val == null)
                {
                    // no-op
                }
                else if (val.IndexOfAny(new[] { '"', ',', '\r', '\n' }) >= 0)
                {
                    tw.Write('"');
                    tw.Write(val.Replace("\"", "\"\""));
                    tw.Write('"');
                }
                else
                {
                    tw.Write(val);
                }
            }
            tw.Write("\r\n");
        }
    }
}

private static string FormatValue(object v)
{
    if (v == null)
    {
        return null;
    }
    if (v is DateTime dt)
    {
        return dt.ToString("O");
    }
    if (v is DateTimeOffset dto)
    {
        return dto.ToString("O");
    }
    if (v is byte[] ba)
    {
        var sb = new StringBuilder(2 + ba.Length * 2);
        sb.Append("0x");
        for (int i = 0; i < ba.Length; i++)
        {
            sb.Append(ba[i].ToString("X2"));
        }
        return sb.ToString();
    }
    return v.ToString();
}

性能似乎受到 GC 处理如此多的字符串分配的限制 - 因此，如果需要更高的性能，将其翻译成非 CLR 语言(例如 C++)可能会与 BCP 的性能相匹配。

使用 SSIS

SSIS 可以在一个包中执行所有步骤。确切的步骤可能最好留给另一个问题，但基本上相当于为“文件编号”合成一列并使用平面文件目标。 Bad example of this

使用 SSIS 生成一个大的 CSV，然后拆分它

如果您使用 SSIS(直接或通过 using the Export Data Wizard )，您将获得一个可以拆分的符合 RFC 4180 的 CSV 文件。拆分此类文件的示例工具是:

class Program
{
    static void Main(string[] args)
    {
        int n = 0;
        using (var src = File.OpenRead("rfc4180_in.csv"))
        using (var dst = new CsvRfc4180SplittingWriteStream(() => File.Create($"rfc4180_out{n++}.csv"), 100 /* mb per chunk */ * 1024 * 1024))
        {
            src.CopyTo(dst);
        }
    }
}

/// <summary>
/// Abstract class which uses ParseDataGetCutPoint to split the files into streams at least 
/// cutAfterPosition bytes long.
/// </summary>
abstract class SplittingWriteStream : Stream
{
    private long _TotalPosition;
    private long CurrentStreamPos;
    private readonly long CutAfterPosition;
    private readonly Func<Stream> StreamCtor;
    private Stream CurrentStream;

    public SplittingWriteStream(Func<Stream> createStream, long cutAfterPosition)
    {
        if (cutAfterPosition < 0L)
        {
            throw new ArgumentOutOfRangeException(nameof(cutAfterPosition));
        }
        this.CutAfterPosition = cutAfterPosition;

        this.StreamCtor = createStream ?? throw new ArgumentNullException(nameof(createStream));
        this.CurrentStream = createStream();
    }

    protected override void Dispose(bool disposing) => CurrentStream.Dispose();

    public override void Flush() => CurrentStream.Flush();

    public override void Write(byte[] buffer, int offset, int count)
    {
        // ignore count to always exceed cutAfterPosition
        var cutPoint = ParseDataGetCutPoint(buffer, offset, count, getCutPoint: CurrentStreamPos > CutAfterPosition);
        if (cutPoint < 0)
        {
            CurrentStream.Write(buffer, offset, count);
        }
        else
        {
            if (cutPoint > 0)
            {
                CurrentStream.Write(buffer, offset, cutPoint);
            }

            try
            {
                CurrentStream.Dispose();
            }
            finally
            {
                CurrentStream = null;
                CurrentStreamPos = 0L;
                CurrentStream = StreamCtor();
            }

            if (cutPoint != count)
            {
                CurrentStream.Write(buffer, offset + cutPoint, count - cutPoint);
            }
        }

        CurrentStreamPos += count;
        _TotalPosition += count;
    }

    protected abstract int ParseDataGetCutPoint(byte[] buffer, int offset, int count, bool getCutPoint);

    #region Stream Write-only stubs

    public override bool CanRead => false;
    public override bool CanSeek => false;
    public override bool CanWrite => true;
    public override long Length => throw new NotSupportedException();
    public override int Read(byte[] buffer, int offset, int count) => throw new NotSupportedException();
    public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException();
    public override void SetLength(long value) => throw new NotSupportedException();

    public override long Position
    {
        get => _TotalPosition;
        set => throw new NotSupportedException();
    }

    #endregion
}

class CsvRfc4180SplittingWriteStream : SplittingWriteStream
{
    public CsvRfc4180SplittingWriteStream(Func<Stream> createStream, long cutAfterPosition)
        : base(createStream, cutAfterPosition)
    {
    }

    bool inQuotedString;
    bool lastWasQuote;
    protected override int ParseDataGetCutPoint(byte[] buffer, int offset, int count, bool getCutPoint)
    {
        int? cutPoint = null;
        for (int n = 0; n < count; n++)
        {
            var i = n + offset;
            StepState(buffer[i]);

            // check for CRLF if desired and not escaped
            if (getCutPoint && !inQuotedString && cutPoint == null
                && buffer[i] == '\r' && n + 1 < count && buffer[i + 1] == '\n')
            {
                cutPoint = n;
            }
        }

        return cutPoint ?? -1;
    }

    private void StepState(byte v)
    {
        var isQuote = v == '"';
        if (lastWasQuote)
        {
            lastWasQuote = false;

            if (isQuote)
            {
                // Double quotes:
                //  nop
                //  Inside quoted string == literal escape
                //  Outside quoted string == empty string
            }
            else
            {
                // quote with non-quote following == toggle quoted string
                inQuotedString ^= true;
            }
        }
        else
        {
            lastWasQuote = isQuote;
        }
    }
}

使用 BCP，然后即时拆分

如果需要 BCP，并且它对 CSV 的(糟糕)处理是可以容忍的，它可以写入命名管道流以动态拆分。

class Program
{
    static void Main(string[] args)
    {
        Thread copyThread;
        var pipeId = $"bcp_{Guid.NewGuid():n}";
        // bcp requires read/write pipe
        using (var np = new NamedPipeServerStream(pipeId))
        {
            copyThread = new Thread(_1 =>
            {
                np.WaitForConnection();
                int n = 0;
                // Use CrlfUtf16leSplittingWriteStream with -w (UTF 16 Little Endian)
                // Use CrlfUtf8SplittingWriteStream other (UTF 8 / ANSII / ASCII / OEM)
                using (var dst = new CrlfUtf16leSplittingWriteStream(() => File.Create($"rfc4180_out{n++}.csv"), 100 /* mb per chunk */ * 1024 * 1024))
                {
                    np.CopyTo(dst);
                }
            });
            copyThread.Name = "Write thread";
            copyThread.IsBackground = true;
            copyThread.Start();

            var bcp = Process.Start(
                @"C:\Program Files\Microsoft SQL Server\Client SDK\ODBC\170\Tools\Binn\bcp.exe",
                $@"FWDB.Rx.RxBatches out \\.\pipe\{pipeId} -S (local) -U sa -P abc -w -t,");
            bcp.WaitForExit();
        }
        copyThread.Join();
    }
}

class CrlfUtf16leSplittingWriteStream : SplittingWriteStream
{
    public CrlfUtf16leSplittingWriteStream(Func<Stream> createStream, long cutAfterPosition)
        : base(createStream, cutAfterPosition)
    {
    }

    protected override int ParseDataGetCutPoint(byte[] buffer, int offset, int count, bool getCutPoint)
    {
        if (getCutPoint)
        {
            for (int n = 0; n < count - 3 /* CR 00 LF 00 */; n++)
            {
                var i = n + offset;
                if (buffer[i] == '\r' && buffer[i + 1] == 0
                    && buffer[i + 2] == '\n' && buffer[i + 3] == 0)
                {
                    // split after CRLF
                    return n + 4;
                }
            }
        }

        return -1;
    }
}

class CrlfUtf8SplittingWriteStream : SplittingWriteStream
{
    public CrlfUtf8SplittingWriteStream(Func<Stream> createStream, long cutAfterPosition)
        : base(createStream, cutAfterPosition)
    {
    }

    protected override int ParseDataGetCutPoint(byte[] buffer, int offset, int count, bool getCutPoint)
    {
        if (getCutPoint)
        {
            for (int n = 0; n < count - 1 /* CR LF */; n++)
            {
                var i = n + offset;
                if (buffer[i] == '\r' && buffer[i + 1] == '\n')
                {
                    // split after CRLF
                    return n + 2;
                }
            }
        }

        return -1;
    }
}

/// <summary>
/// Abstract class which uses ParseDataGetCutPoint to split the files into streams at least 
/// cutAfterPosition bytes long.
/// </summary>
abstract class SplittingWriteStream : Stream
{
    private long _TotalPosition;
    private long CurrentStreamPos;
    private readonly long CutAfterPosition;
    private readonly Func<Stream> StreamCtor;
    private Stream CurrentStream;

    public SplittingWriteStream(Func<Stream> createStream, long cutAfterPosition)
    {
        if (cutAfterPosition < 0L)
        {
            throw new ArgumentOutOfRangeException(nameof(cutAfterPosition));
        }
        this.CutAfterPosition = cutAfterPosition;

        this.StreamCtor = createStream ?? throw new ArgumentNullException(nameof(createStream));
        this.CurrentStream = createStream();
    }

    protected override void Dispose(bool disposing) => CurrentStream.Dispose();

    public override void Flush() => CurrentStream.Flush();

    public override void Write(byte[] buffer, int offset, int count)
    {
        // ignore count to always exceed cutAfterPosition
        var cutPoint = ParseDataGetCutPoint(buffer, offset, count, getCutPoint: CurrentStreamPos > CutAfterPosition);
        if (cutPoint < 0)
        {
            CurrentStream.Write(buffer, offset, count);
        }
        else
        {
            if (cutPoint > 0)
            {
                CurrentStream.Write(buffer, offset, cutPoint);
            }

            try
            {
                CurrentStream.Dispose();
            }
            finally
            {
                CurrentStream = null;
                CurrentStreamPos = 0L;
                CurrentStream = StreamCtor();
            }

            if (cutPoint != count)
            {
                CurrentStream.Write(buffer, offset + cutPoint, count - cutPoint);
            }
        }

        CurrentStreamPos += count;
        _TotalPosition += count;
    }

    protected abstract int ParseDataGetCutPoint(byte[] buffer, int offset, int count, bool getCutPoint);

    #region Stream Write-only stubs

    public override bool CanRead => false;
    public override bool CanSeek => false;
    public override bool CanWrite => true;
    public override long Length => throw new NotSupportedException();
    public override int Read(byte[] buffer, int offset, int count) => throw new NotSupportedException();
    public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException();
    public override void SetLength(long value) => throw new NotSupportedException();

    public override long Position
    {
        get => _TotalPosition;
        set => throw new NotSupportedException();
    }

    #endregion
}

class CsvRfc4180SplittingWriteStream : SplittingWriteStream
{
    public CsvRfc4180SplittingWriteStream(Func<Stream> createStream, long cutAfterPosition)
        : base(createStream, cutAfterPosition)
    {
    }

    bool inQuotedString;
    bool lastWasQuote;
    protected override int ParseDataGetCutPoint(byte[] buffer, int offset, int count, bool getCutPoint)
    {
        int? cutPoint = null;
        for (int n = 0; n < count; n++)
        {
            var i = n + offset;
            StepState(buffer[i]);

            // check for CRLF if desired and not escaped
            if (getCutPoint && !inQuotedString && cutPoint == null
                && buffer[i] == '\r' && n + 1 < count && buffer[i + 1] == '\n')
            {
                cutPoint = n;
            }
        }

        return cutPoint ?? -1;
    }

    private void StepState(byte v)
    {
        var isQuote = v == '"';
        if (lastWasQuote)
        {
            lastWasQuote = false;

            if (isQuote)
            {
                // Double quotes:
                //  nop
                //  Inside quoted string == literal escape
                //  Outside quoted string == empty string
            }
            else
            {
                // quote with non-quote following == toggle quoted string
                inQuotedString ^= true;
            }
        }
        else
        {
            lastWasQuote = isQuote;
        }
    }
}

关于C# 使用 DataReader 和 CSV writer 批量/ block 导出 SQL Server 大表，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/65283808/

文章推荐： flutter - 如何使用 future 延迟

文章推荐：榆树在新标签页中打开网址

sql-server - Windows Server 2008 "Server Core"是否适合 SQL Server 实例？
本周我将在 Windows Server 2008 上设置一个专用的 SQL Server 2005 机器，并希望将其精简为尽可能简单，同时仍能发挥全部功能。为此，“服务器核心”选项听起来很有吸引力
sql-server - 我可以随我的应用程序一起分发和安装 SQL Server Server 管理对象吗？
已关闭。此问题不符合Stack Overflow guidelines 。目前不接受答案。这个问题似乎与 help center 中定义的范围内的编程无关。 . 已关闭 8 年前。 Improve
sql-server - 如何将SQL Server 2014数据库备份恢复到SQL Server 2016？
我获取了 2014 版本数据库的备份，并尝试在另一台服务器中将其恢复到具有相同名称和登录名的数据库中。此 SQL Server 版本是 2016。恢复备份文件时，出现此错误: TITLE: Micr
sql-server - 如何将SQL Server 2014数据库备份恢复到SQL Server 2016？
我获取了 2014 版本数据库的备份，并尝试在另一台服务器中将其恢复到具有相同名称和登录名的数据库中。此 SQL Server 版本是 2016。恢复备份文件时，出现此错误: TITLE: Micr
sql-server - Team Foundation Server - Sql Server 版本管理
TFS 是否提供任何增强的方法来存储对 sql server 数据库所做的更改，而不是使用它来对在数据库上执行的 sql 语句的文本文件进行版本控制？或者我正在寻找的功能是否仅在第 3 方工具(如
sql-server - SQL Server 索引最佳实践 (SQL Server 2008)
关闭。这个问题需要多问focused 。目前不接受答案。想要改进此问题吗？更新问题，使其仅关注一个问题 editing this post . 已关闭 9 年前。 Improve this ques
sql-server - 重命名 SQL Server 主机后更新 sys.servers
我即将将我的 SQL Server 2012 实例升级到 SQL Server 2014。我已经克隆了主机 Windows VM 并将其重命名为 foo-2012至 foo-2014 . 重新启动时
sql-server - 授予 SQL Server 登录数据库访问权限 - SQL Server
我想为 SQL Server 登录授予对数据库的访问权限。我知道 sp_grantdbaccess，但它已被弃用。我可以改用什么以及如何检查登录名是否还没有访问数据库的权限？场景:UserA 创建数
sql-server - 是否可以将数据从sql server 2005 中的表恢复到sql server 2008？
客户别无选择，只能在接下来的几天内从 sql server 2000 迁移到 2008。测试显示 2005 年的重要功能出现了 Not Acceptable 性能下降，但 2008 年却没有。好消息是
sql-server - 将数据库从一个 SQL Server 导出到另一个 SQL Server
我有一个测试数据库，我需要将其导出到我们客户的测试环境中。这将是一次性的工作。我正在使用 SQL Server 2005(我的测试数据库是 SQL Server 2005 Express) 执行此
mongodb - 连接到主机时出错 : could not connect to server: server selection error: server selection timeout current topology: Type: Single Servers
我需要将一个 CSV 文件导入到 mongoDB 不幸的是我遇到了以下错误: error connecting to host: could not connect to server: se
sql-server-2008 - 从sql server 2008升级到sql server 2008 R2
我以为 R2 是一个补丁/服务包。我一直在寻找下载，但没有看到。因此，我假设 R2 是一个新版本，并且我需要 sqlserver 2008 r2 的安装介质来进行升级？另外，我需要为新许可证付费吗？
sql-server - SQL server management studio - 如何使用连接字符串连接到 SQL server？
我无法使用 SQL Server Management Studio 连接到 SQL Server。我有一个连接字符串: 我尝试通过在服务器名中输入 myIP、在登录名中输入 MyID、在密码中
sql-server - 如何在一台 SQL Server 上加密并在另一台 SQL Server 上解密？
我们希望使用 SQL Server 加密来加密数据库中的几个列。我们还需要在生产和测试环境之间传输数据。看来最好的解决方案是在生产和测试服务器上使用相同的主 key 、证书和对称 key ，以便我可以
sql-server - FxCop for SQL Server - 分析 SQL Server 数据库潜在问题的工具
有没有可以分析 SQL Server 数据库潜在问题的工具？例如: a foreign key column that is not indexed 没有 FILL FACTOR 的 uniquei
sql-server - 创建复制错误 : You cannot create a publication from server 'X' because the server is too old
我正在尝试从我的 SQL 2012 BI 版本建立复制，但我收到一条奇怪的错误消息! "You cannot create a publication from server 'X' because
sql-server - SQL Server 2005 : How Secure is SQL Server Authentication?
如果您使用 SQL Server 身份验证 (2005)，登录详细信息是否以明文形式通过网络发送？最佳答案如您所愿，安全无忧... 您可以相当轻松地配置 SSL，如果您没有受信任的证书，如果您强制
sql-server - 将数据从一个 SQL Server 数据库表复制到另一个 SQL Server 数据库表
我想将数据从一个表复制到不同服务器之间的另一个表。如果是在同一服务器和不同的数据库中，我使用了以下 SELECT * INTO DB1..TBL1 FROM DB2..TBL1 (to copy w
sql-server - 无法从另一个 SQL Server 连接到 SQL Server 命名实例
我希望得到一些帮助，因为我在这个问题上已经被困了 2 天了! 场景:我可以从我的开发计算机(和其他同事)连接到 SERVER\INSTANCE，但无法从另一个 SQL Server 连接。我得到的错误
sql-server - 创建复制错误 : You cannot create a publication from server 'X' because the server is too old
我正在尝试从我的 SQL 2012 BI 版本建立复制，但我收到一条奇怪的错误消息! "You cannot create a publication from server 'X' because

行者123

个人简介

我是一名优秀的程序员,十分优秀！

作者热门文章

滴滴打车优惠券免费领取

全站热门文章

首页

博学

6Ren·AI

商城