Developing a Custom Source Component

Składniki urządzenie źródłowe nawiązywania połączenia z zewnętrznymi źródłami danych przy użyciu ConnectionManager obiekty zdefiniowane w pakiet. Wskazują one ich zapotrzebowania dla menedżer połączeń przez dodanie elementu RuntimeConnectionCollection() Kolekcja ComponentMetaData() Właściwość. Ta kolekcja obsługuje dwóch celach — do przechowywania odwołań do menedżer połączeń w pakiet używanych przez składnik i anonsowanie potrzebę menedżer połączeń, który ma być projektanta.Gdy IDTSRuntimeConnection100 został dodany do kolekcja, Zaawansowany edytor Wyświetla Właściwości połączenia kartę, która pozwala użytkownikom na wybranie lub utworzenie połączenia w pakiecie.

Poniższy przykład kodu pokazuje implementację ProvideComponentProperties() dodaje danych wyjściowych i dodaje IDTSRuntimeConnection100 obiekt do RuntimeConnectionCollection().

using System;
using System.Collections;
using System.Data;
using System.Data.SqlClient;
using System.Data.OleDb;
using Microsoft.SqlServer.Dts.Runtime;
using Microsoft.SqlServer.Dts.Runtime.Wrapper;
using Microsoft.SqlServer.Dts.Pipeline;
using Microsoft.SqlServer.Dts.Pipeline.Wrapper;

namespace Microsoft.Samples.SqlServer.Dts
{
    [DtsPipelineComponent(DisplayName = "MySourceComponent",ComponentType = ComponentType.SourceAdapter)]
    public class MyComponent : PipelineComponent
    {
        public override void ProvideComponentProperties()
        {
            // Reset the component.
            base.RemoveAllInputsOutputsAndCustomProperties();
            ComponentMetaData.RuntimeConnectionCollection.RemoveAll();

            IDTSOutput100 output = ComponentMetaData.OutputCollection.New();
            output.Name = "Output";

            IDTSRuntimeConnection100 connection = ComponentMetaData.RuntimeConnectionCollection.New();
            connection.Name = "ADO.NET";
        }

Imports System.Data
Imports System.Data.SqlClient
Imports Microsoft.SqlServer.Dts.Runtime
Imports Microsoft.SqlServer.Dts.Runtime.Wrapper
Imports Microsoft.SqlServer.Dts.Pipeline
Imports Microsoft.SqlServer.Dts.Pipeline.Wrapper

<DtsPipelineComponent(DisplayName:="MySourceComponent", ComponentType:=ComponentType.SourceAdapter)> _
Public Class MySourceComponent
    Inherits PipelineComponent

    Public Overrides Sub ProvideComponentProperties()

        ' Allow for resetting the component.
        RemoveAllInputsOutputsAndCustomProperties()
        ComponentMetaData.RuntimeConnectionCollection.RemoveAll()

        Dim output As IDTSOutput100 = ComponentMetaData.OutputCollection.New()
        output.Name = "Output"

        Dim connection As IDTSRuntimeConnection100 = ComponentMetaData.RuntimeConnectionCollection.New()
        connection.Name = "ADO.NET"

