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

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

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

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

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

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

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

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

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

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

После первого потока данных добавьте второй поток данных. Он будет заполнять строки отношения "многие-ко-многим"

1. Прочитайте файл с источником данных
2. Добавьте поиск, который находит имя сотрудника в базе данных и возвращает ID сотрудника. В результате вы получите ID сотрудника, который был сгенерирован выше. (это обычно называется поиском по деловому или естественному ключу для суррогатного ключа)
3. Добавьте поиск, который находит должность в базе данных и возвращает идентификатор должности
4. Добавьте поиск, который находит офис в базе данных и возвращает идентификатор офиса
5. Снова мультитранслируйте результаты в две копии, один для employee-office и один для employee-title
6. В зависимости от логики, которая вам нужна, возможно, используйте Sort, снова, чтобы дедуплицировать их (зависит от деталей того, как вы нормализуете входные данные)
7. Поместите результаты в таблицы many-to-many с двумя направлениями.

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

1. Создайте три переменные для таблицы Employee, JobTitle и Office (type = Object)
2. Используйте три sql задачи выбрать строки для этих трех таблиц в соответствующие переменные.
3. Добавить задачу потока данных.
4. Выберите из плоского файла, используя место назначения для плоского файла.

5. Вывод поступает в компонент сценария с тремя столбцами в плоском файле в качестве ввода, тремя табличными переменными, импортированными в сценарий, пятью выводами в компоненте сценария, каждый с тем же номером группы исключения, и ввод, отмеченный как синхронный для этого output, семь новых столбцов (3 для emp, по одному для каждого вывода, в котором он будет находиться, 2 для работы, 2 для офиса) добавлены к выходным данным и со следующим кодом (Ссылка на System.xml.dll должна быть добавлена ​​в заставить все это работать.):

    Импортирует System 
   Импортирует System.Data 
   Импортирует System.Math 
   Импортирует Microsoft.SqlServer.Dts.Pipeline.Wrapper 
   Импортирует Microsoft.SqlServer.Dts.Runtime.Wrapper 
   Импортирует систему.Коллекции 
   Импортирует System.Data.OleDb 
    
   Открытый класс ScriptMain 
   Наследует UserComponent 
    
   Private da As New OleDbDataAdapter {{1 }} 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 
    
   Общедоступные переопределения Sub PreExecute () 
   maxempid = 0 
   maxjobid = 0 
   maxoffid = 0 
   da.Fill (emp, Me.Variables.EmpTab) {{1 }} Для каждого 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) 
   Для каждого dr As DataRow в job.Rows 
   jobh.Add (dr.Item ("titleName"), dr.Item ("nID ")) 
   Если (CInt (dr.Item (" nID "). ToString)> maxempid) Then 
   maxjobid = CInt (dr.Item ("nID"). ToString) 
   End If 
   Next 
   da.Fill (off, Me.Variables.OffTab) 
   Для каждого 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. Имя сотрудника).ToString) 
   Row.Emp3Id = CInt (emph.Item (Row.EmployeeName) .ToString) 
   End If 
   If Not jobh.ContainsKey (Row.JobLevelHistory) Then {{1 }} maxjobid + = 1 
   jobh.Add (Row.JobLevelHistory, maxjobid) 
   Row.JobId = maxjobid 
   Row.Job2Id = maxjobid 
   Row.DirectRowToJobTitle ( ) 
   Иначе 
   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) {{ 1}} 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 снимите отметку для проверки ограничений).