Sviluppo di un componente di trasformazione personalizzato con output sincroni

Si applica a:SQL Server Integration Runtime SSIS in Azure Data Factory

I componenti di trasformazione con output sincroni ricevono righe dai componenti a monte, quindi leggono o modificano i valori nelle colonne di queste righe mentre le passano ai componenti a valle. Possono anche definire colonne di output aggiuntive derivate dalle colonne fornite dai componenti a monte, ma non aggiungono righe al flusso di dati. Per altre informazioni sulla differenza tra componenti sincroni e asincroni, vedere Informazioni sulle trasformazioni sincrone e asincrone.

Questo tipo di componente è indicato per attività in cui i dati vengono modificati in linea quando vengono forniti al componente e in cui il componente non deve necessariamente vedere tutte le righe prima di elaborarle. Si tratta del componente più facile da sviluppare, perché le trasformazioni con output sincroni solitamente non si connettono a origini dati esterne, non gestiscono colonne di metadati esterne né aggiungono righe ai buffer di output.

La creazione di un componente di trasformazione con output sincroni richiede l'aggiunta di un oggetto IDTSInput100 che conterrà le colonne a monte selezionate per il componente e di un oggetto IDTSOutput100 che può contenere colonne derivate create dal componente. Include anche l'implementazione dei metodi della fase di progettazione e la specifica di codice che legge o modifica le colonne nelle righe del buffer in ingresso durante l'esecuzione.

In questa sezione vengono fornite le informazioni necessarie per implementare un componente di trasformazione personalizzato, con esempi di codice che consentono di comprenderne meglio i concetti.

Fase di progettazione

Il codice della fase di progettazione per questo componente richiede la creazione degli input e degli output, l'aggiunta di eventuali colonne di output aggiuntive generate dal componente e la convalida della configurazione del componente.

Creazione del componente

Gli input, gli output e le proprietà personalizzate del componente vengono in genere creati durante il metodo ProvideComponentProperties. Sono disponibili due modi per aggiungere l'input e l'output di un componente di trasformazione con output sincroni. È possibile utilizzare l'implementazione della classe di base del metodo e quindi modificare l'input e output predefiniti creati oppure aggiungere manualmente l'input e l'output in modo esplicito.

Nell'esempio di codice seguente è illustrata un'implementazione di ProvideComponentProperties che aggiunge in modo esplicito gli oggetti di input e di output. In un commento è inclusa la chiamata alla classe di base che consente di ottenere lo stesso risultato.

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  

Creazione e configurazione di colonne di output

Anche se i componenti di trasformazione con output sincroni non aggiungono righe ai buffer, possono aggiungere colonne di output aggiuntive all'output. In genere, quando un componente aggiunge una colonna di output, i relativi valori vengono derivati in fase di esecuzione dai dati contenuti in una o più colonne fornite al componente da un componente a monte.

Dopo la creazione di una colonna di output, è necessario impostare le relative proprietà del tipo di dati. L'impostazione delle proprietà del tipo di dati di una colonna di output richiede una gestione speciale e viene eseguita chiamando il metodo SetDataTypeProperties. Questo metodo è necessario perché le proprietà DataType, Length, Precision e CodePage sono singolarmente di sola lettura, in quanto ognuna dipende dalle impostazioni dell'altra. Questo metodo garantisce che i valori delle proprietà siano impostati in modo coerente e l'attività Flusso di dati verifica se sono stati impostati correttamente.

L'oggetto DataType della colonna determina i valori impostati per le altre proprietà. Nella tabella seguente sono illustrati i requisiti delle proprietà dipendenti per ogni oggetto DataType. Le proprietà dipendenti dei tipi di dati non elencati sono impostate su zero.

Poiché le restrizioni sulle proprietà del tipo di dati sono basate sul tipo di dati della colonna di output, è necessario scegliere il tipo di dati di Integration Services corretto quando si utilizzano tipi gestiti. La classe di base fornisce tre metodi di supporto, ConvertBufferDataTypeToFitManaged, BufferTypeToDataRecordType e DataRecordTypeToBufferType, per assistere gli sviluppatori di componenti gestiti nella selezione di un tipo di dati di SQL Server Integration Services dato un tipo gestito. Questi metodi convertono i tipi di dati gestiti nei tipi di dati di SQL Server Integration Services e viceversa.

Fase di esecuzione

Generalmente, l'implementazione della parte runtime del componente è suddivisa in due attività: individuazione delle colonne di input e output del componente nel buffer e lettura o scrittura dei valori di tali colonne nelle righe del buffer in ingresso.

Individuazione di colonne nel buffer

Il numero di colonne nei buffer forniti a un componente durante l'esecuzione supera in genere il numero di colonne presenti nelle raccolte di input o output del componente, perché ogni buffer contiene tutte le colonne di output definite nei componenti in un flusso di dati. Per assicurarsi che le colonne del buffer corrispondano correttamente alle colonne dell'input o dell'output, gli sviluppatori di componenti devono utilizzare il metodo FindColumnByLineageID di BufferManager. Questo metodo individua una colonna nel buffer specificato in base al relativo ID di derivazione. In genere, le colonne vengono individuate durante PreExecute perché si tratta del primo metodo di runtime in cui la proprietà BufferManager diventa disponibile.

