Загрузка нескольких таблиц с использованием служб SSIS с сохранением взаимосвязей внешних ключей

Question

Я пытаюсь загрузить данные из одного файла (с миллионами записей) в несколько таблиц на SQL Server, используя SSIS, сохраняя при этом отношения, определенные в файле.

Чтобы лучше разобраться с примером, давайте предположим, что я пытаюсь загрузить файл, содержащий имя сотрудника, должности, которые они занимали в прошлом, и историю их должностей, разделенных вкладкой.

Файл:

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 для загрузки файла в схему выше Автоматическая генерация идентификаторов для Employee, Office и JobTitle и поддержание отношений между работником и офисами, а также сотрудниками и должностями?

Так и в этом случае. таблицы должны выглядеть так:

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 и не занимался автоматической генерацией идентификаторов и установлением связей по внешнему ключу при выполнении задачи потока данных. Любые указатели будут оценены.

Спасибо!

Answer 1

интересный вопрос. Вот как я бы это сделал (Sql Server 2005). (Я предполагаю, что это ежемесячная работа, а не разовая, поэтому я добавил код для повторяемости.)

1. Создание трех переменных для таблицы Employee, JobTitle и Office (type = 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. Результаты этого сценария (Сценарий генерирует идентификаторы для новых значений во входных данных. Он делает это, загружая существующую таблицу в хеш-таблицы в части предварительного сценария сценария, а затем проверяя наличие имя и на основании этого либо увеличивает maxid и добавляет его в хеш, если он добавляет к хешу, он также добавляет строку в соответствующий (emp, job или off) вывод или извлекает maxid из хеша для каждой строки .) все строки независимо от указанного выше состояния будут записаны на два оставшихся выхода (emp2job и emp2off).

7. Затем отправьте потоки данных для поиска (чтобы проверить наличие строк в таблице назначения, затем в целевых соединителях oledb (emp, job и off установите флажок для вставок идентификаторов, ограничения проверки снятия отметки emp2job и emp2off).

Answer 2

Вот как - немного сложно объяснить только в тексте, но я попробую:

Определите таблицы сотрудников, офисов и должностей в вашей базе данных с помощью столбца идентификаторов, чтобы идентификаторы создавались автоматически.

Определение таблиц "многие ко многим" без (идентификация не нужна или не нужна)

В вашем потоке данных SSIS вы должны сделать это за пару проходов, чтобы сначала установить идентификаторы в базе данных, а затем вернуться и вставить строки «многие ко многим».

Сделать один поток данных:

1. Поместить источник данных для чтения в файл
2. Разделите его на три копии с помощью Multi-Cast. Один будет для сотрудников, один офис, один титул.
3. Для каждого поставьте Сортировку (это, как правило, нет-нет, но, поскольку источником является текст, а не база данных, мы должны идти туда). Установите сортировку так, чтобы она проходила только через одно из трех полей, и отметьте опцию в сортировке, чтобы удалить дубликаты. Это создаст уникальный список (например, выбрать отдельный) для каждой базовой таблицы.
4. Укажите пункт назначения для каждого из трех, подключенных к каждой таблице.

После первого потока данных добавьте второй поток данных. Этот заполнит строки отношения «многие ко многим»

1. Чтение файла с источником данных
2. Добавьте поиск, который находит имя сотрудника в базе данных и возвращает идентификатор сотрудника. Вы получите идентификатор сотрудника, который был сгенерирован выше. (это обычно называется поиском по бизнесу или естественным ключом для суррогатного ключа)
3. Добавить поиск, который найдет заголовок в базе данных и вернет идентификатор заголовка
4. Добавить поиск, который находит Office в базе данных и возвращает идентификатор Office
5. Опять же, мульти-приведения результатов в две копии, один для сотрудника-офиса и один для сотрудника названия
6. В зависимости от логики, которая вам нужна, возможно, снова используйте Sort, чтобы дедуплицировать их (зависит от деталей того, как вы нормализуете от ввода)
7. Поместите результаты в таблицы «многие ко многим» с двумя адресатами.

Answer 3

Если вы уверены, что ваша ссылочная целостность в порядке с данными, которые вы хотите загрузить, вы можете отключить ограничения внешнего ключа в задаче сценария, затем выполнить поток данных с параллельной загрузкой данных и после завершения загрузки данных включить ограничения снова. Если с данными что-то не так, операция завершится неудачей. Вы должны будете разработать стратегию отката или очистки, хотя.

Другой вариант - просто загружать данные последовательно, начиная с основных таблиц и заканчивая дочерними таблицами. Я думаю, что это «более безопасный» вариант, поскольку он не раскрывает вашу целостность данных другим пользователям, которые могут использовать эти таблицы во время загрузки ETL. Я бы предпочел этот вариант.

Answer 4

вы можете написать хранимую процедуру, которая сначала вставляет данные в базовые таблицы (сотрудник, офис и JobTitle), а затем читает ваш файл строка за строкой. после этого вы должны выполнить поиск в основных таблицах для получения идентификаторов и вставить данные в таблицы отношений.