    End Sub
End Class

Po dodaniu do połączenia RuntimeConnectionCollection(), można zastąpić AcquireConnections(Object) Metoda nawiązanie połączenia z danymi zewnętrznymi urządzenie źródłowe. Ta metoda jest wywoływana podczas projektowania i czas wykonania.Składnik należy ustanowić połączenie z określonym przez uruchomienie-Menedżer połączeń czas połączenia, a następnie do zewnętrznego źródła danych.

Po nawiązaniu połączenia powinien być buforowany wewnętrznie przez składnik i zwalniane po ReleaseConnections() Metoda jest wywoływana. The ReleaseConnections() metoda is called at design and execution czas, like the AcquireConnections(Object) metoda. Deweloperzy mogą zastąpić tę metoda, a następnie zwolnij połączenie ustanowione przez składnik podczas AcquireConnections(Object).

Poniższy przykład kodu pokazuje składnik, który łączy się z połączeniem ADO.NET w AcquireConnections(Object) metody i zamyka połączenie ReleaseConnections() Metoda.

private SqlConnection sqlConnection;

public override void AcquireConnections(object transaction)
{
    if (ComponentMetaData.RuntimeConnectionCollection[0].ConnectionManager != null)
    {
        ConnectionManager cm = Microsoft.SqlServer.Dts.Runtime.DtsConvert.GetWrapper(ComponentMetaData.RuntimeConnectionCollection[0].ConnectionManager);
        ConnectionManagerAdoNet cmado = cm.InnerObject as ConnectionManagerAdoNet;

        if (cmado == null)
            throw new Exception("The ConnectionManager " + cm.Name + " is not an ADO.NET connection.");

        sqlConnection = cmado.AcquireConnection(transaction) as SqlConnection;
        sqlConnection.Open();
    }
}

public override void ReleaseConnections()
{
    if (sqlConnection != null && sqlConnection.State != ConnectionState.Closed)
        sqlConnection.Close();
}

Private sqlConnection As SqlConnection

Public Overrides Sub AcquireConnections(ByVal transaction As Object)

    If Not IsNothing(ComponentMetaData.RuntimeConnectionCollection(0).ConnectionManager) Then

        Dim cm As ConnectionManager = Microsoft.SqlServer.Dts.Runtime.DtsConvert.GetWrapper(ComponentMetaData.RuntimeConnectionCollection(0).ConnectionManager)
        Dim cmado As ConnectionManagerAdoNet = CType(cm.InnerObject, ConnectionManagerAdoNet)

        If IsNothing(cmado) Then
            Throw New Exception("The ConnectionManager " + cm.Name + " is not an ADO.NET connection.")
        End If

        sqlConnection = CType(cmado.AcquireConnection(transaction), SqlConnection)
        sqlConnection.Open()

    End If
End Sub

Public Overrides Sub ReleaseConnections()

