可以从命名管道/ fifo读取Sql Server BULK INSERT吗？

Question

BULK INSERT/bcp是否可以从命名管道读取,fifo- style？

也就是说,不是从实际文本文件中读取,而是可以使BULK INSERT/bcp从另一个进程的写入端的命名管道读取？

例如:

1. 创建命名管道
2. 将文件解压缩到命名管道
3. 使用bcp或BULK INSERT从命名管道读取

要么:

1. 创建4个命名管道
2. 将1个文件拆分为4个流,将每个流写入单独的命名管道
3. 从4个命名管道读取4个表w/bcp或BULK INSERT

我发现的最接近的是这个家伙(网站现在无法访问),谁管理,编写到命名管道瓦特/ BCP,用他自己的用途和使用情况,如下所示:

start /MIN ZipPipe authors_pipe authors.txt.gz 9
bcp  pubs..authors out  \\.\pipe\authors_pipe -T -n

但他无法逆转工作.

所以在我开始讨论傻瓜的差事之前,我想知道从带有BULK INSERT或bcp的命名管道读取是否根本可行.如果可能的话,如何设置它？会NamedPipeServerStream或者在.NET别的东西System.IO.Pipes命名空间是足够的？

例如,使用Powershell 的示例:

[reflection.Assembly]::LoadWithPartialName("system.core")
$pipe = New-Object system.IO.Pipes.NamedPipeServerStream("Bob")

然后什么？

Answer 1

不幸的是，SSIS 平面文件适配器、BULK INSERT 和 BCP 都对文件采取独占写入锁定（即使它实际上并不写入文件）。这就是为什么这不起作用。

我不确定管道是否可以设置为允许在同一管道上有两个独占锁，而无需进行严重的黑客攻击。我想你可以绕过它或者侵入 fltmgr.sys :)

正如其他发帖者所建议的，使用 .NET API 进行批量处理或使用 OLEDB 或 ODBC 接口可能更简单，尽管这意味着您必须编写自己的文件解析器。

Answer 2

我已经成功地使用BULK INSERT(但不是BCP)在Windows 7和SQL Server 2008R2上使用命名管道正常工作.有一些技巧.

首先,我必须在两个不同的线程上创建两个命名管道实例,两个线程具有相同的管道名称.SQL Server将打开第一个实例,从中读取几个字节并关闭它,导致WriteFile在第一个线程中引发PipeException.然后,SQL Server将立即重新打开命名管道,并从中传输所有数据.如果我没有第二个线程在后台准备好提供数据,SQL服务器将在我的第一个线程有时间从PipeException恢复之前返回错误.

其次,我必须在一次调用WriteFile时写入所有数据.我开始循环,我在管道中写了多个批次,但是BULK INSERT只使用了我写的第一批.它似乎执行非阻塞读取,并将任何返回零字节的读取视为文件结尾.

第三,必须将XML格式文件(如果使用)写入常规文件.我没有成功让SQL Server从管道中读取格式文件.我不知道它是否可以从管道中读取非XML格式的文件.

Answer 3

我会评论@DanMenes(感谢您的灵感),但出于参考目的,我将其作为单独的答案添加.

我已经在.NET中找到了一个解决方案,它打开了一个管道(实际上是2,第一个像@DanMenes所说的那样被破坏),准备将数据流式传输到它,然后BULK INSERT用自动生成的格式文件启动.

前提是我可以做类似的事情

  var inMemoryData = new[] {
    new[] { "val1", "val2" },
    new[] { "val3", "val4" },
  };

  using (var importer = new Importer(SqlConnection, "MyTable", "Col1", "Col2"))
  {
    importer.Import(inMemoryData);
  }

我将总结Importer的实现:

1.创建管道

var stream = new NamedPipeServerStream(name, PipeDirection.Out, 2, PipeTransmissionMode.Byte, PipeOptions.Asynchronous);
stream.BeginWaitForConnection(OnConnection, this);

2.接受连接

public void OnConnection(IAsyncResult asyncResult)
{
  Stream.EndWaitForConnection(asyncResult);

  var buffer = Encoding.UTF8.GetBytes(data);
  Stream.Write(buffer, 0, buffer.Length);
  Stream.Close();
}

3.启动BULK INSERT

var insertCommand = DbConnection.CreateCommand();
insertCommand.CommandText = "BULK INSERT [MyTable] FROM '\\.\pipe\mypipe' WITH (FORMATFILE='c:\path\to\formatfile')";
insertCommand.ExecuteNonQuery();

有关更多详细信息,请参阅GitHub项目.

注意:我还没有为项目添加性能测试,但初步测试确实显示了相对于事务性能的2x和5x之间的性能提升INSERTs.