非同期出力型のカスタム変換コンポーネントの開発

[アーティクル]
12/15/2008

非同期出力型変換コンポーネントは、変換が入力行をすべて受け取るまで行を出力できないようにする場合や、変換が入力として受信した各行に対して必ずしも 1 つずつ出力行を生成しない場合に使用します。たとえば、集計変換では、行をすべて読み込むまではすべての行の合計を計算できません。これに対し、同期出力型コンポーネントは、データが渡されるたびに各データ行を変更する場合に使用します。各行のデータを順に変更したり、1 つまたは複数の列を新規作成して、各列にすべての入力行ごとの値を設定したりできます。同期コンポーネントと非同期コンポーネントの相違点の詳細については、「同期および非同期変換について」を参照してください。

非同期出力型変換コンポーネントは、変換先コンポーネントと変換元コンポーネントの、両方の機能を果たすという特徴があります。この種のコンポーネントは、上流コンポーネントから行を受け取り、下流コンポーネントで使用される行を追加します。これらの操作を両方とも行うデータフローコンポーネントは、他にはありません。

上流コンポーネントの列が同期出力型コンポーネントで使用可能な場合、自動的に、その下流にあるコンポーネントでも使用可能となります。したがって、同期出力型コンポーネントでは、出力列を定義しなくても次のコンポーネントに行や列を渡すことができます。これに対し、非同期出力型コンポーネントでは、下流コンポーネントに行を渡すために出力列を定義する必要があります。したがって、非同期出力型コンポーネントの方が、デザイン時や実行時に必要な処理が多いため、開発者はより多くのコードを実装する必要があります。

SQL Server 2005 Integration Services (SSIS) には非同期出力型の変換がいくつか組み込まれています。たとえば並べ替え変換では、並べ替えを行う前に行がすべて必要であるため、非同期出力を使用して変換を実行しています。行をすべて受け取ったら並べ替えを実行し、出力に行を追加します。

ここでは、非同期出力型の変換を開発する方法の詳細について説明します。非同期出力型の変換コンポーネントのサンプルについては、「Remove Duplicates コンポーネントサンプル」を参照してください。変換元コンポーネントの開発の詳細については、「カスタム変換元コンポーネントの開発」を参照してください。

デザイン時

コンポーネントの作成

IDTSOutput90 オブジェクトの SynchronousInputID プロパティは、同期出力か非同期出力かの区別を示します。非同期出力を作成するには、コンポーネントに出力を追加し、SynchronousInputID プロパティを 0 に設定します。また、このプロパティを設定することにより、PipelineBuffer オブジェクトをコンポーネントの入出力の両方に割り当てるか、単一のバッファを割り当てて 2 つのオブジェクトで共有するかをデータフロータスクが判断します。

次のサンプルコードは、ProvideComponentProperties を実装し、非同期出力を作成するコンポーネントを示します。

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


namespace Microsoft.Samples.SqlServer.Dts
{
    [DtsPipelineComponent(DisplayName = "AsyncComponent",ComponentType = ComponentType.Transform)]
    public class AsyncComponent : PipelineComponent
    {
        public override void ProvideComponentProperties()
        {
            // Call the base class, which adds a synchronous input
            // and output.
            base.ProvideComponentProperties();

            // Make the output asynchronous.
            IDTSOutput90 output = ComponentMetaData.OutputCollection[0];
            output.SynchronousInputID = 0;
        }
    }
}

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

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

    Public Overrides Sub ProvideComponentProperties()

        ' Call the base class, which adds a synchronous input
        ' and output.
        Me.ProvideComponentProperties()

        ' Make the output asynchronous.
        Dim output As IDTSOutput90 = ComponentMetaData.OutputCollection(0)
        output.SynchronousInputID = 0

    End Sub


End Class

出力列の作成と設定

前述したように、非同期型コンポーネントは出力列コレクションに列を追加し、下流コンポーネントに列を渡します。コンポーネントの要件に応じて、デザイン時に選択できる方法はいくつかあります。たとえば、上流コンポーネントからすべての列をそのまま下流コンポーネントに渡す場合は、OnInputPathAttached メソッドをオーバーライドして列を追加します。これが、コンポーネントで入力列が使用可能となる最初のメソッドだからです。