    If Not IsNothing(sqlConnection) And sqlConnection.State <> ConnectionState.Closed Then
        sqlConnection.Close()
    End If

End Sub

Kolumny wyjściowe składnika urządzenie źródłowe odzwierciedlają kolumny z zewnętrznego urządzenie źródłowe danych, składnika są dodawane do przepływ danych podczas wykonywania.W projekcie czas, dodawanie kolumn wyjściowych po składnik został skonfigurowany w taki sposób, aby połączyć się z zewnętrznym źródłem danych.metoda czas projektowania, której składnik używa do dodania kolumny do swojej kolekcja danych wyjściowych może się różnić w zależności od potrzeb części, mimo że nie powinny one zostać dodane podczas Validate() lub AcquireConnections(Object). Na przykład, składnik, który zawiera instrukcja języka SQL we właściwość niestandardowej, kontrolujące zestaw danych dla składnika może dodać jej kolumn wyjściowych podczas SetComponentProperty(String, Object) Metoda. Składnik sprawdza, czy go korzysta z pamięci podręcznej połączenia, a jeśli tak się stanie, łączy się z danymi urządzenie źródłowe i generuje jej kolumn wyjściowych.

Po utworzeniu kolumna wyprowadzenia zestaw to jego typ danych właściwości, wywołując SetDataTypeProperties(DataType, Int32, Int32, Int32, Int32) Metoda. Ta metoda jest niezbędne ponieważ DataType(), Length(), Precision(), a CodePage() właściwość są tylko do odczytu i poszczególnych właściwość zależy od ustawień z drugiej strony. Ta metoda wymusza konieczność tych wartości, które mają być ustawione w sposób ciągły, a także zadanie przepływ danych sprawdza, czy są one poprawnie ustawione.

The DataType() of the kolumna determines the values that are zestaw for the other properties. W poniższej tabela przedstawiono wymagania dotyczące zależne od właściwości dla każdego DataType(). Typy danych niewymienionych są zależne od właściwości zestaw na zero.

Ponieważ ograniczeń na właściwości typów danych zależą od typu danych kolumna danych wyjściowych, należy wybrać odpowiedniego SSIS Typ danych podczas pracy z typami zarządzanych. Klasa podstawowa oferuje trzy metody pomocnika, ConvertBufferDataTypeToFitManaged(DataType, Boolean%), BufferTypeToDataRecordType(DataType), a DataRecordTypeToBufferType(Type), ułatwiające Deweloperzy składników zarządzanych w wyborze SSIS biorąc pod uwagę typu zarządzanego typu danych. Te metody konwersji typów danych zarządzanych SSIS typy danych i na odwrót.

Następujący kod jak pokazano w przykładzie danych wyjściowych kolumna pobierania składnika jest pusta, oparty na schemacie tabela.Są używane do metody pomocnika klasy podstawowej zestaw typ danych kolumna i właściwości zależne są zestaw na podstawie typu danych.

SqlCommand sqlCommand;

private void CreateColumnsFromDataTable()
{
    // Get the output.
    IDTSOutput100 output = ComponentMetaData.OutputCollection[0];

    // Start clean, and remove the columns from both collections.
    output.OutputColumnCollection.RemoveAll();
    output.ExternalMetadataColumnCollection.RemoveAll();

    this.sqlCommand = sqlConnection.CreateCommand();
    this.sqlCommand.CommandType = CommandType.Text;
    this.sqlCommand.CommandText = (string)ComponentMetaData.CustomPropertyCollection["SqlStatement"].Value;
    SqlDataReader schemaReader = this.sqlCommand.ExecuteReader(CommandBehavior.SchemaOnly);
    DataTable dataTable = schemaReader.GetSchemaTable();

    // Walk the columns in the schema, 
    // and for each data column create an output column and an external metadata column.
    foreach (DataRow row in dataTable.Rows)
    {
        IDTSOutputColumn100 outColumn = output.OutputColumnCollection.New();
        IDTSExternalMetadataColumn100 exColumn = output.ExternalMetadataColumnCollection.New();

        // Set column data type properties.
        bool isLong = false;
        DataType dt = DataRecordTypeToBufferType((Type)row["DataType"]);
        dt = ConvertBufferDataTypeToFitManaged(dt, ref isLong);
        int length = 0;
        int precision = (short)row["NumericPrecision"];
        int scale = (short)row["NumericScale"];
        int codepage = dataTable.Locale.TextInfo.ANSICodePage;

        switch (dt)
        {
            // The length cannot be zero, and the code page property must contain a valid code page.
            case DataType.DT_STR:
            case DataType.DT_TEXT:
                length = precision;
                precision = 0;
                scale = 0;
                break;

            case DataType.DT_WSTR:
                length = precision;
                codepage = 0;
                scale = 0;
                precision = 0;
                break;

            case DataType.DT_BYTES:
                precision = 0;
                scale = 0;
                codepage = 0;
                break;

            case DataType.DT_NUMERIC:
                length = 0;
                codepage = 0;

                if (precision > 38)
                    precision = 38;

                if (scale > 6)
                    scale = 6;
                break;

            case DataType.DT_DECIMAL:
                length = 0;
                precision = 0;
                codepage = 0;
                break;

            default:
                length = 0;
                precision = 0;
                codepage = 0;
                scale = 0;
                break;

        }

        // Set the properties of the output column.
        outColumn.Name = (string)row["ColumnName"];
        outColumn.SetDataTypeProperties(dt, length, precision, scale, codepage);
    }
}

Private sqlCommand As SqlCommand

Private Sub CreateColumnsFromDataTable()

    ' Get the output.
    Dim output As IDTSOutput100 = ComponentMetaData.OutputCollection(0)

    ' Start clean, and remove the columns from both collections.
    output.OutputColumnCollection.RemoveAll()
    output.ExternalMetadataColumnCollection.RemoveAll()

