<Disclaimer: вероятно этот пост будет в последующем редактироваться. Пока публикуется так, как он есть>
Как это полагается в старинных былинах, начну издалека..
Подготовка эксперимента, постановка задачи и т.д.
Экспериментируя с разными задачами, вдруг наткнулся на интересную идею – надо бы организовать параллельную выкачку данных из таблицы в одном ресурсе в таблицу на другом. Фактически, это ситуация, которая достаточно часто встречается в разного рода проектах: есть временная «куча», куда данные заливаются, их надо перенести в «постоянную таблицу» . Потому «куча» будет просто «обнулена».. Целевая таблица может иметь, допустим, кластерный индекс на ней.
Проанализировав различные подходы, решил, что надо использовать технологии на полную катушку и параллельной выкачкой у меня будет заниматься SSIS.
Как уже упоминал, изначальная таблица – куча. Для простоты объяснения ситуации, предположим, что таблица у нас типа:
USE Sample
CREATE TABLE SourceTable (Range int, Value int)
В реальной жизни же, конечно, такими простыми таблицами не отделаешься, но на данный момент не это важно. Важно, чтобы у нас была возможность разбить таблицу на равные диапазоны. В данном примере роль диапазонов играет столбец Range, где у нас «забиты» диапазоны от 0 до 1000, 1001 до 2000, и т.д. до 10 000.
Поле Value пусть будет хранить какие-нибудь числа. Это сейчас не так важно.
Итак, SSIS пакет запустится, начнет «вытаскивать» данные из таблицы и… Вообще-то идеально было бы иметь один пакет, который можно запустить несколько раз, или несколько одинаковых пакетов, но чтобы они крутились параллельно и «читали таблицу. То есть им надо каким-то образом указать, какие данные надо выкачать. Для этой цели была сделана еще одна служебная таблица [dbo].[work_queue], которая выглядит приблизительно вот так:
|
work_id |
group |
assignment |
is_active |
is_done |
|
1 |
SSISLoad |
1000 |
0 |
0 |
|
2 |
SSISLoad |
2000 |
0 |
0 |
|
3 |
SSISLoad |
3000 |
0 |
0 |
|
4 |
SSISLoad |
4000 |
0 |
0 |
|
5 |
SSISLoad |
5000 |
0 |
0 |
|
6 |
SSISLoad |
6000 |
0 |
0 |
|
7 |
SSISLoad |
7000 |
0 |
0 |
|
8 |
SSISLoad |
8000 |
0 |
0 |
|
9 |
SSISLoad |
9000 |
0 |
0 |
То есть, мы задаем идентификатор «работы» (work_id) и диапазон, который должен будет сканироваться (assignment). Для случая. Когда несколько разных типов «работ» будет существовать, можно прописать , какой группе они принадлежат. В данном примере ограничимся одной группой.
Далее, как только диапазон взят «в оборот», поле is_active должно смениться с 0 на 1. Ну и по окончании работы поле is_done тоже сменится с 0 на 1.
Работу по изменению этих полей передадим процедуре:
CREATE PROCEDURE [dbo].[sp_get_work]
@group NVARCHAR(128) = NULL /* The group to get*/
, @work_id INT = NULL OUTPUT /* the work_id assigned */
, @assignment NVARCHAR(MAX) = NULL OUTPUT /* The description of the work to do */
Как видно из кода, процедура возвращает @assignment, т.е. диапазон, который надо «вытащить» из таблицы. Что нам пригодиться потом в SSIS пакетах.
Для того, чтобы данные «дообработать», можно создать некоторое количество таблиц ( по количеству пакетов SSIS, которые будет обрабатывать таблицу).
Собственно, вступительная часть закончена.. Тепрь приступаем к той части, ради которой весь сыр-бор.
Передача переменной внутри SSIS пакета.
Для начала создаем новый проект, в нем новый SSIS пакет. Допустим, так как LoadStaiging1.dstx.
Далее определяем в Control Flow последовательность выполнения задач. Певрое, что мы сделаем – убедимся, что временная таблица, куда мы перетягиваем данные - пуста .
Рис.1 «Обнуление» данных в таблице
Как видно в строке SQLStatement можно прописать совершенно спокойно обычный T-SQL запрос. Чем мы и воспользуемся.
Далее, если еще кто-то помнит, что было во вступительной части – мы создали процедуру, которая фактически возвращает диапазон, который «можно» выкачивать из таблицы и, как –только мы начинаем с диапазоном работать, помечает его как is_active. Так вот для этих целей в пакете можно описать переменные, действующие на протяжении всего пакета. Некоторым из них можно присвоить значения, которые будут использоваться по умолчанию каждый раз, когда пакет запускается.
Рис.2 Опеределение переменных.
Следующей задачей в пакете будет как раз выяснение того, какой диапазон у нас сейчас «свободен». Для этого я взял опять Execute SQL Task.
Рис 3. Execute Task - выяснение, какой диапазон еще не в работе
Как видно из меню, я выбрал в Connection Type ADO.NET соединение. Прелесть именно этого типа, что с ним удобней работать с переменными. Удобность вот в чем:
Вот наш запрос из SQL Statement
EXEC [Sample].[dbo].[sp_get_work] @group, @work_id, @assignment OUTPUT
Достаточно читабельный код, не так ли? В случае OLE DB аналогичный код выглядел бы вот так:
EXEC [Sample].[dbo].[sp_get_work] ?, ?, ? OUTPUT
Меня такой код немного смущает.
Теперь, надо сопоставить переменные внутри этой задачи с теми, что мы определили на уровне пакета.
Рис.4 "Связывание" переменных и параметров
То есть: User::work_id соответствует параметру @work_id,
User::group соответствует параметру @group,
User::assignment соответствует параметру @assignment.
User::assignment в поле Direction указано Output, что достаточно логично указывает на то, что этот параметр будет передан дальше.
«Дальше» означает следующий Execute SQL Task, где в поле SQLStatement будет прописан следующий код:
INSERT INTO Staging1 ([RANGE], [VALUE]) SELECT ST.[RANGE], ST.[VALUE] from Sample.dbo.SourceTable ST where ST.[Range] = @assignment
Где @assignment опять сопоставляется с переменной User::assignment
Вот собственно и все.
Точнее почти все. Я рассказал о том, как параметр передается между Execute SQL Task.
Далее, результат выполнения последнего запроса передается на вход Document Flow.
В моей конкретной задаче мне надо было сравнить, совпадают ли данные во временной таблице с данными в результирующей таблице и те, которые не совпадают перенести в результирующую таблицу.
Это было сделано с помощью достаточно простых манипуляций, которые в SSIS выглядят приблизительно вот так:
