linq2db - 服务器端批量复制答案

【问题标题】：linq2db - server side bulkcopylinq2db - 服务器端批量复制
【发布时间】：2021-04-15 22:52:09
【问题描述】：

我正在尝试使用 linq2db 进行“数据库端”批量复制（即 SELECT INTO/INSERT INTO）。但是，我的代码试图将数据集通过网络传输，考虑到相关数据库的大小，这是不可能的。

我的代码如下所示：

using (var db = new MyDb()) {
    var list = db.SourceTable.
        Where(s => s.Year > 2012).
        GroupBy(s => new { s.Column1, s.Column2 }).
        Select(g => new DestinationTable {
            Property1 = 'Constant Value',
            Property2 = g.First().Column1,
            Property3 = g.First().Column2,
            Property4 = g.Count(s => s.Column3 == 'Y')
        });

    db.Execute("TRUNCATE TABLE DESTINATION_TABLE");
    db.BulkCopy(new BulkCopyOptions {
      BulkCopyType = BulkCopyType.MultipleRows
    }, list);
}

生成的 SQL 如下所示：

BeforeExecute
-- DBNAME SqlServer.2017

TRUNCATE TABLE DESTINATION_TABLE
 DataConnection
Query Execution Time (AfterExecute): 00:00:00.0361209. Records Affected: -1.
 DataConnection
BeforeExecute
-- DBNAME SqlServer.2017
DECLARE @take Int -- Int32
SET     @take = 1
DECLARE @take_1 Int -- Int32
SET     @take_1 = 1
DECLARE @take_2 Int -- Int32
...
SELECT
        (
                SELECT TOP (@take)
                        [p].[YEAR]
                FROM
                        [dbo].[SOURCE_TABLE] [p]
                WHERE
                        (([p_16].[YEAR] = [p].[YEAR] OR [p_16].[YEAR] IS NULL AND [p].[YEAR] IS NULL) AND ...
...)
FROM SOURCE_TABLE p_16
WHERE p_16.YEAR > 2012
GROUP BY
  ...
DataConnection

这就是批量复制因超时失败而记录的全部内容，即 SqlException "Execution Timeout Expired"。

请注意，将此查询作为 INSERT INTO 语句直接在数据库中运行所需的时间不到 1 秒。

PS：任何人都对基于代码的优秀 ETL 工具进行大型 DB (+ 1 TB) ETL 有任何建议。鉴于数据库大小，我需要在数据库中运行而不是通过网络传输数据。我试过 pyspark、python bonobo、c# etlbox，它们都移动了太多的数据。我认为 linq2db 有潜力，即基本上就像一个 C# 到 SQL 的转译器，但它也在尝试移动数据。

【问题讨论】：

标签： sql-server etl linq2db

【解决方案1】：

我建议重写您的查询，因为 group by 不能返回第一个元素。 Truncate 也是图书馆的一部分。

var sourceQuery = 
   from s in db.SourceTable 
   where s.Year > 2012
   select new 
   {
      Source = s,
      Count = Sql.Ext.Count(s.Column3 == 'Y' ? 1 : null).Over()
        .PartitionBy(s.Column1, s.Column2).ToValue()
      RN = Sql.Ext.RowNumber().Over()
        .PartitionBy(s.Column1, s.Column2).OrderByDesc(s.Year).ToValue()
   };

db.DestinationTable.Truncate();

sourceQuery.Where(s => s.RN == 1)
  .Insert(db.DestinationTable, 
    e => new DestinationTable 
    {
       Property1 = 'Constant Value',
       Property2 = e.Source.Column1,
       Property3 = e.Source.Column2,
       Property4 = e.Count
    });

【讨论】：

【解决方案2】：

经过一番调查，我偶然发现了this issue。这使我找到了解决方案。上面的代码需要改成：

db.Execute("TRUNCATE TABLE DESTINATION_TABLE");
db.SourceTable.
  Where(s => s.Year > 2012).
  GroupBy(s => new { s.Column1, s.Column2 }).
  Select(g => new DestinationTable {
    Property1 = 'Constant Value',
    Property2 = g.First().Column1,
    Property3 = g.First().Column2,
    Property4 = g.Count(s => s.Column3 == 'Y')
  }).Insert(db.DestinationTable, e => e);

linq2db 项目的文档还有一些不足之处，但就功能而言，它看起来像是一个很棒的 ETL 项目（没有可怕的 1000 行复制/粘贴 sql/ssis 脚本）。

【讨论】：

很好，linq2db 对于 ETL 来说非常强大。