入力列として選択した列に基づいてコンポーネントが出力列を作成する場合は、出力列を選択してそれらの使い方を示すように、SetUsageType メソッドをオーバーライドします。

非同期型出力型コンポーネントで、上流コンポーネントの列に基づいて出力列を作成する場合、使用できる上流の列が変更されると、コンポーネントは出力列のコレクションを更新する必要があります。変更は Validate メソッド内で検出され、更新は ReinitializeMetaData メソッド内で行われます。

メモ :
出力列コレクションからある出力列を削除すると、データフローの下流に位置し、この列を参照しているコンポーネントに悪影響を及ぼします。下流コンポーネントが悪影響を受けないようにするには、列を削除して再作成する方法を取らずに、出力列を修復する必要があります。たとえば、ある列のデータ型を変更した場合、データ型を更新する必要があります。

出力列コレクションからある出力列を削除すると、データフローの下流に位置し、この列を参照しているコンポーネントに悪影響を及ぼします。下流コンポーネントが悪影響を受けないようにするには、列を削除して再作成する方法を取らずに、出力列を修復する必要があります。たとえば、ある列のデータ型を変更した場合、データ型を更新する必要があります。

次のコード例では、上流コンポーネントから渡される各列に対して出力列を作成し、出力列コレクションに追加する方法を示しています。

public override void OnInputPathAttached(int inputID)
{
   IDTSInput90 input = ComponentMetaData.InputCollection.GetObjectByID(inputID);
   IDTSOutput90 output = ComponentMetaData.OutputCollection[0];
   IDTSVirtualInput90 vInput = input.GetVirtualInput();

   foreach (IDTSVirtualInputColumn90 vCol in vInput.VirtualInputColumnCollection)
   {
      IDTSOutputColumn90 outCol = output.OutputColumnCollection.New();
      outCol.Name = vCol.Name;
      outCol.SetDataTypeProperties(vCol.DataType, vCol.Length, vCol.Precision, vCol.Scale, vCol.CodePage);
   }
}

Public Overrides Sub OnInputPathAttached(ByVal inputID As Integer)

    Dim input As IDTSInput90 = ComponentMetaData.InputCollection.GetObjectByID(inputID)
    Dim output As IDTSOutput90 = ComponentMetaData.OutputCollection(0)
    Dim vInput As IDTSVirtualInput90 = input.GetVirtualInput()

    For Each vCol As IDTSVirtualInputColumn90 In vInput.VirtualInputColumnCollection

        Dim outCol As IDTSOutputColumn90 = output.OutputColumnCollection.New()
        outCol.Name = vCol.Name
        outCol.SetDataTypeProperties(vCol.DataType, vCol.Length, vCol.Precision, vCol.Scale, vCol.CodePage)

    Next
End Sub

実行時

非同期出力型コンポーネントは、他の種類のコンポーネントとは実行時における処理手順も異なります。まず、PrimeOutput と ProcessInput の両方のメソッドに対する呼び出しを受信するのは、この型のコンポーネントのみです。また、非同期出力型コンポーネントは、処理を開始する前に入力行すべてにアクセスできるようになっている必要があります。そのため、EndOfRowset プロパティが true になるまで、入力行を内部キャッシュに保存する必要があります。最後に、他のコンポーネントとは異なり、非同期出力型コンポーネントは入力用と出力用の両方のバッファを受け取ります。

バッファについて

コンポーネントが入力バッファを受け取るのは、ProcessInput メソッドの実行中です。このバッファには、上流コンポーネントによってバッファに追加された行が格納されます。また、このバッファには、上流コンポーネントの出力に存在しながら、非同期型コンポーネントの入力コレクションに含まれていない列を含め、コンポーネントの入力の列が格納されます。

PrimeOutput メソッドからコンポーネントに提供される出力バッファには、最初は行が格納されていません。コンポーネントは行をこのバッファに追加し、バッファがいっぱいになった時点で下流コンポーネントに渡します。出力バッファには、他の下流コンポーネントが出力に追加したすべての列を含め、出力列コレクションで定義されている列が格納されます。

