sql-server - 使用 SSIS 加载多个表并保持外键关系

Question

我正在尝试使用 SSIS 将单个文件(包含数百万条记录)中的数据加载到 SQL Server 上的多个表中，同时维护文件中定义的关系。

为了更好地举例说明，假设我正在尝试加载一个文件，其中包含员工姓名、他们过去占用的办公室以及用选项卡分隔的职位历史记录。

文件:

EmployeeName<tab>OfficeHistory<tab>JobLevelHistory
John Smith<tab>501<tab>Engineer
John Smith<tab>601<tab>Senior Engineer
John Smith<tab>701<tab>Manager
Alex Button<tab>601<tab>Senior Assistant
Alex Button<tab>454<tab>Manager

如果我的 Office 数据库架构具有下表:

Employee (nId, name)
Office (nId, number)
JobTitle (nId, titleName)
Employee2Office (nEmpID, nOfficeId)
Employee2JobTitle (nEmpId, nJobTitleID)

如何使用 SSIS 将文件加载到上述自动生成员工、办公室和职位 ID 的架构中，并维护员工与办公室、员工与职位之间的关系？

所以在这种情况下。表格应如下所示:

Employee
1 John Smith
2 Alex Button

Office
1 501
2 601
3 701
4 454

JobTitle
1 Engineer
2 Senior Engineer
3 Manager
4 Senior Assistant

Employee2Office
1 1
1 2
1 3
2 2
2 4

Employee2JobTitle
1 1
1 2
1 3
2 4
2 3

我是 SSIS 的新手，在执行数据流任务时还没有尝试过自动生成 ID 和建立外键关系。任何指示将不胜感激。

谢谢!

Answer 1

一个有趣的问题。这是我的做法(Sql Server 2005)。 (我假设这是每月一次的工作，而不仅仅是一次，因此我添加了可重复性代码。)

1. 为 Employee、JobTitle 和 Office 表创建三个变量(类型 = Object)
2. 使用三个 SQL 任务将这三个表的行选择到相应的变量中。
3. 添加数据流任务。
4. 使用平面文件目标从平面文件中进行选择。

5. 输出进入脚本组件，其中平面文件中的三列作为输入，导入到脚本中的三个表变量，脚本组件中的五个输出，每个输出具有相同的排除组编号，并且输入标记为对于该输出同步，将七个新列(3 个用于 emp，每个输出对应一个，2 个用于工作，2 个用于办公室)添加到输出中，并使用以下代码(对 System.xml.dll 的引用必须添加以使这一切正常工作。):

   Imports System
   Imports System.Data
   Imports System.Math
   Imports Microsoft.SqlServer.Dts.Pipeline.Wrapper
   Imports Microsoft.SqlServer.Dts.Runtime.Wrapper
   Imports System.Collections
   Imports System.Data.OleDb
   
   Public Class ScriptMain
       Inherits UserComponent
   
       Private da As New OleDbDataAdapter
       Private emp As New DataTable
       Private emph As New Hashtable()
       Private job As New DataTable
       Private jobh As New Hashtable()
       Private off As New DataTable
       Private offh As New Hashtable()
       Private maxempid As Integer
       Private maxjobid As Integer
       Private maxoffid As Integer
   
       Public Overrides Sub PreExecute()
           maxempid = 0
           maxjobid = 0
           maxoffid = 0
           da.Fill(emp, Me.Variables.EmpTab)
           For Each dr As DataRow In emp.Rows
               emph.Add(dr.Item("Name"), dr.Item("nID"))
               If (CInt(dr.Item("nID").ToString) > maxempid) Then
                   maxempid = CInt(dr.Item("nID").ToString)
               End If
           Next
           da.Fill(job, Me.Variables.JobTab)
           For Each dr As DataRow In job.Rows
               jobh.Add(dr.Item("titleName"), dr.Item("nID"))
               If (CInt(dr.Item("nID").ToString) > maxempid) Then
                   maxjobid = CInt(dr.Item("nID").ToString)
               End If
           Next
           da.Fill(off, Me.Variables.OffTab)
           For Each dr As DataRow In off.Rows
               offh.Add(dr.Item("number"), dr.Item("nID"))
               If (CInt(dr.Item("nID").ToString) > maxempid) Then
                   maxoffid = CInt(dr.Item("nID").ToString)
               End If
           Next
           emp.Dispose()
           job.Dispose()
           off.Dispose()
           da.Dispose()
           MyBase.PreExecute()
       End Sub
   
   Public Overrides Sub Input0_ProcessInputRow(ByVal Row As Input0Buffer)
   
       If Not emph.ContainsKey(Row.EmployeeName) Then
           maxempid += 1
           emph.Add(Row.EmployeeName, maxempid)
           Row.EmpId = maxempid
           Row.Emp2Id = maxempid
           Row.Emp3Id = maxempid
           Row.DirectRowToEmployee()
       Else
           Row.EmpId = CInt(emph.Item(Row.EmployeeName).ToString)
           Row.Emp2Id = CInt(emph.Item(Row.EmployeeName).ToString)
           Row.Emp3Id = CInt(emph.Item(Row.EmployeeName).ToString)
       End If
       If Not jobh.ContainsKey(Row.JobLevelHistory) Then
           maxjobid += 1
           jobh.Add(Row.JobLevelHistory, maxjobid)
           Row.JobId = maxjobid
           Row.Job2Id = maxjobid
           Row.DirectRowToJobTitle()
       Else
           Row.JobId = CInt(jobh.Item(Row.JobLevelHistory).ToString)
           Row.Job2Id = CInt(jobh.Item(Row.JobLevelHistory).ToString)
       End If
       If Not offh.ContainsKey(Row.OfficeHistory) Then
           maxoffid += 1
           offh.Add(Row.OfficeHistory, maxoffid)
           Row.OffId = maxoffid
           Row.Off2Id = maxoffid
           Row.DirectRowToOfficeNumber()
       Else
           Row.OffId = CInt(offh.Item(Row.OfficeHistory).ToString)
           Row.Off2Id = CInt(offh.Item(Row.OfficeHistory).ToString)
       End If
       Row.DirectRowToEmp2Job()
       Row.DirectRowToEmp2Off()
   End Sub        
   End Class
   

6. 此脚本的结果(该脚本为输入数据中的新值生成 id。它通过将现有表加载到脚本预执行部分的哈希表中，然后检查是否存在name 并基于此增加 maxid 并将其添加到散列(如果它添加到散列)，它还会将该行添加到适当的(emp、job 或 off)输出，或者从每行的散列中检索 maxid .) 无论上述状态如何，所有行都将写入剩余的两个输出(emp2job 和 emp2off)。

7. 接下来，发送数据流进行查找(检查目标表中的现有行，然后目标 oledb 连接器(emp、job 和 off 选中身份插入框，emp2job 和 emp2off 取消选中检查约束)。