Завершая наш рассказ о службах преобразования данных, мы остановимся на трех прикладных вопросах использования пакетов DTS, именно: как включить задачу в транзакцию, как программным образом перебрать пакеты и как заставить шаг выступить поставщиком данных для внешнего потребителя.
Шаги определяют последовательность выполнения задач пакета. Как только речь заходит о последовательности действий, встает вопрос о том, нельзя ли рассматривать эту последовательность как единую транзакцию. Шаги обладают свойствами, позволяющими объединять их в транзакцию и в зависимости от результата работы шага фиксировать, либо откатывать результаты предшествующих шагов. Как и все распределенные транзакции, транзакции внутри пакета DTS выполняются с помощью MS DTC, поэтому, если Вы собираетесь их использовать, проследите, чтобы его сервис был стартован (snap-in SQL Server -> Support Services -> Distributed Transaction Coordinator). Для того, чтобы соединение, с которым работает задача в шаге, понимало команды фиксации и отката транзакции и могло наследовать контекст внешней транзакции пакета, требуется, чтобы OLE DB-провайдер данного соединения поддерживал интерфейс IТtransactionJoin. В противном случае при выполнении пакета Вы получите сообщение об ошибке: Connection … for task … doesn't support joining distributed transactions or failed when attempting to join. No such interface supported. Кроме того, несмотря на то, что в ходе данного конкретного выполнения пакета несколько шагов могут выполняться параллельно, в пакете в каждый момент времени допускается одна и только одна транзакция. Таково ограничение текущей версии DTS. Это не значит, что в пакете не может быть несколько транзакций, это означает лишь, что в настоящее время они могут выполняться лишь последовательно, друг за другом. Следовательно, когда мы устанавливаем workflow в пакете, необходимо внимательно следить, чтобы ни при каких исходах и ветвлениях логики наши последовательные транзакции не наложились друг на друга. Для разрешения транзакций внутри пакета нужно отметить опцию Use Transactions в диалоговом окне свойств пакета, закладка Advanced. Там же задается уровень изоляции транзакций (Transaction Isolation Level), по умолчанию он установлен в Read Committed. Объединение шагов в транзакции осуществляется с помощью контекстного меню задачи шага, пункт Workflow Properties, закладка Options, опция Join Transaction If Present. Завершение транзакции на данном шаге с фиксацией или откатом изменений данного и предыдущих шагов транзакции управляется опциями Сommit transaction on successful completion of this step и Rollback transaction on failure. Предположим, мы отметили обе эти опции и данный шаг состоит из ActiveX Script Task. Тогда при Main =DTSTaskExecResult_Failure транзакция будет откачена, при Main = DTSTaskExecResult_Success - фиксирована. Тех же результатов можно добиться программным путем, используя практически одноименные свойства объектов Package и Step.
Следующая практическая задача, о которой мне бы хотелось поговорить, - это задача перебора пакетов. Ошибается тот, кто считает, что для этого достаточно пробежаться по какой-нибудь коллекции вроде oSrv.DTS.Packages. Чтобы понять, почему все не так просто, вспомним, что существует три способа хранения пакетов. Работа с пакетом в приложении осуществляется так: в проекте делается ссылка на Microsoft DTSPackage Object Library, создается пустой объект типа пакет, после чего одним из трех методов в зависимости от способа хранения происходит загрузка экземпляра интересующего нас пакета.
Private Sub cmdLoadPkg_Click()
Dim dtsPkg As New dts.Package, i
Select Case lstStorageType.ListIndex
Case 0 'На сервере как локальный пакет
dtsPkg.LoadFromSQLServer ServerName:=txtSrvName.Text, _
Flags:=DTSSQLStgFlag_UseTrustedConnection, _
PackageName:=cmbPkgName.Text
Case 1 'Репозиторий
dtsPkg.LoadFromRepository RepositoryServerName:=txtSrvName.Text, _
RepositoryDatabaseName:="msdb", _
RepositoryUserName:="dbo", RepositoryUserPassword:="", _
PackageID:="", _
Flags:=DTSReposFlag_UseTrustedConnection, _
PackageName:=cmbPkgName.Text
'dtsPkg.LoadFromRepository
Case 2 'В виде СОМ-объекта (.dts-файл)
dtsPkg.LoadFromStorageFile UNCFile:=txtFileName.Text, Password:=""
End Select
EnumPkgCnns dtsPkg
End Sub
Cама форма приведена на рис.10. Мы выбираем способ хранения пакета, выбираем из комбо "Имя пакета" интересующий нас пакет, нажимаем на кнопку "Загрузить пакет в приложение" и в нижнем окне получаем перечень всех соединений внутри пакета. Как происходит загрузка пакета в приложение, мы уже видели из только что приведенного фрагмента кода. Вообще говоря, как мы помним, репозиторные и СОМовские пакеты отличаются не только своими именами, но и версиями внутри одного имени, так что при загрузке необходимо было еще указывать GUID версии. Однако это слишком усложнит наше модельное приложение, поэтому простоты ради будем считать, что: а) внутри каждого пакета находится не более одной версии; б) в одном dts-файле хранится только один пакет. Основным идейным моментом нашего приложения является загрузка списка имен серверных или репозиторных пакетов в комбо. Для перебора локальных пакетов применяется хранимая процедура msdb..sp_enum_dtspackages. Перебор репозиторных пакетов предсталяет собой менее тривиальную задачу, так как подобная процедура для репозитория по понятным причинам отстутствует. Для ее решения я выбрал несколько хакерский путь решения, отследив в SQL Profiler, какие команды поступают на SQL Server, когда мы броузим репозиторий через ММС. После "доработки напильником" родилась следующая "самопальная" хранимая процедура
CREATE PROCEDURE sp_enum_dtspkgs_in_repository AS
set nocount on
create table #MyRTblObjColTemp ( Z_ColNum_Z int NOT NULL DEFAULT 0, IntID binary(8) NOT NULL, Z_BranchID_Z int NOT NULL DEFAULT 0, Z_VS_Z int NOT NULL DEFAULT 0, TypeID binary(8) NOT NULL DEFAULT 0x0000000000000000)
Insert into #MyRTblObjColTemp (IntID) select distinct IntID from RTblVersions A, RTblRelships B where B.DstID = 0x13000000CA320000 and B.RelTypeID = 0x000000001A000000 and A.TypeID = B.OrgID
update #MyRTblObjColTemp set Z_ColNum_Z = 1, Z_BranchID_Z = A.Z_BranchID_Z, Z_VS_Z = A.Z_VS_Z, TypeID = A.TypeID from RTblVersions A, #MyRTblObjColTemp B where A.IntID = B.IntID and B.Z_ColNum_Z = 0 and A.Z_LClock_Z in (Select Max(Z_LClock_Z) from RTblVersions where IntID = A.IntID)
SELECT distinct b.IntID,b.Z_BranchID_Z,b.Z_VS_Z,b.Z_VE_Z,b.Name from RTblVersions a, #MyRTblObjColTemp d, RTblNamedObj b where a.IntID = d.IntID and b.IntID = d.IntID and a.Z_VS_Z >= b.Z_VS_Z and b.Z_VE_Z >= a.Z_VS_Z and d.Z_ColNum_Z = 1
return 0
Процедура была положена в msdb на нескольких серверах и благополучно доказала свою работоспособность, когда Брайан Уэлкер (Brian Welcker) предложил использовать ODBC-шлюз к репозиторию - библиотека Microsoft Repository (\Program Files\Common Files\Microsoft Shared\Repostry\repodbc.dll). Несмотря на то, что в этом случае доступ через ODBC осуществляется несколько медленнее, чем через OLE DB, идея показалась мне более изящной, так как допускает размещение репозитория не только на SQL Server. Кроме того, мы будем использовать в проекте Microsoft ActiveX Data Objects 2.0 Library (d:\Program Files\Common Files\System\ADO\msado20.tlb) и элемент управления Data Bound Combobox из Microsoft DataList Controls 6.0 (OLEDB) (\Winnt4EE\System32\msdatlst.ocx). Для наполнения его списка именами пакета предлагается следующий способ:
Const OBJID_IDtsTransformationPackage = "{{EBB9995C-BA15-11d1-901B-0000F87A3B33},000032CA}"
.....................................................................
Sub EnumDTSPackages()
Dim adoRS As ADODB.Recordset
Set adoRS = New ADODB.Recordset
Select Case lstStorageType.ListIndex
Case 0 'На сервере как локальный пакет
adoRS.ActiveConnection = "Provider = SQLOLEDB.1; Integrated Security = SSPI;Persist Security Info = False; Initial Catalog = msdb; Data Source = " & txtSrvName.Text
adoRS.CursorLocation = adUseClient
adoRS.CursorType = adOpenStatic
adoRS.Source = "sp_enum_dtspackages"
adoRS.Open , , , , adCmdStoredProc
Case 1 'Репозиторий
Dim Rep As New RepositoryTypeLib.Repository
Dim IFaceDef As RepositoryTypeLib.InterfaceDef
Dim Pkg As RepositoryTypeLib.RepositoryObject
Rep.Open ("SERVER=" & txtSrvName & "; DATABASE = msdb; Integrated Security = SSPI")
Set IFaceDef = Rep.object(OBJID_IDtsTransformationPackage)
adoRS.Fields.Append "Name", adVarChar, 50
adoRS.Open
For Each Pkg In IFaceDef.ObjectInstances
adoRS.AddNew "Name", Pkg.Name
Next
End Select
Set cmbPkgName.RowSource = adoRS
cmbPkgName.ListField = "name"
If adoRS.EOF Then
cmbPkgName.Text = ""
Else
cmbPkgName.Text = adoRS(cmbPkgName.ListField).Value
End If
End Sub
Заметим, что никакого ADO Data Control для нашего DataBound Combo в данном случае не требуется. Генерация множества записей в качестве источника данных происходит для него динамически через объект ADO.Recordset.
После того, как пакет загружен, мы получаем доступ из приложения ко всем его свойствам и коллекциям рассматривавшихся нами выше соединений, задач и шагов. Предположим, мы хотим в SQL Server создать прилинкованный сервер для Clipper. Какой провайдер при этом использовать и как его при этом инициализировать, нам неизвестно, но мы знаем, что в состав DTS входит предопределенное соединение с форматом dBase 5, совместимое с файлами CA-Clipper 5.3. Создаем пакет с фиктивным соединением такого типа, и с помощью данного приложения просматриваем коллекцию соединений этого пакета, в частности, свойства ProviderID и Extended Properties.
Sub EnumPkgCnns(dtsPkg As dts.Package)
Dim dtsCnn As dts.Connection
txtPkgCnns.Text = ""
For Each dtsCnn In dtsPkg.Connections
txtPkgCnns.Text = txtPkgCnns.Text & dtsCnn.ProviderID & vbTab & dtsCnn.DataSource & vbTab & dtsCnn.ConnectionProperties("Extended Properties") & vbNewLine
Next
End Sub
Для соединения dBase 5 c помощью нашего приложения получаем: ProviderID = Microsoft.Jet.OLEDB.4.0; DataSource = D:\TEMP\HetQueries; Extended Properties = dBase 5.0. Отсюда с очевидностью создание прилинкованного сервера для dBase 5 / Clipper:
if exists (select srvname from master.dbo.sysservers where srvname='MyDBF')
exec sp_dropserver @server='MyDBF', @droplogins='droplogins'
go
exec sp_addlinkedserver @server='MyDBF', @srvproduct='dBase 5', @provider='Microsoft.Jet.OLEDB.4.0', @datasrc='D:\TEMP\HetQueries', @provstr='dBase 5.0'
go
exec sp_addlinkedsrvlogin @rmtsrvname='MyDBF', @useself=false, @locallogin=NULL, @rmtuser='Admin', @rmtpassword=''
После чего можно работать со всеми dbf-таблицами в каталоге D:\TEMP\HetQueries:
sp_tables_ex MyDBF
go
select * from MyDBF...clp_tbl
Таким образом мы плавно подошли к нашему третьему вопросу: интеграции службы преобразования данных и гетерогенных запросов. Предполагается, что читатели в той или иной мере знакомы с механизмом гетерогенных запросов. Желающие могут обратиться, в частности, к журналу "СУБД", №3 за 1998 г., статья называется "Процессор запросов SQL Server 7.0. Некоторые стратегии оптимизатора при построении сложных, параллельных и распределенных планов". С помощью OLE DB-провайдера DTSPackageDSO пакет DTS или отдельный шаг внутри пакета может выступать в качестве источника данных по отношению к внешнему потребителю. Указанный провайдер на самом деле является оберткой, которая запускает пакет (dtsrun в режиме in-process) и возвращает полученное на выходе заданного шага множество записей. Все вызовы, которые потребитель делает к IRowset провайдера тот на самом деле тут же переправляет к множеству записей, полученному от пакета, т.е. играет роль proxy. Для того, чтобы шаг возвращал наружу результаты своей деятельности, нужно присвоить Step. IsPackageDSORowset=true. В DTS Designer это делается с помощью контекстного меню задачи шага -> Workflow Properties -> закладка Options -> отметить DSO Rowset Provider. При этом во время обычного выполнения пакета данный шаг перестает срабатывать, т.е. он переносит в назначение 0 записей. Если требуется, чтобы шаг срабатывал и в то же время служил как источник данных для внешних потребителей, можно создать его копию с IsPackageDSORowset=false на параллельной ветви. Множество записей из пакета можно получить через интерфейсы IOpenRowset и ICommand. Первый способен возвращать только одно множество записей. Если шаг указан, он трактуется как таблица, если нет, пакет выполняется до первого шага, возвращающего результат наружу. Если несколько шагов пакета имеют IsPackageDSORowset=true, можно воспользоваться ICommand и вернуть в приложение сразу несколько множеств записей за один раз, к которым можно обращаться через IMultipleResults::GetResult. Пример создания прилинкованного сервера для DTS-пакета:
exec sp_addlinkedserver @server='MyDTS', @srvproduct='PDSO', @provider='DTSPackageDSO', @datasrc='/E /S /NDemoPack: SQL->dbf'
Обратим внимание, что для источника данных употребляется строка, аналогичная той, которая применяется в утилите dtsrun. В данном примере мы создали прилинкованный сервер из пакета с именем DemoPack: SQL->dbf, находящийся на локальном сервере (пустая строка после ключа /S), на который мы ходим под безопасностью Windows NT (ключ /E). Пакет совершает перенос таблицы pubs..employee из SQL Server в dBase 5. В опциях задачи переноса (Workflow Properties) отмечено DSO Rowset Provider. До этого мы убедились в том, что пакет действительно работает. Тогда select * from openquery(MyDTS, 'select *') дает клиенту в точности то множество записей, которое должна получить таблица назначения. Аналогичный результат можно получить без предварительного создания прилинкованного сервера:
SELECT * FROM OPENROWSET('DTSPackageDSO', '/E /S /NDemoPack: SQL->dbf', 'Select *')
Схожим образом осуществляется работа с выделенной версией того или иного пакета. Перенесем наш пакет в репозиторий (Package->Save As, Location=SQL Server Repository), посмотрим GUID версии (папка Data Transformation Services в ММС -> Metadata -> закладка Package в правой части экрана -> раскрыть пакет с нужным названием -> кликнуть на версию за интересующую дату) и перепишем предыдущий запрос так:
SELECT * FROM OPENROWSET('DTSPackageDSO', '/E /S /R /NDemoPack: SQL->dbf /V {B50FA41D-DEF5-11D2-BECD-C05A4A000000}', 'Select *')
Представим себе, что каждое преобразование в пакете, показанном на рис.6, помечено как DSO Rowset Provider, нас же интересует результат только одного из них. Загрузим пакет из файла в DTS Designer, выбрав версию за нужную нам дату. В свойствах пакета посмотрим Version GUID. В Workflow Properties найдем имя интересующего шага. Окончательный запрос может выглядеть так:
SELECT * FROM OPENROWSET('DTSPackageDSO', '/FD:\Temp\Olap\NorthwindMart_Fill.dts /V{10E05829-C260-11D2-BE93-0020AFB67169}', 'Select * from DTSStep_DTSDataPumpTask_4'),
где /V характеризует версию пакета, а DTSStep_DTSDataPumpTask_4 - имя шага.