この動作は、単一のバッファを共有する同期出力型コンポーネントの動作とは異なります。同期出力型コンポーネントの共有バッファには、上流コンポーネントと下流コンポーネントの出力に追加された列を含め、コンポーネントの入力列および出力列の両方が格納されます。

行の処理

入力行のキャッシュ

非同期出力型コンポーネントを記述する場合、出力バッファに行を追加するには 3 つの方法があります。それは、入力行を受け取るたびに追加する方法、上流コンポーネントから行をすべて受け取るまでキャッシュする方法、またはコンポーネントを処理する上で適切な時点で追加する方法です。方法は、コンポーネントの要求に応じて選択します。たとえば並べ替えコンポーネントの場合、上流から行をすべて受け取ってからでないと並べ替え処理はできません。したがって、EndOfRowset を受け取るまで待機してから、出力バッファに行を追加します。

入力バッファで受け取った行は、処理の準備ができるまで、コンポーネントの内部にキャッシュする必要があります。入力バッファ内の行は、データテーブルや多次元アレイのほか、コンポーネントの内部構造の任意の場所にキャッシュできます。EndOfRowset が true になるまでの間、入力バッファ内の行を内部的にキャッシュする例については、「Remove Duplicates コンポーネントサンプル」を参照してください。

出力行の追加

出力バッファへの行の追加は、出力バッファに AddRow メソッドを呼び出すことによって行います。これは、行を受け取るたびに出力バッファに追加する場合でも、すべての行を受け取ってから追加する場合でも同じです。行を追加したら、新しい行の各列に値を設定します。

出力バッファには、コンポーネントの出力列コレクションにはない列が含まれる場合があるため、値を設定するには、バッファの該当する列のインデックスを探す必要があります。BufferManager プロパティの FindColumnByLineageID メソッドは、バッファ行のうち、指定された系列 ID を持つ列のインデックスを返します。これを使用して、バッファ列に値を割り当てます。

PreExecute メソッドは、PrimeOutput メソッドまたは ProcessInput メソッドの前に呼び出されますが、ここで初めて BufferManager プロパティが使用可能になり、初めて入力バッファおよび出力バッファの各列のインデックスを探すことができるようになります。

サンプル

次の例は、行を受け取るたびに出力バッファに行を追加する、非同期出力型の簡単な変換コンポーネントを示しています。ただし、この例ではここで説明したメソッドや機能のすべてが使われているわけではありません。カスタムの非同期出力型変換コンポーネントでオーバーライドしなければならない重要なメソッドは示していますが、デザイン時の検証のためのコードは含まれていません。また、ProcessInput メソッド内のコードは、入力列コレクションの各列に対応して、出力列コレクションに 1 つずつ出力列があるものと想定しています。非同期出力型の変換コンポーネントの完全なサンプルについては、「Remove Duplicates コンポーネントサンプル」を参照してください。

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

namespace Microsoft.Samples.SqlServer.Dts
{
   [DtsPipelineComponent(DisplayName = "AsynchronousOutput")]
   public class AsynchronousOutput : PipelineComponent
   {
      PipelineBuffer outputBuffer;
      int[] inputColumnBufferIndexes;
      int[] outputColumnBufferIndexes;

      public override void ProvideComponentProperties()
      {
         // Let the base class add the input and output objects.
         base.ProvideComponentProperties();

         // Name the input and output, and make the
         // output asynchronous.
         ComponentMetaData.InputCollection[0].Name = "Input";
         ComponentMetaData.OutputCollection[0].Name = "AsyncOutput";
         ComponentMetaData.OutputCollection[0].SynchronousInputID = 0;
      }
      public override void PreExecute()
      {
         IDTSInput90 input = ComponentMetaData.InputCollection[0];
         IDTSOutput90 output = ComponentMetaData.OutputCollection[0];

         inputColumnBufferIndexes = new int[input.InputColumnCollection.Count];
         outputColumnBufferIndexes = new int[output.OutputColumnCollection.Count];

         for (int col = 0; col < input.InputColumnCollection.Count; col++)
            inputColumnBufferIndexes[col] = BufferManager.FindColumnByLineageID(input.Buffer, input.InputColumnCollection[col].LineageID);

         for (int col = 0; col < output.OutputColumnCollection.Count; col++)
            outputColumnBufferIndexes[col] = BufferManager.FindColumnByLineageID(output.Buffer, output.OutputColumnCollection[col].LineageID);

      }

