Взгляд на объектную модель служб SSIS

Службы SQL Server Integration Services (SSIS) – предпочтительное средство извлечения, преобразования и загрузки (ETL) для многих администраторов и разработчиков. Они подключаются к разнообразным источникам и точкам назначения данных и совместимы с любыми преобразованиями данных. Причина такой универсальности — в надежной объектной модели. Объектная модель служб SSIS обеспечивает согласованную инфраструктуру для построения различных компонентов для всевозможных источников данных.

Компоненты потока данных — важная часть инфраструктуры SSIS, так как с их помощью можно подключаться к разнообразным источникам данных, а затем преобразовывать и маршрутизировать данные на высокой скорости. В состав SSIS входят готовые компоненты потока данных, но ваши возможности не ограничиваются работой с ними. Расширяемая объектная модель служб SSIS позволяет создавать собственные настраиваемые компоненты в соответствии с нуждами ETL. Например, можно создать исходный компонент для считывания из источника данных Apache Hadoop, MongoDB, Couchbase или любой базы данных NoSQL. Можно подготовить компоненты SSIS, выполняющие чтение или запись в сторонние «облачные» приложения, такие как Salesforce.com и Workday.

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

Типы компонентов потока данных

Все компоненты потока данных сгруппированы под заголовком Data Flow Task на панели элементов служб SSIS. Компоненты потока данных отличаются от других компонентов SSIS, так как они сгруппированы в три категории.

Компоненты источника. Компоненты источника позволяют подключаться к внешним источникам и считывать данные. Примером может служить компонент источника OLE DB.

Компоненты преобразования. С помощью таких компонентов можно преобразовывать данные. Пример — компонент Sort.

Целевые компоненты. Целевые компоненты позволяют записывать данные во внешние источники. Пример — целевой компонент Excel.

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

Рассмотрим компонент источника OLE DB на рисунке 1. Он имеет один общий вывод и один вывод ошибок. Если попытаться подключить вывод другого компонента к этому компоненту источника OLE DB, то будет возвращена ошибка (см. рисунок 2). Ошибка возникает, так как компоненты источника не могут иметь вводов. Их задача — только читать данные из внешнего источника данных.

Рисунок 1. Компонент источника OLE DB с двумя выводами

Рисунок 2. Сообщение об ошибке при попытке подсоединения к выводу компонента источника OLE DB

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

Синхронные и асинхронные компоненты

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

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

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

Асинхронные компоненты преобразования не выполняют немедленной записи данных в буферы данных. Примером может служить компонент Sort. Он ожидает, пока будут получены все строки из буфера данных, чтобы начать сортировать строки. Затем сортированные строки записываются в новый буфер данных.

Режимы разработки и выполнения

Все компоненты потока данных работают в двух режимах: разработки и выполнения. В режиме разработки все действия (например, редактирование свойств компонента, подключение вводов и выводов) выполняются в конструкторе пакетов SSIS. В режиме выполнения все действия (например, доставка данных, доступ к ним в буфере) осуществляются при выполнении пакета.

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

Базовая структура настраиваемого компонента потока данных

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

public class MySSISComponent:
Microsoft.SqlServer.Dts.Pipeline.PipelineComponent
{
}

Как можно заметить, компонент потока данных, в сущности, представляет собой класс, производный от базового класса Microsoft.SqlServer.Dts.Pipeline.PipelineComponent. Базовый класс PipelineComponent располагает многими механизмами, необходимыми для взаимодействия компонента в режимах разработки и выполнения. Класс PipelineComponent реализует интерфейсы IDTSDesignTimeComponent100 и IDTSRunTimeComponent100. Как видно из названия, интерфейс IDTSRunTimeComponent100 предоставляет методы, вызываемые при выполнении пакета. При построении компонента потока данных необходимо переопределить эти методы и предоставить пользовательские реализации, где необходимо.