Nell'esempio di codice seguente è illustrato un componente che individua gli indici delle colonne nella propria raccolta di colonne di input e output durante PreExecute. Gli indici delle colonne vengono archiviati in una matrice di valori integer e il componente può accedervi 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  

Elaborazione di righe

I componenti ricevono oggetti PipelineBuffer che contengono righe e colonne nel metodo ProcessInput. Durante questo metodo vengono scorse le righe nel buffer e vengono lette e modificate le colonne identificate durante PreExecute . Il metodo viene chiamato ripetutamente dall'attività Flusso di dati finché non vengono più fornite righe dal componente a monte.

Una singola colonna nel buffer viene letta o scritta tramite il metodo di accesso dell'indicizzatore di matrici oppure con il metodo Get o Set. I metodi Get e Set sono più efficienti e devono essere usati quando il tipo di dati della colonna nel buffer è noto.

Nell'esempio di codice seguente è illustrata un'implementazione del metodo ProcessInput che elabora le righe in ingresso.

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  

Esempio

Nell'esempio seguente è illustrato un componente di trasformazione semplice con output sincroni che converte i valori di tutte le colonne di tipo stringa in lettere maiuscole. Nell'esempio non vengono dimostrati tutti i metodi e le funzionalità descritti nell'argomento. Vengono descritti i metodi importanti di cui ogni componente di trasformazione personalizzato con output sincroni deve eseguire l'override, ma non è incluso codice per la convalida in fase di progettazione.

using System;  
using System.Collections;  
using Microsoft.SqlServer.Dts.Pipeline;  
using Microsoft.SqlServer.Dts.Pipeline.Wrapper;  
using Microsoft.SqlServer.Dts.Runtime.Wrapper;  
  
namespace Uppercase  
{  
  [DtsPipelineComponent(DisplayName = "Uppercase")]  
  public class Uppercase : PipelineComponent  
  {  
    ArrayList m_ColumnIndexList = new ArrayList();  
  
    public override void ProvideComponentProperties()  
    {  
      base.ProvideComponentProperties();  
      ComponentMetaData.InputCollection[0].Name = "Uppercase Input";  
      ComponentMetaData.OutputCollection[0].Name = "Uppercase Output";  
    }  
  
    public override void PreExecute()  
    {  
      IDTSInput100 input = ComponentMetaData.InputCollection[0];  
      IDTSInputColumnCollection100 inputColumns = input.InputColumnCollection;  
  
      foreach (IDTSInputColumn100 column in inputColumns)  
      {  
        if (column.DataType == DataType.DT_STR || column.DataType == DataType.DT_WSTR)  
        {  
          m_ColumnIndexList.Add((int)BufferManager.FindColumnByLineageID(input.Buffer, column.LineageID));  
        }  
      }  
    }  
  
    public override void ProcessInput(int inputID, PipelineBuffer buffer)  
    {  
      while (buffer.NextRow())  
      {  
        foreach (int columnIndex in m_ColumnIndexList)  
        {  
          string str = buffer.GetString(columnIndex);  
          buffer.SetString(columnIndex, str.ToUpper());  
        }  
      }  
    }  
  }  
}  

Imports System   
Imports System.Collections   
Imports Microsoft.SqlServer.Dts.Pipeline   
Imports Microsoft.SqlServer.Dts.Pipeline.Wrapper   
Imports Microsoft.SqlServer.Dts.Runtime.Wrapper   
Namespace Uppercase   
  
 <DtsPipelineComponent(DisplayName="Uppercase")> _   
 Public Class Uppercase   
 Inherits PipelineComponent   
   Private m_ColumnIndexList As ArrayList = New ArrayList   
  
   Public  Overrides Sub ProvideComponentProperties()   
     MyBase.ProvideComponentProperties   
     ComponentMetaData.InputCollection(0).Name = "Uppercase Input"   
     ComponentMetaData.OutputCollection(0).Name = "Uppercase Output"   
   End Sub   
  
   Public  Overrides Sub PreExecute()   
     Dim input As IDTSInput100 = ComponentMetaData.InputCollection(0)   
     Dim inputColumns As IDTSInputColumnCollection100 = input.InputColumnCollection   
     For Each column As IDTSInputColumn100 In inputColumns   
       If column.DataType = DataType.DT_STR OrElse column.DataType = DataType.DT_WSTR Then   
         m_ColumnIndexList.Add(CType(BufferManager.FindColumnByLineageID(input.Buffer, column.LineageID), Integer))   
       End If   
     Next   
   End Sub   
  
   Public  Overrides Sub ProcessInput(ByVal inputID As Integer, ByVal buffer As PipelineBuffer)   
     While buffer.NextRow   
       For Each columnIndex As Integer In m_ColumnIndexList   
         Dim str As String = buffer.GetString(columnIndex)   
         buffer.SetString(columnIndex, str.ToUpper)   
       Next   
     End While   
   End Sub   
 End Class   
End Namespace  

Vedere anche

Sviluppo di un componente di trasformazione personalizzato con output asincroni
Informazioni sulle trasformazioni sincrone e asincrone
Creazione di una trasformazione sincrona con il componente script