Métodos de tempo de execução de um componente de fluxo de dados

Aplica-se a: SQL Server SSIS Integration Runtime em Azure Data Factory

Em tempo de execução, a tarefa de fluxo de dados examina a sequência de componentes, prepara um plano de execução e gere um conjunto de threads de trabalho que executam o plano de trabalho. A tarefa carrega linhas de dados das fontes, processa-as através de transformações e depois guarda-as para os destinos.

Sequência de execução do método

Durante a execução de um componente de fluxo de dados, é chamado um subconjunto dos métodos da PipelineComponent classe base. Os métodos, e a sequência em que são chamados, são sempre os mesmos, com exceção dos PrimeOutput métodos e.ProcessInput Estes dois métodos são chamados com base na existência e configuração de um componente IDTSInput100 e IDTSOutput100 dos objetos.

A lista seguinte mostra os métodos na ordem em que são chamados durante a execução dos componentes. Note-se que PrimeOutput, quando chamado, é sempre chamado antes ProcessInputde .

Método PrimeOutput

O PrimeOutput método é chamado quando um componente tem pelo menos uma saída, ligada a um componente a jusante através de um IDTSPath100 objeto, e a SynchronousInputID propriedade da saída é zero. O PrimeOutput método é chamado para componentes fonte e para transformações com saídas assíncronas. Ao contrário do ProcessInput método descrito abaixo, o PrimeOutput método é chamado apenas uma vez para cada componente que o necessite.

Método ProcessInput

O ProcessInput método é chamado para componentes que têm pelo menos uma entrada ligada a um componente a montante por um IDTSPath100 objeto. O ProcessInput método é chamado para componentes de destino e para transformações com saídas síncronas. ProcessInput é chamada repetidamente até não haver mais linhas para processar a partir dos componentes a montante.

Trabalhar com Entradas e Saídas

Em tempo de execução, os componentes do fluxo de dados executam as seguintes tarefas:

  • Os componentes de origem adicionam linhas.

  • Os componentes de transformação com saídas síncronas recebem linhas fornecidas pelos componentes de origem.

  • Componentes de transformação com saídas assíncronas recebem linhas e somam linhas.

  • Os componentes de destino recebem linhas e depois carregam-nas num destino.

Durante a execução, a tarefa de fluxo de PipelineBuffer dados aloca objetos que contêm todas as colunas definidas nas coleções de colunas de saída de uma sequência de componentes. Por exemplo, se cada um dos quatro componentes numa sequência de fluxo de dados adicionar uma coluna de saída à sua coleção de colunas de saída, o buffer fornecido a cada componente contém quatro colunas, uma para cada coluna de saída por componente. Devido a este comportamento, um componente por vezes recebe buffers que contêm colunas que não utiliza.

Como os buffers recebidos pelo seu componente podem conter colunas que o componente não irá usar, deve localizar as colunas que pretende usar nas coleções de colunas de entrada e saída do seu componente no buffer fornecido ao componente pela tarefa de fluxo de dados. Faz-se isto utilizando o FindColumnByLineageID método da BufferManager propriedade. Por razões de desempenho, esta tarefa é normalmente realizada durante o PreExecute método, em vez de em PrimeOutput ou ProcessInput.

PreExecute é chamado antes dos PrimeOutput métodos e ProcessInput , e é a primeira oportunidade para um componente realizar este trabalho depois de o BufferManager ficar disponível para o componente. Durante este método, o componente deve localizar as suas colunas nos buffers e armazenar esta informação internamente para que as colunas possam ser usadas nos PrimeOutput métodos ou ProcessInput .

O exemplo de código seguinte demonstra como um componente de transformação com saída síncrona localiza as suas colunas de entrada no buffer durante PreExecute.

private int []bufferColumnIndex;  
public override void PreExecute()  
{  
    IDTSInput100 input = ComponentMetaData.InputCollection[0];  
    bufferColumnIndex = new int[input.InputColumnCollection.Count];  
  
    for( int x=0; x < input.InputColumnCollection.Count; x++)  
    {  
        IDTSInputColumn100 column = input.InputColumnCollection[x];  
        bufferColumnIndex[x] = BufferManager.FindColumnByLineageID( input.Buffer, column.LineageID);  
    }  
}  

Dim bufferColumnIndex As Integer()  
  
    Public Overrides Sub PreExecute()  
  
        Dim input As IDTSInput100 = ComponentMetaData.InputCollection(0)  
  
        ReDim bufferColumnIndex(input.InputColumnCollection.Count)  
  
        For x As Integer = 0 To input.InputColumnCollection.Count  
  
            Dim column As IDTSInputColumn100 = input.InputColumnCollection(x)  
            bufferColumnIndex(x) = BufferManager.FindColumnByLineageID(input.Buffer, column.LineageID)  
  
        Next  
  
    End Sub

