| Oracle® Fusion Middleware Oracle Data Integratorナレッジ・モジュール開発者ガイド 12c (12.1.2) E49826-02 |
|
 前 |
 次 |
| Oracle® Fusion Middleware Oracle Data Integratorナレッジ・モジュール開発者ガイド 12c (12.1.2) E49826-02 |
|
前 |
次 |
この章では、マッピングで使用される統合戦略について説明します。この戦略は統合ナレッジ・モジュールに実装されます。
この章では、次の項目について説明します。
統合プロセスは、マッピングで常に必要になります。このプロセスは、一時統合表を使用して、ソースからのデータの統合や、ターゲット・データストアへの表のロードを行います。
統合プロセスは、統合プロセスに必要な手順を定義する統合戦略を使用します。次に統合戦略の例を示します。
追加: オプションとして、ターゲット・データストア内のすべてのレコードを削除し、すべてのフローをターゲットに挿入します。
制御追加: オプションとして、ターゲット・データベース内のすべてのレコードを削除し、すべてのフローをターゲットに挿入します。この戦略にはオプションのフロー制御が含まれます。
増分更新: オプションとして、ターゲット・データベース内のすべてのレコードを削除します。フローをターゲットと比較して、新しいレコードと既存のレコードを識別します。次に新しいレコードをターゲットに挿入して既存のレコードを更新します。この戦略にはオプションのフロー制御が含まれます。
緩やかに変化するディメンション: タイプ2の緩やかに変化するディメンションを実装し、変更時にターゲット・レコードでの単純な更新が必要なフィールド、前のレコードの状態の履歴が必要なフィールドを識別します。
このフェーズでは、ステージング領域とターゲットが同じデータ・サーバーに存在する場合には単一サーバーが、ステージング領域とターゲットが別のサーバー上にある場合には2つのサーバーが関与します。
統合プロセスは、使用される戦略によって大きく異なります。
統合プロセスでは次の要素が使用されます。
統合表(フロー表とも呼ばれる)は、ステージング領域の変換がすべて行われた後でデータをステージングするために必要になることがあります。このロード表は、ターゲット表名の前に接頭辞I$を付けた名前になります。この統合表は、戦略を実装するために必要な追加フィールドを含むターゲット表のイメージです。この表内のデータにはフラグが付き、ターゲット表に統合される前に変換またはチェックされます。
LKMによって作成されたソース表またはロード表(あるいはその両方) 。統合プロセスでは、これらの表から統合表または直接ターゲット表にデータがロードされます。
チェック・ナレッジ・モジュール。IKMによりフロー・チェック・フェーズが開始され、統合表のデータがターゲット表の一部の制約に適合しているかチェックされます。不正なデータは統合表から削除されます(フローから削除)。
挿入、更新、UD1などのマッピング・メタデータ、または緩やかに変化するディメンションの動作などのモデル・メタデータは、統合フェーズで統合戦略の属性レベルの動作をパラメータ設定するために使用されます。
一般的な統合プロセスは、次のようになります。
必要に応じて一時統合表を作成します。たとえば、 値がIまたはUの更新フラグにより、挿入する行と更新する行を区別します。
ソースとロード表からこの統合表にデータをロードし、ステージング領域で指定された変換(結合、フィルタ、マッピング)を実行します。
統合表で変換を実行し、統合戦略を実施します。たとえば、統合表の内容をターゲット表と比較して、更新フラグを設定します。
統合表からターゲット表にデータをロードして内容を変更します。
次の項では、Oracle Data Integratorで使用される一部の統合戦略について説明します。これらは次の2つの種類に分類されます。
この戦略は、ステージング領域スキーマがターゲット表スキーマと同じデータ・サーバーにある場合に使用されます。この構成では、複雑な統合を実行できます。
この戦略は、受け取ったデータ・フローをターゲット・データストアに単純に挿入します。 できれば事前にターゲットの内容を削除します。
この統合戦略には次の手順が含まれます。
ターゲット表からすべてのレコードを削除します(または切り詰めます)。この手順は、通常、KMオプションに依存します。
同じサーバー上にあるソースおよびステージング領域のロード表のデータを変換して挿入します。リモート・ソース・データを処理する場合、LKMにはすでにロード表が用意されています。同じサーバーにあるソースは、直接読み込むことができます。統合操作には、直接的なINSERT/SELECT文が利用され、SELECT句にはステージング領域で実行されるすべての変換、INSERT句にはターゲットでのすべての変換が含まれます。
トランザクションをコミットします。ターゲット上で実行される操作はトランザクション内で行い、操作がすべて完了してからコミットする必要があります。コミットは一般的にはCOMMITと呼ばれるKMオプションによってトリガーされます。
同じ統合戦略は、制御追加戦略を使用して、有効なフロー制御を選択せずに取得できます。
追加戦略では、データはフロー制御を使用せずに単純にターゲット表に挿入されます。この方法は、フロー・データを統合表(I$)に格納してから、CKMをコールしてエラーのあるレコードをエラー表(E%)に隔離する別の手順を追加することによって改善できます。
この統合戦略には次の手順が含まれます。
ステージング領域で統合表を削除(存在する場合)して作成します。これは、フロー制御のためにCKMに渡せるように、ターゲット表と同じ属性を使用して作成されます。
ソースのロード表に、追加戦略でターゲットをロードするのと同じ単一のINSERT/SELECT文を使用してデータを挿入します。
フロー制御のためにCKMをコールします。CKMは統合表データに対してターゲット表に定義された各制約を評価します。エラー表を作成し、エラーのあるレコードをこの表に挿入します。また、エラーのあるレコードを統合表から削除します。
CKMが完了すると、統合表に含まれるのは有効なレコードのみになります。これでターゲット表への挿入を安全に行えるようになりました。
ターゲット表からすべてのレコードを削除します。この手順は、マッピングの設計者によって設定されるオプション値に依存させることができます。
単一のINSERT/SELECT文で、統合表からターゲット表へレコードを追加します。
トランザクションをコミットします。
一時統合表を削除します。
エラーのリサイクル
場合によっては、フローに追加して再びターゲットに適用されるように、前の実行によるエラーをリサイクルすると便利です。この方法は、存在しない可能性のある商品IDを参照する、1日の販売取引を受け取る場合などに役立ちます。たとえば、参照される商品IDが商品表に存在しないため、販売レコードが拒否されてエラー表に隔離されたとします。これは、マッピングの最初の実行時に発生します。まもなく、データ管理者によって欠落している商品IDが作成されます。これにより、拒否されたレコードが有効になるため、マッピングの次の実行時にターゲットに適用しなおす必要が出てきます。
このメカニズムは、エラー表から統合表へのこのマッピングの前の実行で拒否されたレコードをすべて挿入する追加のタスクによって、IKMに実装されます。この操作は、データ品質チェックのためにCKMをコールする前に行われ、通常はRECYCLE_ERRORSと呼ばれるKMオプションによって条件設定されます。
増分更新戦略は、更新キーと呼ばれる一連の属性に従って、フローのレコードをターゲットの既存のレコードと比較して、ターゲット表にデータを統合するために使用されます。同じ更新キーを持つレコードは、関連付けられているデータが同じでない場合に更新されます。ターゲットに存在しないレコードは挿入されます。多くの場合、この戦略は、変更されたレコードを追跡する必要がない場合に、ディメンション表に使用されます。
このようなIKMの課題は、パフォーマンスの問題につながることが多い行単位の処理方法を使用するかわりに、セット指向のSQLに基づくプログラミングを使用してすべての操作を実行することです。このような戦略を構築するための最も一般的な方法は、通常、変換済実行ユニットを格納する統合表(I$)に依存します。この方法の詳細は次のとおりです。
ステージング領域で統合表を削除(存在する場合)して作成します。これは、フロー制御のためにCKMに渡せるように、ターゲット表と同じ属性を使用して作成されます。この表には、挿入対象のレコード(I)および更新対象のレコード(U)のフラグ付けに使用されるIND_UPDATE属性も含まれます。
ソースのロード表のデータが変換され、単一のINSERT/SELECT文を使用してロード表に挿入されます。IND_UPDATE属性はデフォルトで"I"に設定されます。
RECYCLE_ERROR KMオプションが選択されている場合は、前の実行で拒否されたレコードをリサイクルして統合表に追加します。
フロー制御のためにCKMをコールします。CKMは統合表データに対してターゲット表に定義された各制約を評価します。エラー表を作成し、エラーのあるレコードをこの表に挿入します。また、エラーのあるレコードを統合表から削除します。
統合表を更新して、ターゲットと同じ更新キーの値を持つすべてのレコードに対してIND_UPDATEフラグを「U」に設定します。したがって、すでにターゲットに存在するレコードには「U」フラグが付きます。この手順には通常、UPDATE/SELECT文を使用します。
統合表を再び更新し、すでに「U」フラグが付いており、属性の値がターゲットと完全に同じであるすべてのレコードに対してIND_UPDATE属性を「N」に設定します。これらのフロー・レコードは、ターゲット・レコードと完全に一致するため、ターゲット・データの更新に使用する必要はありません。
この手順が終了すると、統合表にフラグ付きのレコードが含まれるため、ターゲットに変更を適用するための統合表の準備が完了します。
I: ターゲットに挿入するレコードです。
U: ターゲットの更新に使用するレコードです。
N: すでにターゲットに存在するため無視するレコードです。
統合表の「U」フラグ付きレコードを使用してターゲットを更新します。操作するデータの量を最小限にするために、一般的には更新文をINSERT文の前に実行することに注意してください。
統合表の「I」フラグ付きレコードをターゲットに挿入します。
トランザクションをコミットします。
一時統合表を削除します。
最適化
この方法は、基礎となるデータベースに応じて最適化できます。このような最適化の例を次に示します。
Teradataでは、フローデータとターゲット表の間に左側外部結合を使用して、IND_UPDATE属性がすでに適切に設定されている統合表を移入する方が効率的な場合があります。
Oracleでは、場合によっては、UPDATE文、INSERT文の順に使用するよりも、ターゲット表でMERGE INTO文を使用する方が効率的なことがあります。
更新キー
更新キーは、常に一意であることが必要です。ほとんどの場合、主キーが更新キーとして使用されます。ただし、ID属性、ランク関数、または順序のように、増分を使用して自動的に算出される場合は、主キーを使用できません。この場合は、ソースにある、属性に基づく更新キーを使用する必要があります。
Nullの比較
更新の対象外を決定するためにデータの値を比較する場合、統合表とターゲット表の間の結合は、属性ごとに次のように表されます。
<target_table>.AttributeN = <loading_table>.AttributeN or (<target_table> is null and <loading_table>.AttributeN is null)
これを行うことで、NULL値を別のNULL値と一致させるためのNULL値の比較が許可されます。より簡潔に記述するには、結合関数を使用します。したがって、WHERE条件は次のように記述できます。
<%=odiRef.getColList("","coalesce(" + odiRef.getTable("L", "INT_NAME", "A") + ".[COL_NAME], 0) = coalesce(T.[COL_NAME], 0)", " \nand\t", "", "((UPD and !TRG) and !UK) ")%>
属性レベルの挿入/更新動作
UPDATE文によって更新された属性は、INSERT文で使用されるものと同じではありません。UPDATE文では、セレクタ「UPD and not UK」を使用して、マッピング内の「Update」のマークが付いたマッピングのうち、更新キーに属さない属性マッピングのみがフィルタ処理されます。INSERT文では、セレクタ「INS」を使用して、マッピング内の「insert」のマークが付いた属性マッピングのみが取得されます。
トランザクション
ターゲットのUPDATE文とINSERT文が、同じトランザクションに属することが重要です。1つでも条件を満たさない場合は、ターゲットへのデータの挿入または更新は一切行われません。
タイプ2の緩やかに変化するディメンション(SCD)は、データ・ウェアハウスのロードに使用される戦略です。多くの場合、特定の属性で発生した変更を追跡するために、ディメンション表のロードに使用されます。緩やかに変化するディメンションの一般的な表には、次の属性が含まれます。
サロゲート・キー。これは通常は数値属性で、ID属性、ランク関数、または順序を使用して自動生成された番号が含まれます。
ナチュラル・キー。業務システムの実際の主キーを表す属性のリストです。
変更時に上書きする必要のある属性。
変更時に行を追加する必要のある属性。
データ・ウェアハウスでレコードが作成された日付を示す開始日属性。
レコードが不要になった日付(終了日)を示す終了日属性。
レコードが現在のレコード(1)か古いレコード(0)かを示す、現在のレコード・フラグ。
次の例は、緩やかに変化するディメンションの動作を示しています。
業務システムでは、商品は主キーの役割をするIDによって定義されます。各商品には、名前、サイズ、仕入先、およびファミリがあります。業務システムで仕入先またはファミリが更新されるたびに、この商品の新しいバージョンがデータ・ウェアハウスに格納されます。
この例では、2006年3月12日にデータ・ウェアハウスで商品のディメンションが最初に初期化されています。すべてのレコードが挿入され、算出されたサロゲート・キーおよび2400年1月1日に設定された偽の終了日が割り当てられています。これらのレコードは業務システムの現在の状態を表すため、現在のレコード・フラグは1に設定されています。最初のロードの後、業務システムで次の変更が発生します。
商品P1の仕入先が更新されます。
商品P2のファミリが更新されます。
商品P3の名前が更新されます。
商品P5が追加されます。
これらの更新によるデータ・ウェアハウス・ディメンションへの影響は、次のとおりです。
P1の仕入先の更新の結果、新しい現在のレコード(サロゲート・キー5)が作成され、前のレコード(サロゲート・キー1)が終了します。
P2のファミリの更新の結果、新しい現在のレコード(サロゲート・キー6)が作成され、前のレコード(サロゲート・キー2)が終了します。
P3の名前の更新では、単純にターゲット・レコードがサロゲート・キー3を使用して更新されます。
新しい商品P5により、新しい現在のレコード(サロゲート・キー7)が作成されます。
この動作を実装するナレッジ・モジュールを作成するには、サロゲート・キー、ナチュラル・キー、開始日などの役割をする属性を把握しておく必要があります。Oracle Data Integratorでは、モデルの各属性に対する「説明」タブの「緩やかに変化するディメンションの動作」フィールドに、この情報が格納されます。
マッピングでこのようなデータストアを移入する場合、IKMは、getColList()置換メソッドでSCD_xxセレクタを使用して、このメタデータにアクセスします。
Oracle Data Integratorでは、タイプ2の緩やかに変化するディメンションの実装は、次のように行われます。
ステージング領域で統合表を削除(存在する場合)して作成します。
ナチュラル・キーの属性、変更時に上書きする属性および変更時に行を追加する属性に適用するマッピングのみを使用して、統合表にフロー・データを挿入します。開始日を現在の日付に設定し、終了日を定数に設定します。
拒否された前のレコードをリサイクルします。
フローに対するデータ品質チェックを実行するために、CKMをコールします。
ターゲットの現在のレコードと比較して、ナチュラル・キーの列と変更時に行を追加する列が変更されていない場合は、統合表のレコードに「U」フラグを付けます。
「U」フラグでフィルタ処理された統合表の内容を使用して、変更時に上書きされる列でターゲットを更新します。
(統合表にナチュラル・キーが存在する)古いレコードを終了し、そのレコードの現在のレコード・フラグを0に、終了日を現在の日付にそれぞれ設定します。
現在のレコード・フラグが1に設定されている、新しい変更レコードを挿入します。
統合表を削除します。
もう一度、この方法を適用できます。場合によっては、作成されたSQLを、さらに調整および最適化する必要があります。
この戦略は、ステージング領域をターゲット・データストアと同じデータ・サーバー上に配置できない場合に使用されます。この設定は、主には変換機能のないデータ・サーバー(たとえばファイル・サーバーなど)に使用されます。この構成では、単純な統合戦略のみが可能です。
ソースが単一のファイルで構成されており、そのファイルを最も効率的な方法でターゲット表に直接ロードする場合があります。Oracle Data Integratorのデフォルト設定では、ステージング領域をターゲット・サーバー上に置き、LKMを使用してソース・ファイルをロード表にステージングし、それからIKMを使用してロードされたデータをターゲット表に統合することが推奨されます。
ソース・データが変換されていない場合は、ロード・フェーズは不要です。
この状況では、ファイル・データをターゲットに直接ロードするIKMを使用します。そのためには、ステージング領域をソース・ファイルの論理スキーマに設定する必要があります。この設定を行うと、Oracle Data Integratorによってリモート・ステージング領域とターゲット間でデータを移動する複数接続IKMの使用が自動的に推奨されます。
このようなIKMではローダーが使用され、次のような手順が含まれます。
適切なロード・ユーティリティ・スクリプトを生成します。
ローダー・ユーティリティを実行します。
このようなKMの例として、IKMファイルからTeradata (TTU)があります。
ターゲットと異なるステージング領域を使用し、このステージング領域をRDBMSに設定する場合は、ステージング領域からリモート・ターゲットに変換済データを移動するIKMを使用できます。このタイプのIKMはLKMに非常に似ており、同じ規則に従います。
エージェントを使用する場合の手順は、通常は次のようになります。
ターゲット表からすべてのレコードを削除します(または切り詰めます)。この手順は、通常、KMオプションに依存します。
ステージング領域からターゲットへデータを挿入します。この手順には、ステージング領域で実行される「ソースに対するコマンド」タブのSELECT文が含まれます。INSERT文は「ソースに対するコマンド」タブのバインド変数を使用して記述され、ターゲット表でバッチごとに実行されます。
IKM SQLからSQLへの追加は、このようなKMの典型的な例です。
この戦略の一種として、ステージング領域からターゲットへのデータのロードに、エージェントのかわりにローダーまたはデータベース固有のメソッドを使用する方法もあります。
ターゲット・データストアがファイルまたはJMSキューまたはトピックの場合、ステージング領域はターゲット以外の場所に設定されます。したがって、ファイルをターゲットにする場合やデータストアをキューする場合は、ステージング領域からこのターゲットへ変換済データを統合する複数接続IKMを使用する必要があります。このデータ移動を実行する方法は、ターゲット・テクノロジによって異なります。たとえば、エージェントやターゲットの特定の機能(Java APIなど)を使用できます。
このようなIKMの一般的な手順は、次のようになります。
オプションに依存するターゲット・ファイルまたはキューをリセットします。
ステージング領域からファイルまたはキューへデータをアンロードします。
この項では、統合戦略とカスタマイズの例を示します。
すべてのソース・データがすでにステージング領域に存在する場合、既存のターゲット表にデータを統合するための最も単純な戦略は、ターゲットでレコードを置換して挿入することです。したがって、最も単純なIKMは、次の2つの手順で構成されます。
ターゲット表からすべてのレコードを削除します。この手順は、マッピングの設計者によって設定されるオプションに依存させることができます。
すべてのデータセットのソース・レコードを変換して挿入します。リモート・ソース・データを処理する場合、LKMには、事前変換済の結果セットを使用したロード表がすでに用意されています。ターゲット(およびステージング領域)と同じサーバーにあるソース・データ・セットをマッピングが使用する場合、データ・セットは他のロード表に結合されます。そのため、統合操作は、ターゲットTeradataボックスの変換能力をすべて利用する単純なINSERT/SELECT文になります。
これらの手順の詳細を次の例で示します。
このタスクはターゲット表からデータを削除します。この命令はトランザクションで実行され、コミットされません。DELETE_ALLナレッジ・モジュール・オプションが選択されている場合に実行されます。
ターゲットに対するコマンド
delete from <%=odiRef.getTable("L","INT_NAME","A")%>
このタスクは、ステージング表からターゲット表に行を挿入します。このコマンドはターゲットで行われるすべての操作と同じトランザクションで実行され、コミットされません。最後のコミット・トランザクション・コマンドにより、ターゲット上でコミットがトリガーされます。
このコマンドにより、マッピングに定義された別のデータセットのデータが選択されます。forループを使用してすべてのデータセットを検討し、各データセットに対してSELECT問合せを生成します。生成された問合せは設定ベースの操作(UNION、INTERSECTなど)を使用してマージされ、結果のデータ・フローがターゲット表に挿入されます。
ターゲットに対するコマンド
insert into <%=odiRef.getTable("L","TARG_NAME","A")%> ( <%=odiRef.getColList("", "[COL_NAME]", ",\n\t", "", "((INS and !TRG) and REW)")%> <%=odiRef.getColList(",", "[COL_NAME]", ",\n\t", "", "((INS and TRG) and REW)")%> ) select <%=odiRef.getColList("", "[COL_NAME]", ",\n\t", "", "((INS and !TRG) and REW)")%> <%=odiRef.getColList(",", "[EXPRESSION]", ",\n\t", "", "((INS and TRG) and REW)")%> FROM ( <%for (int i=0; i < odiRef.getDataSetCount(); i++){%><%=odiRef.getDataSet(i, "Operator")%>select <%=odiRef.getPop("DISTINCT_ROWS")%> <%=odiRef.getColList(i,"", "[EXPRESSION] [COL_NAME]", ",\n\t", "", "((INS and !TRG) and REW)")%> from <%=odiRef.getFrom(i)%>where <% if (odiRef.getDataSet(i, "HAS_JRN").equals("1")) { %> JRN_FLAG <> 'D '<%} else {%> (1=1) <% } %><%=odiRef.getJoin(i)%><%=odiRef.getFilter(i)%><%=odiRef.getJrnFilter(i)%><%=odiRef.getGrpBy(i)%><%=odiRef.getHaving(i)%><%}%>)
プロジェクトの要件の1つが、現在のデータをロードする前に各データ・ウェアハウス表をバックアップすることです。この要件は、たとえば大きな問題が発生したときに、データ・ウェアハウスを前の状態にリストアする場合に役立ちます。バックアップ表の名前は、データ表の名前に接尾辞「_BCK」を追加したものになります。
この要件を解決する最初の方法として、各ターゲット・データストアのデータを対応するバックアップ・データストアに複製するためのマッピングの開発があげられます。これらのマッピングは、データ・ウェアハウスを移入するマッピングより先にトリガーされます。あいにく、この解決方法では、ターゲット・データストアごとに追加のマッピングを作成する必要があるため、大規模な開発およびメンテナンスが発生します。開発およびメンテナンスするマッピングの数は、少なくとも2倍になります。
より簡潔な解決方法として、ターゲット・データストアの移入に使用するIKMで、この動作を実装することがあげられます。これは、ターゲットを変更する直前に作成してバックアップ表に移入する単一のCREATE AS SELECT文を使用して行います。したがって、バックアップ操作が自動化され、インタフェース開発者の介入が不要になります。
この例では、IKM Oracle増分更新でこの動作を実装する方法を示します。
ターゲットを変更する既存行の更新および新しい行の挿入タスクの前に、次のタスクを追加します。
このタスクは、バックアップ表を削除します。
ターゲットに対するコマンド
Drop table <%=odiRef.getTable("L","TARG_NAME","A")%>_BCK
一部のデータ・ウェアハウス・プロジェクトでは、法的コンプライアンスのために、ターゲット表に対する挿入または更新の各操作を追跡する必要があります。このような追跡により、業務分析者は、特定の期間にデータに何が発生したかを把握することができます。
この動作は、緩やかに変化するディメンション・ナレッジ・モジュールを使用して実現できる場合でも、フロー・データをターゲット表に適用する前に、単純にそのコピーを作成して実現することが可能です。
各ターゲット表に、接尾辞「_RGG」の対応する追跡表があり、すべてのデータ列と次のような追加の法的コンプライアンスの列が含まれるとします。
ジョブID
ジョブ名
操作の日時
操作の種類(挿入または更新)
ターゲットに挿入および更新を適用した後、IKMの終了直前に統合表から直接この表を移入するとします。
たとえば、Oracle増分更新IKMの場合、ターゲットを変更する既存行の更新および新しい行の挿入タスクの直後に次のタスクを追加します。
このタスクは、追跡表にデータをロードします。
ターゲットに対するコマンド
insert into <%=odiRef.getTable("L","TARG_NAME","A")%>_RGC(JOBID,JOBNAME,OPERATIONDATE,OPERATIONTYPE,<%=odiRef.getColList("", "[COL_NAME]", ",\n\t", "")%>)select <%=odiRef.getSession("SESS_NO")%> /* JOBID */,<%=odiRef.getSession("SESS_NAME")%> /* JOBNAME */,Current_timestamp /* OPERATIONDATE */,Case when IND_UPDATE = 'I' then 'Insert' else 'Update' end /* OPERATIONTYPE */,<%=odiRef.getColList("", "[COL_NAME]", ",\n\t", "")%>from <%=odiRef.getTable("L","INT_NAME","A")%>where IND_UPDATE <> 'N'
もちろん、追跡表が存在しない場合は、IKMを使用して追跡表を自動作成することによってカスタマイズを拡張できます。