    Me.sqlCommand = sqlConnection.CreateCommand()
    Me.sqlCommand.CommandType = CommandType.Text
    Me.sqlCommand.CommandText = CStr(ComponentMetaData.CustomPropertyCollection("SqlStatement").Value)

    Dim schemaReader As SqlDataReader = Me.sqlCommand.ExecuteReader(CommandBehavior.SchemaOnly)
    Dim dataTable As DataTable = schemaReader.GetSchemaTable()

    ' Walk the columns in the schema, 
    ' and for each data column create an output column and an external metadata column.
    For Each row As DataRow In dataTable.Rows

        Dim outColumn As IDTSOutputColumn100 = output.OutputColumnCollection.New()
        Dim exColumn As IDTSExternalMetadataColumn100 = output.ExternalMetadataColumnCollection.New()

        ' Set column data type properties.
        Dim isLong As Boolean = False
        Dim dt As DataType = DataRecordTypeToBufferType(CType(row("DataType"), Type))
        dt = ConvertBufferDataTypeToFitManaged(dt, isLong)
        Dim length As Integer = 0
        Dim precision As Integer = CType(row("NumericPrecision"), Short)
        Dim scale As Integer = CType(row("NumericScale"), Short)
        Dim codepage As Integer = dataTable.Locale.TextInfo.ANSICodePage

        Select Case dt

            ' The length cannot be zero, and the code page property must contain a valid code page.
            Case DataType.DT_STR
            Case DataType.DT_TEXT
                length = precision
                precision = 0
                scale = 0

            Case DataType.DT_WSTR
                length = precision
                codepage = 0
                scale = 0
                precision = 0

            Case DataType.DT_BYTES
                precision = 0
                scale = 0
                codepage = 0

            Case DataType.DT_NUMERIC
                length = 0
                codepage = 0

                If precision > 38 Then
                    precision = 38
                End If

                If scale > 6 Then
                    scale = 6
                End If

            Case DataType.DT_DECIMAL
                length = 0
                precision = 0
                codepage = 0

            Case Else
                length = 0
                precision = 0
                codepage = 0
                scale = 0
        End Select

        ' Set the properties of the output column.
        outColumn.Name = CStr(row("ColumnName"))
        outColumn.SetDataTypeProperties(dt, length, precision, scale, codepage)
    Next
End Sub

Należy sprawdzić poprawność urządzenie źródłowe składników i sprawdź, czy w kolumnach wyznaczonych w jego danych wyjściowych kolekcji kolumn pasują kolumn na danych zewnętrznych urządzenie źródłowe.Czasami sprawdzenie kolumn wyjściowych wobec z zewnętrznym źródłem danych może być niemożliwe, takie jak w stanie rozłączenia lub zaleca się unikać długiej podróży round na serwerze.W takich sytuacjach kolumny w danych wyjściowych nadal mogą być sprawdzone za pomocą ExternalMetadataColumnCollection() dane wyjściowe obiektu. Aby uzyskać więcej informacji zobaczValidating a Data Flow Component.

Ta kolekcja, znajduje się na obu danych wejściowych i obiektów danych wyjściowych i użytkownik może wypełnić go wykorzystując kolumny z zewnętrznego urządzenie źródłowe danych.Ta kolekcja służy do sprawdzania poprawności kolumn wyjściowych podczas SSIS Projektant jest w trybie offline, gdy składnik jest odłączony lub ValidateExternalMetadata() Właściwość jest false. Kolekcja powinien najpierw pusta w tym samym czasie utworzonych kolumn wyjściowych.Dodawanie kolumn zewnętrznych metadane do kolekcja jest stosunkowo łatwe, ponieważ kolumna zewnętrznych metadane początkowo powinna być taka sama kolumna wyprowadzenia.Właściwości typu danych kolumna powinny zostały już ustawione poprawnie i właściwości mogą być skopiowane bezpośrednio do IDTSExternalMetadataColumn100 obiekt.

Następujący kod dodaje kolumna zewnętrzną metadane, oparty na kolumnie nowo utworzone dane wyjściowe.Przyjęto założenie, że kolumna wyprowadzenia została już utworzona.

private void CreateExternalMetaDataColumn(IDTSOutput100 output, IDTSOutputColumn100 outputColumn)
{
    
    // Set the properties of the external metadata column.
    IDTSExternalMetadataColumn100 externalColumn = output.ExternalMetadataColumnCollection.New();
    externalColumn.Name = outputColumn.Name;
    externalColumn.Precision = outputColumn.Precision;
    externalColumn.Length = outputColumn.Length;
    externalColumn.DataType = outputColumn.DataType;
    externalColumn.Scale = outputColumn.Scale;

    // Map the external column to the output column.
    outputColumn.ExternalMetadataColumnID = externalColumn.ID;

}

Private Sub CreateExternalMetaDataColumn(ByVal output As IDTSOutput100, ByVal outputColumn As IDTSOutputColumn100)