На рисунке 3 показано высокоуровневое представление различных классов и интерфейсов, которые являются частью PipelineComponent. Обратите внимание, что сюда не входят объекты и методы, по теме выходящие за рамки данной статьи. Как мы видим, PipelineComponent предоставляет реализации базовых классов различных методов, в том числе AcquireConnections, Validate и PrimeOutput. Все эти методы вызываются конструктором SSIS во время разработки или средой выполнения SSIS в процессе выполнения пакета.

Рисунок 3. Высокоуровневая структура класса PipelineComponent

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

На рисунке 3 обратите внимание на свойство ComponentMetaData, которое реализует интерфейс IDTSComponentMetaData100. Как следует из названия, ComponentMetaData содержит метаданные компонента, в том числе имя и настраиваемые свойства. Самыми важными свойствами являются две коллекции: InputCollection и OutputCollection. В этих массивах хранятся вводы и выводы компонента. При сохранении пакета SSIS в файле объект ComponentMetaData сериализуется в XML и хранится в dtsx-файле.

Вводы и выводы компонента

Каждый компонент потока данных может иметь любое число вводов и выводов. В конструкторе SSIS ввод или вывод представлен стрелкой, направленной в компонент потока данных или из него. В базовом коде, реализующем компонент потока данных, ввод представляет собой объект, реализующий интерфейс IDTSInput100. Вывод — объект, реализующий интерфейс IDTSOutput100. Все элементы, с которыми вы работаете в конструкторе SSIS — вводы, выводы, свойства, выводы ошибки и т.д. — представляют собой объекты, реализующие определенные интерфейсы инфраструктуры SSIS.

Как отмечалось выше, конструктор SSIS распознает эти интерфейсы. Например, обнаруживая объект, реализующий IDTSOutput100, конструктор SSIS обращается к свойству OutputColumnCollection, чтобы получить столбцы в выводе. После того как столбцы получены, конструктору известно, что каждый столбец будет иметь тип IDTSOutputColumn100. Он выясняет значение свойств каждого столбца (например, Length, Precision).

IDTSOutput100 и IDTSInput100 — сложные объекты. В статье подробно рассказывается о IDTSOutput100, но те же принципы действуют и для IDTSInput100.

На рисунке 4 показана объектная модель для IDTSOutput100.

Рисунок 4. Модель объекта IDTSOutput100

Как мы видим, IDTSOutput100 находится наверху иерархической структуры. В нем содержится коллекция выходных столбцов (IDTSOutputColumnCollection100), которая может включать любое число выходных столбцов (IDTSOutputColumn100), в том числе их может не быть вообще (пустая коллекция). Каждый выходной столбец, в свою очередь, содержит свойства столбца (например, DataType, Length, Precision), выделенные в нижней части рисунка 4.

При создании компонента потока данных необходимо задать свойства столбца таким образом, чтобы они соответствовали значениям свойств во внешнем источнике данных или точке назначения. Если этого не сделать, могут возникнуть ошибки из-за несовместимости типов столбцов. Например, если свойству DataType выходного столбца присвоено значение String, а компонент потока данных пытается записать в него значение Int, произойдет ошибка.

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

Buffer. Каждому объекту IDTSOutput100 выделяется объект буфера данных, который используется для передачи данных. В этом свойстве хранится целочисленный идентификатор буфера.

CustomPropertyCollection. Это свойство содержит коллекцию пользовательских свойств для вывода.

ExternalMetadataColumnCollection. Это свойство содержит коллекцию внешних метаданных (более подробно об этом будет рассказано ниже).

SynchronousInputId. Это свойство содержит идентификатор объекта ввода, который доставляет данные в текущий вывод.

Буферы данных

С каждым вводом и выводом для компонента SSIS связан буфер, который представляет собой просто объект типа PipelineBuffer. Как отмечалось, идентификатор буфера данных хранится в свойстве Buffer объекта ввода или вывода. Объект PipelineBuffer представляет собой размещаемый в памяти двухмерный объект со строками и столбцами. Буферы данных — общие для всех компонентов в дереве выполнения в пакете SSIS. Дерево выполнения, в сущности, используется для группирования компонентов в потоке данных. Оно начинается с источника или асинхронного компонента преобразования и заканчивается точкой назначения или другим асинхронным компонентом преобразования. Различные типы компонентов используют PipelineBuffer для разных целей. Например, целевые компоненты считывают данные из PipelineBuffer и записывают их во внешний источник данных. Компоненты источника, наоборот, читают данные из внешнего источника данных и записывают данные в объект PipelineBuffer.

