使用逻辑将一对多 CSV 文件移至 SQL答案

【问题标题】：One to Many CSV file moved to SQL using logic使用逻辑将一对多 CSV 文件移至 SQL
【发布时间】：2012-05-02 15:39:17
【问题描述】：

我搜索了很多地方，很少使用论坛发帖...所以我相信我在这里遇到了一个罕见的问题。我有一个包含一对多关系的 csv 文件。比如说，前 10 个字段包含一条记录。这继续到一个空白字段，然后在每个附加字段中是另一个分隔符，它是一个包含字段的管道。这实际上是与单个调用相关的事件的日志文件。

例如。

ID;MACHINEID;ANI;;COMMAND|ARGUMENTS|20120112.06:15:32

粗体文本可以重复多次或多次。这是 SQL 连接的多方面。我需要把它恢复到 SQL 中。我在想 SSIS 脚本块。我在其中循环并从一侧添加到带有 ID 的多个表中。使用管道作为分隔符拆分内部字段。我只是想知道是否有人以前见过这个并且有一个我可能不知道的简单解决方案。我想这在 XML 中创建会更好，但事实并非如此。

感谢任何帮助。如果它主要是建设性的，我什至会接受批评。提前非常感谢。

为了展示餐桌妆

CREATE TABLE [dbo].[tblIVRCalls](
    [CALLID] [char](50) NOT NULL,
    [MACHINEID] [char](50) NOT NULL,
    [CHANNEL] [char](50) NOT NULL,
    [DNIS] [char](50) NOT NULL,
    [ANI] [char](50) NOT NULL,
    [STARTTIME] [smalldatetime] NOT NULL,
    [TRUNK] [char](50) NOT NULL,
    [APPLICATION] [char](50) NULL,
    [PLANID] [char](50) NULL,
    [DERIVEDID] [char](50) NULL,
    [TOTALTIME] [smalldatetime] NOT NULL,
 CONSTRAINT [PK_tblIVRCalls] PRIMARY KEY CLUSTERED 

CREATE TABLE [dbo].[IVRCallActions](
    [pk] [bigint] IDENTITY(1,1) NOT NULL,
    [fkCALLID] [char](50) NOT NULL,
    [SequenceNumber] [bigint] NOT NULL,
    [Command] [char](50) NOT NULL,
    [Arguments] [varchar](max) NOT NULL,
    [ExecutionTime] [datetime] NOT NULL,
 CONSTRAINT [PK_IVRCallActions] PRIMARY KEY CLUSTERED

【问题讨论】：

SSIS 以行方式工作，所以不要指望用 SSIS 解决这个问题。应该用一个简单的C#程序来处理
@JotaBE：您将如何在 C# 中以不遍历行的方式实现这一点？
你的桌子是什么样子的？对于您给出的示例，预期结果是什么？ table1 中的 1 行，table2 中的 N 行？信息将始终分为三个部分，还是该变量也是可变的？
link 尝试使用此方法使用 C# 代码并根据记录选择它的输出。我正在考虑为每条记录运行一个用于非重复的 A 输出和一个用于使用在变量中捕获的外键重复的 B 记录。

标签： .net sql csv ssis

【解决方案1】：

一种选择是将数据导入临时表，其中重复的元素存储为 varchar(max)（或纯粹主义者的 nvarchar(max)）。

然后执行以下操作：

将非重复字段插入表 1。
使用@@Identity 捕获刚刚插入的行的标识
循环遍历重复字段，首先提取行（由某个分隔符定义），然后是行的组成部分（由“|”定义）。
将每一行插入表 2，标识来自 (2)。

我承认第 (3) 步并不是最简单的事情。但是，如果你写一个存储过程并使用拆分函数（http://stackoverflow.com/questions/2647/split-string-in-sql），那么是可行的。

根据我的经验，复杂的字符串操作在数百兆字节内都可以正常工作。如果您需要一次处理多个 GB，那么您可能需要自定义代码。

【讨论】：

我知道 VBA 和 C# 中的 split 函数，我想它是相似的。我会看看你提供的链接。你如何在 SQL 中嵌套循环。我不是那么熟悉。它会是像 CTE 一样的递归查询吗？我对 CTE 不太熟悉，但我想它们听起来很符合要求。

【解决方案2】：

我建议创建一个控制台程序（使用 C#、vbs 或任何你可以使用的东西）从原始文件中获取 2 个文件。

然后你可以在SSIS中处理这两个文件，它非常擅长管理空字段，更改数据类型等等。

程序将简单地逐行读取原始文件，对其进行处理以获取父行和相关子行，并将父行写入一个文件，将子行写入另一个文件。这些文件将是 SSIS 包的条目。

不该做什么：

在 T-SQL 中进行这种处理要困难得多，因为它没有太多允许处理字符串的函数。
无法在 SSIS 中进行此操作。没有任何组件可以将一行分解为几行。更不用说它应该产生两种不同的行。（自定义 SSIS 组件可以做到这一点，但不值得这样做）

【讨论】：

C# 可以通过脚本块添加到 SSIS。 SSIS Design Pattern: Loading Variable-Length Row 展示了如何做到这一点。我将不得不循环和创建。我希望有一个更简单的方法，但这似乎是一个复杂的问题。
令人惊讶。这是绝对正确的。我一直找不到在 sibgle 脚本组件中创建多个缓冲区输出的方法。很好的解决方案！

【解决方案3】：

根据在SSIS Design Pattern: Loading Variable-Length Rows 找到的信息，我能够创建以下过程。 SSIS 脚本块可以做任何你可以用 C# 做的事情，并且可以安装到 SQL Server 以按设定的时间间隔运行。为处理创建了 2 个输出。您通过此脚本为每个缓冲区填充。因此，一个平面文件包含一列。 2 个输出和中间的这个脚本。

    public override void Input0_ProcessInputRow(Input0Buffer Row)
    {
        IVRCallsBuffer.AddRow();
        string[] splitEntireRow = null;
        var byteRow = Row.Column0.GetBlobData(0,(int)Row.Column0.Length);
        var entireRow = System.Text.Encoding.ASCII.GetString(byteRow);
        splitEntireRow = entireRow.Split(';');
        int fieldPosition = 0;
        string CallID = "";
        while (splitEntireRow[fieldPosition] != "")
        {
            var splitIVRCall = splitEntireRow[fieldPosition].Split('=');
            switch (splitIVRCall[0])
            {
                case "CALLID":
                    IVRCallsBuffer.CALLID = splitIVRCall[1];
                    CallID = splitIVRCall[1];
                    break;
                case "MACHINEID":
                    IVRCallsBuffer.MACHINEID = splitIVRCall[1];
                    break;
                case "CHANNEL":
                    IVRCallsBuffer.CHANNEL = splitIVRCall[1];
                    break;
                case "DNIS":
                    IVRCallsBuffer.DNIS = splitIVRCall[1];
                    break;
                case "ANI":
                    IVRCallsBuffer.ANI = splitIVRCall[1];
                    break;
                case "STARTTIME":
                    IVRCallsBuffer.STARTTIME = DateTime.ParseExact(splitIVRCall[1], "yyyyMMdd.HH:mm:ss.fff", System.Globalization.CultureInfo.CurrentUICulture);
                    break;
                case "TRUNK":
                    IVRCallsBuffer.TRUNK = splitIVRCall[1];
                    break;
                case "APPLICATION":
                    IVRCallsBuffer.Application = splitIVRCall[1];
                    break;
                case "PLANID":
                    IVRCallsBuffer.PLANID = splitIVRCall[1];
                    break;
                case "DERIVEDID":
                    IVRCallsBuffer.DERIVEDID = splitIVRCall[1];
                    break;
                case "TOTALTIME":
                    IVRCallsBuffer.TOTALTIME = DateTime.ParseExact(splitIVRCall[1],"mm:ss.fff",System.Globalization.CultureInfo.CurrentUICulture);
                    break;   
            }
            fieldPosition++;
        }
        fieldPosition++;
        uint sequence = 1;
        while (splitEntireRow.GetUpperBound(0) -1 > fieldPosition)
        {
            IVRCallActionsBuffer.AddRow();
            var splitIVRCallAction = splitEntireRow[fieldPosition].Split('|');
            IVRCallActionsBuffer.fkCALLID = CallID;
            IVRCallActionsBuffer.SequenceNumber = sequence;
            IVRCallActionsBuffer.Command = splitIVRCallAction[0];
            IVRCallActionsBuffer.Arguments = splitIVRCallAction[1];
            IVRCallActionsBuffer.ExecutionTime = DateTime.ParseExact(splitIVRCallAction[2],"yyyyMMdd.HH:mm:ss.fff", System.Globalization.CultureInfo.CurrentUICulture);
            fieldPosition++;
            sequence++;
        }

    }
}

【讨论】：