主页 / dba / 问题 / 338129
Accepted
questionto42
Asked: 2024-03-28 07:10:05 +0800 CST

如何通过 C# DataTable 从数据源创建和填充临时表，而无需离开 SSIS 数据流任务？

  • 772

在脚本组件中编码一段时间后，我来问这个问题，主要测试如何循环对象的列Row，请参阅Looping Through Columns in SSIS Script Component - Stack Overflow以及下面要求我打开一个新问题。我还在脚本转换中对多个输入列应用行转换中彻底检查了关于同一件事的更旧的问题，以便我找到了如何循环遍历列。因此，这不是这里的问题。

我将一些 SQL 代理作业迁移到 SSIS。代理作业运行 SQL 并具有大量嵌套select:s 或 CTE:s。我的目标是获取每个给定的 SQL 代码，并将其按每个嵌套查询级别进行拆分，以便在查看 SSIS 数据流任务时，每个级别都能一目了然。一步一步比查看大型嵌套 SQL 查询要好。

我不是第一个提出有关留在 DFT 内并更改数据源的问题的人，请参阅SSIS data flow to update source table rows after copying to destination，尽管这个问题是关于更改源表本身，而这个问题是关于进行更改放在源表的顶部，并将其作为 DFT 中的新输出。

采用这个模式：

select abc.*, abc.column1 + abc.column2 as column3 from (select * from xyz) abc

或者使用 CTE：

with abc as (
select * from xyz
)
select abc.*, abc.column1 + abc.column2 as column3 from abc

两者是相同的，我尝试将它们在数据流任务（DFT）中拆分为两个步骤，一个接一个，这样每个嵌套查询级别最多只有一层 SQL 查询：

OLE DB 源“DataSourceAbc”：

select * from xyz

脚本组件：

create ##tmpDataSourceAbc as (...);
select abc.*, abc.column1 + abc.column2 as column3 from ##tmpDataSourceAbc;

通常只是添加一些列或稍微更改一下列，或者进行查找等，为此，SSIS 或第三方为您提供诸如派生列或查找组件之类的工具。但我想避免将手头的 SQL 代码重写为具有自己的数据类型的 SSIS 语言。你可以说我懒惰或害怕，但我想按原样接管 SQL 代码，而不是使其成为 SSIS 代码。

更重要的是，我想采用允许 SQL 的内置工具，并尽可能避免使用脚本组件，这就是为什么我一开始就采用“OLE DB Source”从服务器使用 SQL 获取数据，而不仅仅是选择脚本组件内的临时表。由于我想进一步处理该查询的输出select，因此我需要在“OLE DB Source”后面有一个脚本组件及其输出箭头，以便它可以将所有列传递到下游。为了清楚起见，如果我让数据流到目的地，数据流将停止，因为它没有输出箭头，并且我必须在进入控制流中的下一步后添加列，并且我不这样做想要那个。我想一直留在 DFT 中，因为 SQL 代理查询也只是一项数据流任务。

因此，计划创建一个脚本组件，该组件从 OLE DB 源（例如）获取数据，然后创建一个临时表并用数据源填充该临时表。然后，它会在临时表之上添加新列或更改某些内容，并且由于它是临时表，因此只需在同一个 DFT 中使用 SQL 即可完成新列。

我知道但仍然不想做的事情：

  • 我知道如何在 C# 中使用 SQL 将数据选择到临时表中，请参阅选择解释/临时表。但这意味着有一个 SQL 来填充临时表，而我想用从数据流上游组件获取的导入数据来填充临时表。

  • 我还知道我可以在控制流中创建临时表，请参阅如何在SSIS控制流任务中创建临时表，然后在数据流任务中使用它？。然而，我希望在需要时在 DFT 中运行时创建临时表，例如嵌套查询或 CTE，只是被临时表查询替换。然后，如果我在 DFT 上持续监督整个数据流而不返回控制流，那么该包的可读性会更高。

因此，计划是：

  • DataTable 存储输入行中的数据及其所有原始列。
  • 在PreExecute方法中，根据输入列在DataTable中动态创建列。
  • 在Input0_ProcessInputRow 方法中，用每个输入行填充DataTable。
  • 在 PostExecute 方法中，将 DataTable 中的所有行插入到临时表中。
  • 添加一列。

这将是一个向数据源添加一列的脚本组件。

但我很难将 Row 对象列与 DataTable 列进行映射。因此，问题是：

有人知道如何将 Row 对象的列映射到我用来填充临时表的 DataTable 吗？就像是：

DataTable dt = new DataTable();
DataRow dr = dt.NewRow(); 

...       

foreach (var c in columns){
             dr[c] = c;
         }
         dt.Rows.Add(dr);

我得到以下代码：“列 'myCol' 不属于 table 。” 因此，在该对象的对象中找不到该对象c的列名。RowDataRowdrDateTabledt

Row 对象列并不或并不总是与 DataTable 的列名称匹配。如何将对象的所有列粘贴到RowDataTable 中的每一行，以便最终可以将完整数据从 DataTable 复制到临时表？

sql-server

2 个回答

  1. Best Answer

    经过长时间检查后，我找到了一种方法来做到这一点。要理解的主要内容是Row对象的列名称已更改，以便仅保留字母，例如：my_column-1 b变为mycolumn1b。如果您映射缩短的列名称，以便在另一侧（即数据源原始列名称）也缩短它们，则它可以工作。

    代码

    #region About me