Важно отметить, что столбцы в PipelineBuffer отличаются от входных и выходных столбцов в компоненте, с которым связан PipelineBuffer. Рассмотрим пакет SSIS на рисунке 5. В этом пакете компонент источника OLE DB читает единственный столбец, OLEDB_Col, из внешнего источника данных. Преобразование Character Map добавляет еще один столбец, CMap_Col, который преобразует текст в OLEDB_Col к нижнему регистру. Дерево выполнения начинается компонентом источника OLE DB и завершается компонентом Flat File Destination. Рядом с деревом выполнения показано состояние буфера данных. Обратите внимание, что имена столбцов в буферах данных указаны только для справки. На практике столбцы буфера данных не имеют имен.

Рисунок 5. Сопоставление столбцов объекта PipelineBuffer и столбцов ввода и вывода компонентов

Character Map — синхронное преобразование, то есть оно изменяет буфер, полученный как ввод, и немедленно передает измененный буфер в вывод. Любопытно, что каждый раз, когда среда выполнения SSIS создает буфер данных для компонента источника OLE DB, автоматически вводится дополнительный столбец в буфер данных во время выполнения. Таким образом, когда компонент источника OLE DB начинает записывать данные в буфер, дополнительный столбец уже доступен, но остается пустым. Преобразование Character Map просто приводит значение в OLEDB_Col к нижнему регистру и заполняет им дополнительный столбец.

В результате объект PipelineBuffer для ввода и вывода будет иметь больше столбцов, чем столбцов в вводе и выводе. Например, вывод для компонента источника OLE DB имеет только один столбец (OLEDB_Col), но в соответствующем буфере данных — два столбца. Учитывая, что столбцы в буфере данных не имеют имен, не существует явного способа сопоставить выходной столбец OLEDB_Col столбцу буфера данных. К счастью, класс BufferManager предоставляет метод FindColumnByLineageID. Этот метод обнаруживает столбец в буфере данных на основе свойства LineageID выходного столбца.

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

Образец программного кода для компонента потока данных

Теперь, выяснив принципы объектной модели SSIS, рассмотрим образец программного кода. В листинге 1 показан исходный текст для базового настраиваемого компонента потока данных SSIS. В первую очередь обратите внимание, что класс компонента должен иметь атрибут DtsPipelineComponent. Этот атрибут в основном предоставляет информацию для конструктора SSIS, в частности имя, тип (источник, целевой, преобразование) и описание компонента. Атрибут DtsPipelineComponent необходим для любого настраиваемого компонента SSIS.

Следует отметить еще пространства имен, импортированные с использованием директивы. Такие пространства имен предоставляют различные классы и интерфейсы, составляющие объектную модель SSIS. В приведенной таблице показаны сборки, на которые необходимо ссылаться в проекте, чтобы использовать эти пространства имен.

Кроме того, обратите внимание на инструкции public override. Предполагается, что настраиваемый компонент переопределит методы, унаследованные от PipelineComponent. Эти методы могут быть вызваны несколько раз, как при редактировании компонента в режиме разработки в Visual Studio, так и во время выполнения пакета. Важно четко представлять себе цель каждого метода.

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

В PipelineComponent существует много методов, но некоторые из них имеют особое значение. Самые важные методы этапа разработки и среды исполнения — AcquireConnections, ProvideComponentProperties, ReinitializeMetaData и Validate. Самые важные методы только среды исполнения — PreExecute, PrimeOutput и ProcessInput. Рассмотрим каждый из этих методов.

Метод AcquireConnections

