Taal

Runtimemethoden van een gegevensstroomonderdeel

Van toepassing op:SQL Server SSIS Integration Runtime in Azure Data Factory

Tijdens runtime onderzoekt de dataflow-taak de volgorde van componenten, stelt een uitvoeringsplan op en beheert een pool van worker-threads die het werkplan uitvoeren. De taak laadt rijen data uit bronnen, verwerkt ze via transformaties en slaat ze vervolgens op als bestemming.

Volgorde van methode-uitvoering

Tijdens de uitvoering van een dataflowcomponent wordt een deelverzameling van de methoden in de PipelineComponent basisklasse aangeroepen. De methoden, en de volgorde waarin ze worden aangeroepen, zijn altijd hetzelfde, met uitzondering van de PrimeOutput en-methoden ProcessInput . Deze twee methoden worden genoemd op basis van het bestaan en de configuratie van een component en IDTSInput100IDTSOutput100 objecten.

De volgende lijst toont de methoden in de volgorde waarin ze worden aangeroepen tijdens componentuitvoering. Merk op dat PrimeOutput, wanneer aangeroepen, altijd vóór ProcessInputwordt aangeroepen.

PrimeOutput-methode

De PrimeOutput methode wordt aangeroepen wanneer een component ten minste één output heeft, verbonden met een downstream component via een IDTSPath100 object, en de SynchronousInputID eigenschap van de output nul is. De PrimeOutput methode wordt aangeroepen voor broncomponenten en voor transformaties met asynchrone uitgangen. In tegenstelling tot de ProcessInput hieronder beschreven methode wordt de PrimeOutput methode slechts één keer aangeroepen voor elke component die deze vereist.

ProcessInput-methode

De ProcessInput methode wordt aangeroepen voor componenten waarbij ten minste één invoer door een IDTSPath100 object aan een upstream component is gekoppeld. De ProcessInput methode wordt aangeroepen voor bestemmingscomponenten en voor transformaties met synchrone uitgangen. ProcessInput wordt herhaaldelijk aangeroepen totdat er geen rijen meer zijn om te verwerken vanuit upstream componenten.

Werken met Inputs en Outputs

Tijdens uitvoering voeren datastroomcomponenten de volgende taken uit:

  • Broncomponenten voegen rijen toe.

  • Transformatiecomponenten met synchrone uitgangen ontvangen rijen geleverd door broncomponenten.

  • Transformatiecomponenten met asynchrone uitgangen ontvangen rijen en voegen rijen op.

  • Bestemmingscomponenten ontvangen rijen en laden ze vervolgens in een bestemming.

Tijdens uitvoering wijst PipelineBuffer de dataflow-taak objecten toe die alle kolommen bevatten die zijn gedefinieerd in de outputkolomcollecties van een reeks componenten. Als bijvoorbeeld elk van de vier componenten in een dataflowsequentie één outputkolom toevoegt aan zijn output column collectie, bevat de buffer die aan elke component wordt geleverd vier kolommen, één voor elke outputkolom per component. Door dit gedrag ontvangt een component soms buffers die kolommen bevatten die het niet gebruikt.

Aangezien de buffers die je component ontvangt kolommen kunnen bevatten die de component niet zal gebruiken, moet je de kolommen vinden die je wilt gebruiken in de input- en outputkolomcollecties van je component in de buffer die de dataflow-taak aan de component krijgt. Je doet dit door de FindColumnByLineageID methode van de BufferManager eigenschap te gebruiken. Om prestatieredenen wordt deze taak gewoonlijk uitgevoerd tijdens de PreExecute methode, in plaats van in of PrimeOutputProcessInput.

PreExecute wordt aangeroepen vóór de PrimeOutput en-methoden ProcessInput , en is de eerste kans voor een component om dit werk uit te voeren nadat de beschikbaar BufferManager is voor de component. Tijdens deze methode moet de component zijn kolommen in de buffers lokaliseren en deze informatie intern opslaan zodat de kolommen gebruikt kunnen worden in de of ProcessInput methodePrimeOutput.

Het volgende codevoorbeeld laat zien hoe een transformatiecomponent met een synchrone output zijn invoerkolommen in de buffer plaatst tijdens 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

Rijen toevoegen

Componenten leveren rijen aan downstream componenten door rijen toe te voegen aan PipelineBuffer objecten. De dataflow-taak biedt een array van outputbuffers - één voor elk IDTSOutput100 object dat is verbonden met een downstreamcomponent - als parameter voor de PrimeOutput methode. Broncomponenten en transformatiecomponenten met asynchrone uitgangen voegen rijen toe aan de buffers en roepen de SetEndOfRowset methode aan wanneer ze klaar zijn met het toevoegen van rijen. De dataflow-taak beheert de outputbuffers die het aan componenten levert en, zodra een buffer vol raakt, verplaatst het automatisch de rijen in de buffer naar de volgende component. De PrimeOutput methode wordt één keer per component genoemd, in tegenstelling tot de ProcessInput methode die herhaaldelijk wordt aangeroepen.

Het volgende codevoorbeeld laat zien hoe een component tijdens de PrimeOutput methode rijen toevoegt aan zijn outputbuffers en vervolgens de SetEndOfRowset methode aanroept.

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

Voor meer informatie over het ontwikkelen van componenten die rijen toevoegen aan outputbuffers, zie Developing a Custom Source Component en Developing a Custom Transformation Component with Asynchronous Outputs.

Ontvangende rijen

Componenten ontvangen rijen van upstream componenten in PipelineBuffer objecten. De dataflow-taak biedt een PipelineBuffer object dat de rijen bevat die door upstream-componenten aan de datastroom zijn toegevoegd als parameter voor de ProcessInput methode. Deze invoerbuffer kan worden gebruikt om de rijen en kolommen in de buffer te onderzoeken en aan te passen, maar kan niet worden gebruikt om rijen toe te voegen of te verwijderen. De ProcessInput methode wordt herhaaldelijk aangeroepen totdat er geen buffers meer beschikbaar zijn. De laatste keer dat het wordt genoemd, is de EndOfRowset eigenschap waar. Je kunt over de verzameling rijen in de buffer itereren met behulp van de methode NextRow , die de buffer naar de volgende rij verplaatst. Deze methode geeft false wanneer de buffer op de laatste rij in de collectie staat. Je hoeft de EndOfRowset eigenschap niet te controleren tenzij je een extra actie moet uitvoeren nadat de laatste rijen data zijn verwerkt.

De volgende tekst toont het juiste patroon voor het gebruik van de NextRow methode en de EndOfRowset eigenschap:

while (buffer.NextRow())

{

// Do something with each row.

}

if (buffer.EndOfRowset)

{

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

}

Het volgende codevoorbeeld laat zien hoe een component de rijen in inputbuffers verwerkt tijdens de ProcessInput methode.

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

Voor meer informatie over het ontwikkelen van componenten die rijen ontvangen in inputbuffers, zie Developing a Custom Destination Component en Developing a Custom Transformation Component with Synchronous outputs.

Verwante onderwerpen

  • Last updated on