        ' Set the properties of the external metadata column.
        Dim externalColumn As IDTSExternalMetadataColumn100 = output.ExternalMetadataColumnCollection.New()
        externalColumn.Name = outputColumn.Name
        externalColumn.Precision = outputColumn.Precision
        externalColumn.Length = outputColumn.Length
        externalColumn.DataType = outputColumn.DataType
        externalColumn.Scale = outputColumn.Scale

        ' Map the external column to the output column.
        outputColumn.ExternalMetadataColumnID = externalColumn.ID

    End Sub

Bufor wyjściowy dla składnika zawiera w kolumnach wyznaczonych przez element i wszystkie kolumny dodawane do wyników ze składnika niższego rzędu.Na przykład jeśli składnik urządzenie źródłowe zawiera trzy kolumna dane wyjściowe i kolejny składnik dodaje czwarty wyjściowy kolumna bufor wyjściowy, pod warunkiem, że do użytku przez urządzenie źródłowe składnikiem zawiera te cztery kolumna.

Kolejność kolumn w wierszu bufor nie jest zdefiniowany przez indeks kolumna danych wyjściowych w kolekcja kolumn wyjściowych.Kolumna wyprowadzenia mogą tylko być dokładnie zlokalizowane w wierszu bufora za pomocą FindColumnByLineageID(Int32, Int32) Metoda BufferManager(). Ta metoda lokalizuje kolumna z IDENTYFIKATOREM rodowodu określony w określonym buforze i zwraca jego lokalizacji w wierszu.Indeksy kolumny wyjściowe znajdują się zwykle w PreExecute() metoda i zapisane do użytku podczas PrimeOutput(Int32, array<Int32[], array<PipelineBuffer[]).

W poniższym przykładzie kodu znajdzie lokalizację kolumn wyjściowych buforu wyjściowego podczas wywoływania PreExecute()i przechowuje je w strukturę wewnętrzną. Nazwa kolumna jest również przechowywane w strukturze i jest używany w przykładzie kodu PrimeOutput(Int32, array<Int32[], array<PipelineBuffer[]) Metoda w następnej sekcji tego tematu.

ArrayList columnInformation;

private struct ColumnInfo
{
    public int BufferColumnIndex;
    public string ColumnName;
}

public override void PreExecute()
{
    this.columnInformation = new ArrayList();
    IDTSOutput100 output = ComponentMetaData.OutputCollection[0];

    foreach (IDTSOutputColumn100 col in output.OutputColumnCollection)
    {
        ColumnInfo ci = new ColumnInfo();
        ci.BufferColumnIndex = BufferManager.FindColumnByLineageID(output.Buffer, col.LineageID);
        ci.ColumnName = col.Name;
        columnInformation.Add(ci);
    }
}

Public Overrides Sub PreExecute()

    Me.columnInformation = New ArrayList()
    Dim output As IDTSOutput100 = ComponentMetaData.OutputCollection(0)

    For Each col As IDTSOutputColumn100 In output.OutputColumnCollection