Большинство компонентов источника спроектированы для чтения данных из источника данных, такого как базы данных SQL Server или файлы Excel. Цель метода AcquireConnections — установить соединение с источником данных и сохранить соединение в локальной переменной. Например, если компонент подключен к базе данных SQL Server, можно создать объект SqlConnection и использовать AcquireConnections для организации соединения, как показано в листинге 2.

Главная цель этого метода — «кэшировать» объект соединения с внешним источником данных. Кроме того, если объект соединения по какой-то причине имеет значение null, метод AcquireConnections должен повторно установить соединение с внешним источником данных.

Метод ProvideComponentProperties

Метод ProvideComponentProperties вызывается лишь однажды — когда компонент впервые добавляется к пакету в конструкторе SSIS. Этот метод используется для добавления вводов, выводов и пользовательских свойств к компоненту.

Обратите внимание, что при разработке компонента источника нет необходимости добавлять вводы. Аналогично, нет необходимости добавлять выводы для целевых компонентов. В первую очередь рекомендуется удалить вводы, выводы и пользовательские свойства, уже присоединенные к компоненту (то есть начать с чистого листа). Затем добавьте новые вводы, выводы и пользовательские свойства. Например, программный код в листинге 3 добавляет к компоненту единственный ввод и пользовательское свойство.

На рисунке 6 показано, как будет выглядеть этот компонент в конструкторе SSIS.

Рисунок 6. Компоненты, созданные кодом листинга 3

При работе с компонентом в конструкторе SSIS непрерывно выполняется проверка компонента. Это означает, что конструктор SSIS вызывает метод Validate в фоновом режиме.

Как видно из названия, цель метода Validate — проверить пакет и обнаружить любые отклонения. Важно понять, что необходимо обеспечить логику проверки. Это может быть просто проверка вводов и выводов. Например, исходный текст в листинге 4 проверяет компонент источника данных, который должен иметь один вывод и не иметь выводов.

Метод Validate возвращает перечисление DTSValidationStatus. Существует четыре возможных возвращаемых значения.

VS_ISVALID. Компонент правильно настроен и готов к выполнению.

VS_ISBROKEN. Компонент настроен неправильно. Как правило, это означает, что значение свойства задано неверно.

VS_ISCORRUPT. Компонент необратимо поврежден и должен быть полностью сброшен. В ответ конструктор вызывает метод ProvideComponentProperties компонента.

VS_NEEDSNEWMETADATA. Метаданные компонента устарели или испорчены. Конструктор вызывает метод ReinitializeMetaData компонента для восстановления компонента.

В листинге 4, если условия проверки не выполнены, то метод Validate возвращает значение VS_ISCORRUPT. Кроме того, выводится соответствующее сообщение об ошибке.

Рассмотрим подробнее, когда следует использовать значение VS_NEEDSNEWMETADATA при проектировании программного кода проверки. Предположим, имеется компонент источника, который выполняет чтение из таблицы базы данных и составляет коллекцию на основе столбцов исходной таблицы, как показано на рисунке 7.

Рисунок 7. Исходный компонент, который строит коллекцию  выходных столбцов на основе столбцов исходной таблицы

Если удалить один из столбцов в исходной таблице, то компонент должен также удалить соответствующий выходной столбец из коллекции выходных столбцов. В этом случае метод Validate должен располагать логикой для выявления этой аномалии. Если есть рассогласование между столбцами в исходной таблице и выходными столбцами компонента, то метод Validate должен возвратить значение VS_NEEDSNEWMETADATA. Но у такой проверки есть недостаток: при каждом вызове метод Validate должен повторно устанавливать соединение с источником внешних данных или точкой назначения. Это может чрезвычайно замедлить процесс проектирования компонента, особенно если внешний источник данных или точка назначения размещаются в «облаке».

