Språk

Körningsmetoder för en dataflödeskomponent

Gäller för:SQL Server SSIS Integration Runtime i Azure Data Factory

Vid körning undersöker dataflödesuppgiften sekvensen av komponenter, förbereder en exekveringsplan och hanterar en pool av arbetstrådar som utför arbetsplanen. Uppgiften laddar rader av data från källor, bearbetar dem genom transformationer och sparar dem sedan till destinationer.

Sekvens av metodexekvering

Under exekveringen av en dataflödeskomponent anropas en delmängd av metoderna i PipelineComponent basklassen. Metoderna, och sekvensen i vilken de anropas, är alltid desamma, med undantag för PrimeOutput och ProcessInput metoderna. Dessa två metoder kallas baserat på existensen och konfigurationen IDTSInput100 av en komponents och IDTSOutput100 objekt.

Följande lista visar metoderna i den ordning de anropas under komponentexekveringen. Observera att PrimeOutput, när den anropas, kallas alltid före ProcessInput.

PrimeOutput-metoden

Metoden PrimeOutput kallas när en komponent har minst en utgång, kopplad till en nedströms komponent via ett IDTSPath100 objekt, och egenskapen SynchronousInputID för utgången är noll. Metoden PrimeOutput kallas för källkomponenter och för transformationer med asynkrona utgångar. Till skillnad från metoden ProcessInput som beskrivs nedan anropas metoden PrimeOutput endast en gång för varje komponent som kräver den.

ProcessInput-metoden

Metoden ProcessInput kallas för komponenter som har minst en ingång kopplad till en uppströmskomponent av ett IDTSPath100 objekt. Metoden ProcessInput kallas för destinationskomponenter och för transformationer med synkrona utgångar. ProcessInput anropas upprepade gånger tills det inte finns fler rader att bearbeta från uppströmskomponenter.

Arbete med in- och utgångar

Vid körning utför dataflödeskomponenter följande uppgifter:

  • Källkomponenter lägger till rader.

  • Transformationskomponenter med synkrona utgångar tar emot rader som tillhandahålls av källkomponenter.

  • Transformationskomponenter med asynkrona utgångar tar emot rader och lägger till rader.

  • Destinationskomponenter tar emot rader och laddar dem sedan in i en destination.

Under exekveringen allokerar PipelineBuffer dataflödesuppgiften objekt som innehåller alla kolumner definierade i utdatakolumnsamlingarna för en sekvens av komponenter. Till exempel, om var och en av fyra komponenter i en dataflödessekvens lägger till en utdatakolumn till sin utdatakolumnsamling, innehåller bufferten som ges till varje komponent fyra kolumner, en för varje utdatakolumn per komponent. På grund av detta beteende får en komponent ibland buffertar som innehåller kolumner den inte använder.

Eftersom buffertarna som tas emot av din komponent kan innehålla kolumner som komponenten inte kommer att använda, måste du hitta de kolumner du vill använda i komponentens in- och utdatakolumnsamlingar i bufferten som tillhandahålls komponenten av dataflödesuppgiften. Du gör detta genom att använda FindColumnByLineageID egenskapens BufferManager metod. Av prestandaskäl utförs denna uppgift vanligtvis under metoden PreExecute , snarare än i PrimeOutput eller ProcessInput.

PreExecute kallas före och PrimeOutputProcessInput metoderna, och är den första möjligheten för en komponent att utföra detta arbete efter att den blir BufferManager tillgänglig för komponenten. Under denna metod bör komponenten placera sina kolumner i buffertarna och lagra denna information internt så att kolumnerna kan användas i antingen eller PrimeOutputProcessInput metoderna.

Följande kodexempel visar hur en transformationskomponent med synkron utgång placerar sina indatakolumner i bufferten under 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

Att lägga till rader

Komponenter tillhandahåller rader till nedströmskomponenter genom att lägga till rader till PipelineBuffer objekt. Dataflödesuppgiften tillhandahåller en array av utdatabuffertar – en för varje IDTSOutput100 objekt som är kopplad till en nedströms komponent – som parameter för metoden PrimeOutput . Källkomponenter och transformationskomponenter med asynkrona utgångar lägger till rader i buffertarna och anropar SetEndOfRowset metoden när de är klara med att lägga till rader. Dataflödesuppgiften hanterar de utdatabuffertar som den levererar till komponenterna och, när en buffert blir full, flyttar den automatiskt raderna i bufferten till nästa komponent. Metoden PrimeOutput kallas en gång per komponent, till skillnad från metoden ProcessInput som kallas upprepade gånger.

Följande kodexempel visar hur en komponent lägger till rader i sina utdatabuffertar under metoden PrimeOutput , och sedan anropar metoden SetEndOfRowset .

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

För mer information om att utveckla komponenter som lägger till rader i utdatabuffertar, se Utveckling av en anpassad källkomponent och Utveckling av en anpassad transformationskomponent med asynkrona utgångar.

Mottagande rader

Komponenter tar emot rader från uppströms komponenter i PipelineBuffer objekt. Dataflödesuppgiften tillhandahåller ett PipelineBuffer objekt som innehåller rader som lagts till dataflödet av uppströmskomponenter som en parameter till metoden ProcessInput . Denna indatabuffert kan användas för att undersöka och modifiera rader och kolumner i bufferten, men kan inte användas för att lägga till eller ta bort rader. Metoden ProcessInput anropas upprepade gånger tills inga fler tillgängliga buffertar finns. Senast den kallas är egenskapen EndOfRowsetsann. Du kan iterera över samlingen av rader i bufferten genom att använda metoden NextRow , som för bufferten vidare till nästa rad. Denna metod returnerar false när bufferten är på sista raden i samlingen. Du behöver inte kontrollera egenskapen EndOfRowset om du inte måste utföra en ytterligare åtgärd efter att de sista dataraderna har behandlats.

Följande text visar det korrekta mönstret för att använda NextRow metoden och egenskapen EndOfRowset :

while (buffer.NextRow())

{

// Do something with each row.

}

if (buffer.EndOfRowset)

{

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

}

Följande kodexempel visar hur en komponent bearbetar raderna i indatabuffertar under metoden ProcessInput .

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

För mer information om utveckling av komponenter som tar emot rader i inmatningsbuffertar, se Utveckling av en anpassad destinationskomponent och Utveckling av en anpassad transformationskomponent med synkrona utgångar.

Relaterat innehåll

  • Uppdaterades senast