Desenvolvendo um componente de transformação personalizado com saídas síncronas

As entradas, saídas e propriedades personalizadas do componente normalmente são criadas durante o método ProvideComponentProperties. Há dois modos de adicionar a entrada e a saída de um componente de transformação com saídas síncronas. Você pode usar a implementação da classe base do método e modificar a entrada e a saída padrão que ele cria, ou adicionar explicitamente a entrada e a saída.

O exemplo de código a seguir mostra uma implementação de ProvideComponentProperties que explicitamente adiciona os objetos de entrada e saída. A chamada para a classe base que realizaria a mesma função é incluída em um comentário.

using Microsoft.SqlServer.Dts.Pipeline;
using Microsoft.SqlServer.Dts.Pipeline.Wrapper;
using Microsoft.SqlServer.Dts.Runtime;

namespace Microsoft.Samples.SqlServer.Dts
{
    [DtsPipelineComponent(DisplayName = "SynchronousComponent", ComponentType = ComponentType.Transform)]
    public class SyncComponent : PipelineComponent
    {

        public override void ProvideComponentProperties()
        {
            // Add the input.
            IDTSInput100 input = ComponentMetaData.InputCollection.New();
            input.Name = "Input";

            // Add the output.
            IDTSOutput100 output = ComponentMetaData.OutputCollection.New();
            output.Name = "Output";
            output.SynchronousInputID = input.ID;

            // Alternatively, you can let the base class add the input and output
            // and set the SynchronousInputID of the output to the ID of the input.
            // base.ProvideComponentProperties();
        }
    }
}

Imports Microsoft.SqlServer.Dts.Pipeline
Imports Microsoft.SqlServer.Dts.Pipeline.Wrapper
Imports Microsoft.SqlServer.Dts.Runtime

<DtsPipelineComponent(DisplayName:="SynchronousComponent", ComponentType:=ComponentType.Transform)> _
Public Class SyncComponent
    Inherits PipelineComponent

    Public Overrides Sub ProvideComponentProperties()

        ' Add the input.
        Dim input As IDTSInput100 = ComponentMetaData.InputCollection.New()
        input.Name = "Input"

        ' Add the output.
        Dim output As IDTSOutput100 = ComponentMetaData.OutputCollection.New()
        output.Name = "Output"
        output.SynchronousInputID = Input.ID

        ' Alternatively, you can let the base class add the input and output
        ' and set the SynchronousInputID of the output to the ID of the input.
        ' base.ProvideComponentProperties();

    End Sub

End Class

Embora componentes de transformação com saídas síncronas não acrescentem linhas a buffers, eles podem acrescentar colunas de saída extras à sua saída. Normalmente, quando um componente adiciona uma coluna de saída, os valores dessa nova coluna são derivados em tempo de execução dos dados contidos em uma ou mais das colunas fornecidas ao componente por um componente upstream.

Depois que uma coluna de saída foi criada, suas propriedades de tipo de dados devem ser definidas. Definir as propriedades de tipo de dados de uma coluna de saída requer manipulação especial e é executado chamando o método SetDataTypeProperties. Esse método é necessário por que as propriedades DataType, Length, Precision e CodePage são individualmente somente leitura, porque cada uma delas depende das configurações da outra. Esse método garante que os valores das propriedades sejam definidos de forma consistente, e a tarefa de fluxo de dados valida se eles foram definidos corretamente.

O DataType da coluna determina os valores que são definidos para as outras propriedades. A tabela a seguir mostra os requisitos nas propriedades dependentes para cada DataType. Os tipos de dados não listados têm as propriedades dependentes definidas como zero.

Como as restrições nas propriedades de tipo de dados são baseadas no tipo de dados da coluna de saída, você deve escolher o tipo de dados do Integration Services correto quando trabalhar com tipos gerenciados. A classe base fornece três métodos auxiliares, ConvertBufferDataTypeToFitManaged, BufferTypeToDataRecordType e DataRecordTypeToBufferType, que auxiliam desenvolvedores de componente gerenciados a selecionar um tipo de dados do SSIS que recebeu um tipo gerenciado. Esses métodos convertem tipos de dados gerenciados em tipos de dados do SSIS e vice-versa.