Объектная модель SSIS позволяет сократить затраты, создавая коллекцию метаданных, которая наследует от IDTSExternalMetadataColumnCollection100. Эта коллекция столбцов должна отражать столбцы из внешнего источника данных в том смысле, что атрибуты (например, DataType, Length, Name) должны соответствовать внешнему источнику данных. После заполнения этой коллекции компонент сопоставит входные и выходные столбцы с внешней коллекцией метаданных без подключения к внешнему источнику данных. Подключение к внешнему источнику данных будет выполняться лишь периодически, чтобы удостовериться, что столбцы во внешней коллекции метаданных верны. Подготовка программного кода для построения внешней коллекции метаданных не рассматривается в данной статье, но необходимая логика высокого уровня показана на рисунке 8.

Рисунок 8. Проверка входных и выходных столбцов по?коллекции внешних метаданных

Метод PreExecute

Как отмечалось в разделе «Буферы данных», настраиваемый компонент SSIS должен обеспечивать сопоставление между столбцами в объекте PipelineBuffer и столбцами в вводе и выводе компонента. Рекомендуется использовать метод PreExecute для такого сопоставления.

PreExecute — метод только среды исполнения, который вызывается один раз при каждом выполнении. Он вызывается непосредственно перед основными методами, когда самое время произвести некоторые вспомогательные операции, в том числе определить сопоставление между входными и выходными столбцами и столбцами буфера данных. Например, исходный текст в листинге 5 формирует структуру ColumnInfo, которая сопоставляет каждый выходной столбец в компоненте соответствующему индексу столбца в объекте PipelineBuffer. В этом программном коде ключевой метод — FindColumnByLineageID. На основании идентификатора LineageID выходного столбца метод FindColumnByLineageID находит соответствующий столбец в объекте PipelineBuffer.

Метод PrimeOutput

Единственное назначение метода PrimeOutput — добавлять строки данных в выходные буферы. Метод только среды исполнения вызывается для компонентов источника и асинхронных компонентов преобразования. Он не используется для целевых компонентов, так как у них нет выводов. Он не применяется для синхронных компонентов преобразования, так как они не добавляют новых строк в выходные буферы.

У метода PrimeOutput существует три обязательных компонента:

(int) outputs. Этот параметр указывает число выводов в компоненте.

Int [] outputIDs. Этот параметр представляет собой массив, содержащий идентификаторы выводов компонента.

PipelineBuffer[] buffers. Этот параметр представляет собой массив выходных буферов. Для каждого вывода имеется один буфер.

В листинге 6 приведен пример реализации метода PrimeOutput. Сначала данные считываются и загружаются в обычную. NET DataTable. Затем данные строка за строкой добавляются в объект PipelineBuffer.

Обратите внимание на использование структуры ColumnInfo, которая была сформирована в образце программного кода PreExecute в листинге 5. Метод PrimeOutput задействует ее для сопоставления столбца в буфере с соответствующим столбцом в DataTable. В конце программного кода вызывается метод SetEndOfRowset, указывающий, что все строки данных успешно добавлены.

Метод ProcessInput

Метод ProcessInput предоставляет компонентам объект PipelineBuffer. С помощью этого метода строки в буфере перебираются и обрабатываются компонентом. Для целевых компонентов строки в буфере записываются во внешнюю точку назначения. Для асинхронных компонентов преобразования строки обрабатываются в соответствии с логикой преобразования и записываются в новый буфер данных для передачи нижележащим компонентам.

Отдельный столбец в буфере можно прочитать или записать одним из двух способов. Один подход — использовать метод доступа индексатора массива с оператором []. Иначе можно задействовать один из методов Get или Set. Методы Get и Set более эффективны; их следует применять, когда известен тип данных столбца в буфере. В листинге 7 метод ProcessInput обрабатывает входящие столбцы с использованием обоих подходов.

Ядро SSIS

Объектная модель представляет собой ядро SSIS. Как таковое, оно обеспечивает надежный фундамент для построения различных типов ETL-компонентов. В этой статье показаны принципы, которые необходимо знать, чтобы создать настраиваемый компонент потока данных. В следующей статье я расскажу о том, как применить эти принципы к проектированию настраиваемого компонента потока данных, который подключается к источнику данных и читает из него данные.