        Dim ci As ColumnInfo = New ColumnInfo()
        ci.BufferColumnIndex = BufferManager.FindColumnByLineageID(output.Buffer, col.LineageID)
        ci.ColumnName = col.Name
        columnInformation.Add(ci)
    Next
End Sub

Wiersze są dodawane do buforu wyjściowego, wywołując AddRow() Metoda, która tworzy nowy wiersz buforu o wartości puste w jej kolumnach. Składnik przypisuje następnie wartości do poszczególnych kolumn.Bufory wyjściowe, dostarczane ze składnikiem są tworzone i monitorowane przez zadanie przepływ danych.Jak staną się pełne, wiersze w buforze są przenoszone do następnego składnika.Nie ma możliwości do ustalenia, kiedy wsadowego wierszy została wysłana do następnego składnika ponieważ przemieszczanie wierszy zadanie przepływ danych jest niewidoczny dla deweloperów składników, RowCount() Właściwość zawsze wynosi zero buforów wyjściowych. Po zakończeniu części urządzenie źródłowe Dodawanie wierszy do buforu wyjściowego, powiadamia zadania przepływ danych przez wywołanie SetEndOfRowset() Metoda PipelineBuffer, a pozostałe wiersze w buforze są przekazywane do następnego składnika.

W czasie, gdy składnik urządzenie źródłowe odczytuje wiersze z zewnętrznego urządzenie źródłowe danych, może zajść potrzeba zaktualizowania "Wiersze odczytu" lub "BLOB Bajty odczytane" liczników wydajności poprzez wywołanie IncrementPipelinePerfCounter(UInt32, UInt32) Metoda. Aby uzyskać więcej informacji zobaczMonitoring the Performance of the Data Flow Engine.

Poniższy przykład kodu pokazuje, wiersze są dodawane do bufor wyjściowy w przez składnik PrimeOutput(Int32, array<Int32[], array<PipelineBuffer[]). Za pomocą zostały indeksów kolumn wyprowadzenia w buforze PreExecute() w poprzednim przykładzie kodu.

public override void PrimeOutput(int outputs, int[] outputIDs, PipelineBuffer[] buffers)
{
    IDTSOutput100 output = ComponentMetaData.OutputCollection[0];
    PipelineBuffer buffer = buffers[0];

    SqlDataReader dataReader = sqlCommand.ExecuteReader();

    // Loop over the rows in the DataReader, 
    // and add them to the output buffer.
    while (dataReader.Read())
    {
        // Add a row to the output buffer.
        buffer.AddRow();

        for (int x = 0; x < columnInformation.Count; x++)
        {
            ColumnInfo ci = (ColumnInfo)columnInformation[x];
            int ordinal = dataReader.GetOrdinal(ci.ColumnName);

            if (dataReader.IsDBNull(ordinal))
                buffer.SetNull(ci.BufferColumnIndex);
            else
            {
                buffer[ci.BufferColumnIndex] = dataReader[ci.ColumnName];
            }
        }
    }
    buffer.SetEndOfRowset();
}

Public Overrides Sub PrimeOutput(ByVal outputs As Integer, ByVal outputIDs As Integer(), ByVal buffers As PipelineBuffer())

    Dim output As IDTSOutput100 = ComponentMetaData.OutputCollection(0)
    Dim buffer As PipelineBuffer = buffers(0)

    Dim dataReader As SqlDataReader = sqlCommand.ExecuteReader()

    ' Loop over the rows in the DataReader, 
    ' and add them to the output buffer.
    While (dataReader.Read())

        ' Add a row to the output buffer.
        buffer.AddRow()

        For x As Integer = 0 To columnInformation.Count

            Dim ci As ColumnInfo = CType(columnInformation(x), ColumnInfo)

            Dim ordinal As Integer = dataReader.GetOrdinal(ci.ColumnName)

            If (dataReader.IsDBNull(ordinal)) Then
                buffer.SetNull(ci.BufferColumnIndex)
            Else
                buffer(ci.BufferColumnIndex) = dataReader(ci.ColumnName)

            End If
        Next

    End While

    buffer.SetEndOfRowset()
End Sub