O número de colunas nos buffers fornecido a um componente durante a execução provavelmente excederá o número de colunas nas coleções de entrada ou saída do componente. Isso se dá por que cada buffer contém todas as colunas de saída definidas nos componentes em um fluxo de dados. Para que as colunas do buffer sejam corretamente associadas às colunas da entrada ou da saída, os desenvolvedores de componentes devem usar o método FindColumnByLineageID do BufferManager. Esse método localiza uma coluna no buffer especificado por sua ID de linhagem. Normalmente, são localizadas colunas durante PreExecute por que este é o primeiro método de tempo de execução em que a propriedade BufferManager fica disponível.

O exemplo de código seguinte mostra um componente que localiza índices das colunas em sua coleção de colunas de entrada e saída durante PreExecute. Os índices da coluna são armazenados em uma matriz de inteiro e podem ser acessados pelo componente durante ProcessInput.

int []inputColumns;
int []outputColumns;

public override void PreExecute()
{
    IDTSInput100 input = ComponentMetaData.InputCollection[0];
    IDTSOutput100 output = ComponentMetaData.OutputCollection[0];

    inputColumns = new int[input.InputColumnCollection.Count];
    outputColumns = new int[output.OutputColumnCollection.Count];

    for(int col=0; col < input.InputColumnCollection.Count; col++)
    {
        IDTSInputColumn100 inputColumn = input.InputColumnCollection[col];
        inputColumns[col] = BufferManager.FindColumnByLineageID(input.Buffer, inputColumn.LineageID);
    }

    for(int col=0; col < output.OutputColumnCollection.Count; col++)
    {
        IDTSOutputColumn100 outputColumn = output.OutputColumnCollection[col];
        outputColumns[col] = BufferManager.FindColumnByLineageID(input.Buffer, outputColumn.LineageID);
    }

}

Public Overrides Sub PreExecute()

    Dim input As IDTSInput100 = ComponentMetaData.InputCollection(0)
    Dim output As IDTSOutput100 = ComponentMetaData.OutputCollection(0)


    ReDim inputColumns(input.InputColumnCollection.Count)
    ReDim outputColumns(output.OutputColumnCollection.Count)

    For col As Integer = 0 To input.InputColumnCollection.Count

        Dim inputColumn As IDTSInputColumn100 = input.InputColumnCollection(col)
        inputColumns(col) = BufferManager.FindColumnByLineageID(input.Buffer, inputColumn.LineageID)
    Next

    For col As Integer = 0 To output.OutputColumnCollection.Count

        Dim outputColumn As IDTSOutputColumn100 = output.OutputColumnCollection(col)
        outputColumns(col) = BufferManager.FindColumnByLineageID(input.Buffer, outputColumn.LineageID)
    Next

End Sub

Componentes recebem objetos PipelineBuffer que contêm linhas e colunas no método ProcessInput. Durante esse método são iteradas as linhas do buffer e as colunas identificadas durante PreExecute são lidas e modificadas. O método é chamado repetidamente pela tarefa de fluxo de dados até que nenhuma linha seja mais fornecida a partir do componente upstream.

Uma coluna individual no buffer é lida ou gravada através do método de acesso do indexador matriz ou um dos métodos Get ou Set. Os métodos Get e Set são mais eficientes e devem ser usados quando o tipo de dados da coluna no buffer for conhecido.

O exemplo de código seguinte mostra uma implementação do método ProcessInput que processa linhas de entrada.

public override void ProcessInput( int InputID, PipelineBuffer buffer)
{
       while( buffer.NextRow())
       {
            for(int x=0; x < inputColumns.Length;x++)
            {
                if(!buffer.IsNull(inputColumns[x]))
                {
                    object columnData = buffer[inputColumns[x]];
                    // TODO: Modify the column data.
                    buffer[inputColumns[x]] = columnData;
                }
            }
        
      }
}

Public Overrides Sub ProcessInput(ByVal InputID As Integer, ByVal buffer As PipelineBuffer)

        While (buffer.NextRow())

            For x As Integer = 0 To inputColumns.Length

                if buffer.IsNull(inputColumns(x)) = false then

                    Dim columnData As Object = buffer(inputColumns(x))
                    ' TODO: Modify the column data.
                    buffer(inputColumns(x)) = columnData

                End If
            Next

        End While
End Sub