using Microsoft.SqlServer.Dts.Pipeline;
using Microsoft.SqlServer.Dts.Pipeline.Wrapper;
using Microsoft.SqlServer.Dts.Runtime;
using Microsoft.SqlServer.Dts.Runtime.Wrapper;
using System.IO;
namespace Microsoft.Samples.SqlServer.Dts
{
[DtsPipelineComponent(DisplayName = «SampleComponent»
, ComponentType = ComponentType.SourceAdapter)]
public class MySSISComponent: PipelineComponent
{
public override void ProvideComponentProperties(){; }
public override
void AcquireConnections(object transaction)
{; }
public override IDTSCustomProperty100
SetComponentProperty(string propertyName
, object propertyValue)
{; }
public override DTSValidationStatus Validate(){; }
public override void ReinitializeMetaData(){; }
public override void PreExecute(){; }
public override void PrimeOutput
(int outputs
, int[] outputIDs
, PipelineBuffer[] buffers)
{; }
}
}

namespace Microsoft.Samples.SqlServer.Dts
{
[DtsPipelineComponent
(DisplayName = «SampleComponent»
, ComponentType = ComponentType.SourceAdapter)]
public class MySSISComponent: PipelineComponent
{
private SqlConnection sqlConn;
public override void
AcquireConnections(object transaction)
{
if (sqlConn == null)
{
sqlConn = new SqlConnection(connectionString);
.
.
}
}
}
}

{
// Удаление всех существующих вводов, выводов и
// пользовательских свойств.
base.RemoveAllInputsOutputsAndCustomProperties();
// Добавление пользовательского свойства.
IDTSCustomProperty100 logFilePath =
ComponentMetaData.CustomPropertyCollection.New();
logFilePath.Description = «Log File Path»;
logFilePath.Name = «Log File Path»;
logFilePath.Value = String.Empty;
// Добавление одного ввода с именем «Output».
IDTSOutput100 output
= ComponentMetaData.OutputCollection.New();
output.Name = «Output»;
}

public override DTSValidationStatus Validate()
{
// Проверка #1. Убедитесь, что имеется только один вывод
// и ни одного ввода.
bool pbCancel = false;
if (ComponentMetaData.OutputCollection.Count! = 1
|| ComponentMetaData.InputCollection.Count! = 0)
{
ComponentMetaData.FireError(0, ComponentMetaData.Name
, «The component should have only one output and no inputs.»
, "", 0, out pbCancel);
return DTSValidationStatus.VS_ISCORRUPT;
}
}

public struct ColumnInfo
{
public int BufferColumnIndex;
public String columnName;
}
public override void PreExecute()
{
IDTSOutput100 output =
ComponentMetaData.OutputCollection[0];
foreach (IDTSOutputColumn100 outputCol
in output.OutputColumnCollection)
{
ColumnInfo ci = new ColumnInfo();
ci.columnName = outputCol.Name;
ci.BufferColumnIndex
= BufferManager.FindColumnByLineageID(output.Buffer,
outputCol.LineageID);
colInfo.Add(ci);
}
}

public override void PrimeOutput
(int outputs
, int[] outputIDs
, PipelineBuffer[] buffers)
{
PipelineBuffer buffer = buffers[0];
IDTSOutput100 output =
ComponentMetaData.OutputCollection[0];
// Вызов метода извлечения данных.
DataTable dt = GetDataTable();
foreach (DataRow row in dt.Rows)
{
buffer.AddRow();
// Местонахождение индекса столбца в буфере данных.
foreach (ColumnInfo ci in colInfo)
{
buffer[ci.BufferColumnIndex] = row[ci.columnName];
}
}
buffer.SetEndOfRowset();
}

public override void ProcessInput
( int InputID
, PipelineBuffer buffer)
{
List lst = new List();
while( buffer.NextRow())
{
for(int x=0; x < inputColumns.Length; x++)
{
if(!buffer.IsNull(inputColumns[x]))
{
String str1 = (String) buffer[inputColumns[x]];
String str2 = buffer.GetString(inputColumns[x]);
lst.add(str1);
}
}
}
// Записать lst во внешнее целевое местонахождение данных.
.
.
.
}