/* We use a DataTable to store the data from the input rows.
 * In the PreExecute method, we dynamically create columns in the DataTable based on the input columns.
 * In the Input0_ProcessInputRow method, we add each input row to the DataTable.
 * In the PostExecute method, we use SqlBulkCopy to efficiently insert all rows from the DataTable into the temporary table.
 * Once that is done, we can add columns with `select` queries from the temporary table in mere SQL.
 * The last step is not shown here; it could be either in the PostExecute or in another Script Component.
*/
#endregion

#region Namespaces
using System;
using System.Data;
using System.Data.SqlClient;
using Microsoft.SqlServer.Dts.Pipeline.Wrapper;
using System.Diagnostics;
#endregion

[Microsoft.SqlServer.Dts.Pipeline.SSISScriptComponentEntryPointAttribute]
public class ScriptMain : UserComponent
{
    private SqlConnection conn = null;
    private static string _connectionString => @"myconnectionstring";
    private DataTable dt = new DataTable();
    //private SqlCommand command = null;

    public override void PreExecute()
    {
        base.PreExecute();
        conn = new SqlConnection() { ConnectionString = _connectionString };
        conn.Open();
        // create temp table, make sure it matches your input
        using (SqlCommand cmd = new SqlCommand(@"CREATE TABLE ##TempTable (
my_column1 varchar(50));", conn))
        {
            cmd.ExecuteNonQuery();
        }
    }

    public override void Input0_ProcessInputRow(Input0Buffer Row)
    {
        // Create DataRow to store values
        DataRow dr = dt.NewRow();

        // Iterate over input columns to get original column names
        foreach (IDTSInputColumn100 inputColumn in this.ComponentMetaData.InputCollection[0].InputColumnCollection)
        {
            // Get the original column name
            string originalColumnName = inputColumn.Name;

            // Add column to DataTable with the original column name
            dt.Columns.Add(originalColumnName, typeof(object));

            // Get modified column name for checking null values in Row object
            string modifiedColumnName = originalColumnName.Replace("_", "").Replace(" ", "").Replace("-", "");

            // Check if the column is null
            PropertyInfo isNullProperty = typeof(Input0Buffer).GetProperty(modifiedColumnName + "_IsNull");

            if (isNullProperty != null)
            {
                bool isNull = (bool)isNullProperty.GetValue(Row);

                if (isNull)
                {
                    // Handle null value as needed
                    // For example, you can set it to DBNull.Value
                    dr[originalColumnName] = DBNull.Value;
                    continue; // Skip to the next column
                }
            }

            // Get property with matching name from Row object
            PropertyInfo property = typeof(Input0Buffer).GetProperty(modifiedColumnName);

            // Check if the property is null
            object value = property.GetValue(Row);

            // Check if the value is null
            if (value == null)
            {
                // Handle null value as needed
                // For example, you can set it to DBNull.Value
                dr[originalColumnName] = DBNull.Value;
            }
            else
            {
                // Add value to DataRow
                dr[originalColumnName] = value;
            }
        }
        // Add populated DataRow to DataTable
        dt.Rows.Add(dr);
    }

    public override void PostExecute()
    {
        base.PostExecute();
        // here we would bulk copy data into the temp table
        // you may need to batch this operation or handle it differently based on your situation
        using (SqlBulkCopy bulkCopy = new SqlBulkCopy(conn))
        {
            // Now write that DataTable to the database 
            bulkCopy.DestinationTableName = "##tmpTable";
            bulkCopy.WriteToServer(dt);
        }
        this.conn.Close();
    }
}

    输出

    临时表的输出保存在服务器的 tempdb 数据库中。它不会保存在您使用的数据库中。

    请参阅如何查看 sql server 中代码创建的临时表？。

    • 0

  2. 首先建议您打开问题并添加一些细节并尝试给出深思熟虑的答案。老实说，您的答案让我想得更多，并且只要public class ScriptMain在数据集级别创建一次就可以工作，我相信这是可能的，但我还没有尝试过。

    但是，您可能需要在最终解决方案中考虑一些设计点。

    • SSIS 旨在通过一系列转换任务将表从源传递到目标。因此，对于大多数用例来说，有一些组件可以查询子查询（原始来源）并添加列、计算、转换所有这些，而无需脚本。
    • 表现。在原始源定义中使用 CTE/子查询的性能可能会呈指数级提高，因为它将减少实际拉入 SSIS 服务器到内存/磁盘以在其余步骤中处理的行数/数据量。
    • Input0_ProcessInputRow(Input0Buffer Row)被设计为每行转换，其输出可供后续步骤使用。换句话说，它的设计目的是直接在原始行中添加新列或转换列的内容，然后修改包中后续步骤可用的表格数据集。
    • 使用 DataTable 和临时表也会导致大型数据集中出现内存和性能问题。

    所以我认为在确定像这样的每行脚本任务的使用时我会问

    • 可以在 aDerived Column Transformation或 aLookup, Grouping, Conditional Split等中完成吗
    • 可以使用原始源查询中的 SQL 公式或其他转换来完成吗
    • C# 是否允许我做一些在 SQL 中做不到的事情
    • 对于这种特定的转换是否有更好的工具

    我使用脚本已经有好几年了，因为我已经放弃了 SSIS，转而使用其他具有现代 API 和云架构的工具。然而，在 SSIS 中使用脚本通常（不仅仅是）从某个地方获取源，需要抓取网站、与 Microsoft Exchange 交互，做一些实际上不太理想的事情。在纯 SQL ETL 和归档过程中，SSIS 应该能够在其本机组件中处理它。

    • 0

相关问题