      public override void PrimeOutput(int outputs, int[] outputIDs, PipelineBuffer[] buffers)
      {
         if (buffers.Length != 0)
            outputBuffer = buffers[0];
      }
      public override void ProcessInput(int inputID, PipelineBuffer buffer)
      {
         // Check to see whether there are rows in the buffer.
         if (!buffer.EndOfRowset)
         {
            // Advance the buffer to the next row.
            while (buffer.NextRow())
            {
               // Add a row to the output buffer.
               outputBuffer.AddRow();
               for (int x = 0; x < inputColumnBufferIndexes.Length; x++)
               {
                  // Copy the data from the input buffer column to the output buffer column.
                  outputBuffer[outputColumnBufferIndexes[x]] = buffer[inputColumnBufferIndexes[x]];
               }
            }
         }
         else
         {
            // EndOfRowset on the input buffer is true.
            // Set EndOfRowset on the output buffer.
            outputBuffer.SetEndOfRowset();
         }
      }
   }
}

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

Namespace Microsoft.Samples.SqlServer.Dts

    <DtsPipelineComponent(DisplayName:="AsynchronousOutput")> _
    Public Class AsynchronousOutput


        Inherits PipelineComponent

        Private outputBuffer As PipelineBuffer
        Private inputColumnBufferIndexes As Integer()
        Private outputColumnBufferIndexes As Integer()

        Public Overrides Sub ProvideComponentProperties()

            ' Let the base class add the input and output objects.
            Me.ProvideComponentProperties()

            ' Name the input and output, and make the
            ' output asynchronous.
            ComponentMetaData.InputCollection(0).Name = "Input"
            ComponentMetaData.OutputCollection(0).Name = "AsyncOutput"
            ComponentMetaData.OutputCollection(0).SynchronousInputID = 0
        End Sub

        Public Overrides Sub PreExecute()

            Dim input As IDTSInput90 = ComponentMetaData.InputCollection(0)
            Dim output As IDTSOutput90 = ComponentMetaData.OutputCollection(0)

            ReDim inputColumnBufferIndexes(input.InputColumnCollection.Count)
            ReDim outputColumnBufferIndexes(output.OutputColumnCollection.Count)

            For col As Integer = 0 To input.InputColumnCollection.Count
                inputColumnBufferIndexes(col) = BufferManager.FindColumnByLineageID(input.Buffer, input.InputColumnCollection(col).LineageID)
            Next

            For col As Integer = 0 To output.OutputColumnCollection.Count
                outputColumnBufferIndexes(col) = BufferManager.FindColumnByLineageID(output.Buffer, output.OutputColumnCollection(col).LineageID)
            Next

        End Sub
        Public Overrides Sub PrimeOutput(ByVal outputs As Integer, ByVal outputIDs As Integer(), ByVal buffers As PipelineBuffer())

            If buffers.Length <> 0 Then
                outputBuffer = buffers(0)
            End If

        End Sub

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

            ' Check to see whether there are rows in the buffer.
            If buffer.EndOfRowset = False Then

                ' Advance the buffer to the next row.
                While (buffer.NextRow())

                    ' Add a row to the output buffer.
                    outputBuffer.AddRow()
                    For x As Integer = 0 To inputColumnBufferIndexes.Length

                        ' Copy the data from the input buffer column to the output buffer column.
                        outputBuffer(outputColumnBufferIndexes(x)) = buffer(inputColumnBufferIndexes(x))

                    Next
                End While
            Else

                ' EndOfRowset on the input buffer is true.
                ' Set the end of row set on the output buffer.
                outputBuffer.SetEndOfRowset()
            End If
        End Sub
    End Class
End Namespace

参照

概念

同期出力型のカスタム変換コンポーネントの開発
 同期および非同期変換について
 スクリプトコンポーネントによる非同期変換の作成

ヘルプおよび情報

SQL Server 2005 の参考資料の入手

次の方法で共有