Adição de Linhas

Os componentes fornecem linhas aos componentes a jusante adicionando linhas a objetos PipelineBuffer . A tarefa de fluxo de dados fornece um array de buffers de saída – um para cada IDTSOutput100 objeto ligado a um componente a jusante – como parâmetro do PrimeOutput método. Componentes de origem e componentes de transformação com saídas assíncronas adicionam linhas aos buffers e chamam o SetEndOfRowset método quando terminam de adicionar linhas. A tarefa de fluxo de dados gere os buffers de saída que fornece aos componentes e, à medida que um buffer se enche, move automaticamente as linhas do buffer para o componente seguinte. O PrimeOutput método chama-se uma vez por componente, ao contrário do ProcessInput método, que é chamado repetidamente.

O exemplo de código seguinte demonstra como um componente adiciona linhas aos seus buffers de saída durante o PrimeOutput método e depois chama o SetEndOfRowset método.

public override void PrimeOutput( int outputs, int []outputIDs,PipelineBuffer []buffers)  
{  
    for( int x=0; x < outputs; x++ )  
    {  
        IDTSOutput100 output = ComponentMetaData.OutputCollection.GetObjectByID( outputIDs[x]);  
        PipelineBuffer buffer = buffers[x];  
  
        // TODO: Add rows to the output buffer.  
    }  
    foreach( PipelineBuffer buffer in buffers )  
    {  
        /// Notify the data flow task that no more rows are coming.  
        buffer.SetEndOfRowset();  
    }  
}  

public overrides sub PrimeOutput( outputs as Integer , outputIDs() as Integer ,buffers() as PipelineBuffer buffers)  
  
    For x As Integer = 0 To outputs.MaxValue  
  
        Dim output As IDTSOutput100 = ComponentMetaData.OutputCollection.GetObjectByID(outputIDs(x))  
        Dim buffer As PipelineBuffer = buffers(x)  
  
        ' TODO: Add rows to the output buffer.  
  
    Next  
  
    For Each buffer As PipelineBuffer In buffers  
  
        ' Notify the data flow task that no more rows are coming.  
        buffer.SetEndOfRowset()  
  
    Next  
  
End Sub

Para mais informações sobre o desenvolvimento de componentes que adicionam linhas a buffers de saída, consulte Desenvolvimento de um Componente Fonte Personalizado e Desenvolvimento de um Componente de Transformação Personalizado com Saídas Assíncronas.

Linhas de Receção

Os componentes recebem linhas de componentes a montante nos PipelineBuffer objetos. A tarefa de fluxo de dados fornece um PipelineBuffer objeto que contém as linhas adicionadas ao fluxo de dados pelos componentes a montante como parâmetro do ProcessInput método. Este buffer de entrada pode ser usado para examinar e modificar as linhas e colunas do buffer, mas não pode ser usado para adicionar ou remover linhas. O ProcessInput método é chamado repetidamente até não haver mais buffers disponíveis. Da última vez que é chamado, a EndOfRowset propriedade é verdadeira. Pode iterar sobre a coleção de linhas no buffer usando o NextRow método, que avança o buffer para a linha seguinte. Este método retorna false quando o buffer está na última linha da coleção. Não precisa de verificar a EndOfRowset propriedade a menos que tenha de realizar uma ação adicional após o processamento das últimas linhas de dados.

O texto seguinte mostra o padrão correto para usar o NextRow método e a EndOfRowset propriedade:

while (buffer.NextRow())

{

// Do something with each row.

}

if (buffer.EndOfRowset)

{

// Optionally, do something after all rows have been processed.

}

O exemplo de código seguinte demonstra como um componente processa as linhas nos buffers de entrada durante o ProcessInput método.

public override void ProcessInput( int inputID, PipelineBuffer buffer )  
{  
    {  
        IDTSInput100 input = ComponentMetaData.InputCollection.GetObjectByID(inputID);  
        while( buffer.NextRow())  
        {  
            // TODO: Examine the columns in the current row.  
        }  
}  

Public Overrides Sub ProcessInput(ByVal inputID As Integer, ByVal buffer As PipelineBuffer)  
  
        Dim input As IDTSInput100 = ComponentMetaData.InputCollection.GetObjectByID(inputID)  
        While buffer.NextRow() = True  
  
            ' TODO: Examine the columns in the current row.  
        End While  
  
End Sub

Para mais informações sobre o desenvolvimento de componentes que recebem linhas em buffers de entrada, consulte Desenvolvimento de um Componente de Destino Personalizado e Desenvolvimento de um Componente de Transformação Personalizado com Saídas Síncronas.

Conteúdo relacionado

  • Last updated on