データ・フローを使用したデータの収集および変換

データ・フローは、データベースやフラット・ファイルなどのソース・データ・アセットからデータ・レイクやデータ・ウェアハウスなどのターゲット・データ・アセットへのデータのフローを表す論理ダイアグラムです。

ソースからターゲットへのデータのフローで、データを集計、クレンジングおよび整形するための一連の変換を実行できます。これにより、データ・エンジニアおよびETL開発者は、インサイトを分析または収集し、そのデータを使用してビジネス上の大きな意思決定を下せるようになります。

このチュートリアルでは:

データ・フローを保存できるプロジェクトを作成します。
ソース演算子を追加し、データ・フローで使用するデータ・エンティティを選択します。
整形演算子を使用し、変換を適用します。
データをロードするためのターゲット・データ・アセットを識別します。

開始する前に

データ・フローを使用してデータを収集および変換するには、次のものが必要です:

データ統合ワークスペースへのアクセス権。データ統合への接続を参照してください。
作成されたソース・データ・アセットおよびターゲット・データ・アセット。
ステージング・バケットで有効になっているPAR_MANAGE権限。
```
allow any-user to manage buckets in compartment <compartment-name> where ALL {request.principal.type = 'disworkspace', request.principal.id = '<workspace-ocid>', request.permission = 'PAR_MANAGE'}
```
自律型データベースは、データのステージングにオブジェクト・ストレージを使用し、事前認証済リクエストが必要です。

1. プロジェクトおよびデータ・フローの作成

Oracle Cloud Infrastructure Data Integrationでは、データ・フローおよびタスクはプロジェクトまたはフォルダ内にのみ作成できます。

プロジェクトおよびデータ・フローを作成するには:

ワークスペースのホーム・ページで、「プロジェクト」をクリックします。

タブ・バーで「開く」タブ(プラス・アイコン)をクリックして、「プロジェクト」を選択することもできます。
「プロジェクト」ページで、「プロジェクトの作成」をクリックします。
「プロジェクトの作成」ページで、「名前」にDI_Labと入力し、「作成」をクリックします。

プロジェクトを作成した後は、顧客(CUSTOMERS.json)および注文(REVENUE.csv)の情報を含む2つのソース・ファイルからデータを収集するためのデータ・フローを作成できます。
DI_Labプロジェクトの詳細ページで、左側のサブメニューで「データ・フロー」をクリックします。
「プロジェクト」ページを表示している場合は、DI_Labをクリックしてプロジェクトの詳細ページを開きます。
「データ・フローの作成」をクリックします。

データ・フロー・デザイナが新しいタブで開きます。
「プロパティ」パネルで、「名前」にLoad Customers and Revenue Dataと入力します。

「識別子」の値は、データ・フロー名に入力した値に基づいて自動的に生成されます。生成された値を変更できますが、データ・フローを保存した後は、識別子の更新は許可されません。
「作成」をクリックします。

デザイナは開いたままになっており、編集を続行できます。

2. ソース演算子の追加

ソース演算子を追加して、データ・フローに使用するデータ・エンティティを識別します。このチュートリアルでは、データ・エンティティはデータベース表を表します。

演算子の詳細を参照してください。

「演算子」パネルから、ソース演算子をキャンバスにドロップします。
キャンバスで、SOURCE_1を選択します(まだ選択されていない場合)。

「プロパティ」パネルに、フォーカスがある演算子の詳細が表示されます。
「プロパティ」パネルの「詳細」タブで、次の各オプションの横にある「選択」をクリックして選択します。
- 「データ・アセット」で、「Data_Lake」を選択します。
- 「接続」で、「デフォルトの接続」を選択します。
- 「スキーマ」で、コンパートメント、バケットの順に選択します。このチュートリアルの目的上、オブジェクト・ストレージがソース・データ・アセットとなるため、ここではバケットを選択します。
- 「データ・エンティティ」で、「名前で参照」をクリックし、「CUSTOMERS.json」を選択します。「ファイル・タイプ」で、「JSON」を選択します。
SOURCE_1の選択を完了すると、データ・エンティティの選択を反映して演算子名がCUSTOMERS_JSONになります。「識別子」フィールドで、ソース演算子の名前をCUSTOMERSに変更します。
ステップ1から3を繰り返して、次の値で2つ目のソース演算子を追加します:
- 「データ・アセット」で、「Data_Lake」を選択します。
- 「接続」で、「デフォルトの接続」を選択します。
- 「スキーマ」で、コンパートメント、バケットの順に選択します。このチュートリアルでは、オブジェクト・ストレージがソース・データ・アセットとなるため、ここではバケットを選択します。
- 「データ・エンティティ」に「REVENUE.csv」を選択し、ファイル・タイプに「CSV」を選択します。その他の項目については、デフォルト値を受け入れます。
2番目のソース演算子の選択を完了すると、データ・エンティティの選択を反映して演算子名がREVENUE_CSVになります。「識別子」フィールドで、ソース演算子の名前をREVENUEに変更します。
(オプション)「パラメータの割当て」をクリックして、データ・フローの公開時にソース詳細がコンパイル済コードにバインドされないようにします。

パラメータの詳細を参照してください。
(オプション)「属性」タブでは、データ・エンティティの属性を表示し、それぞれの「アクション」メニュー()から、除外ルールまたは名前変更ルールを属性に適用できます。
「名前」または「タイプ」列のフィルタ・アイコンを使用して、除外する属性に1つ以上のフィルタを適用することもできます。
(オプション)「データ」タブでは、ソース・データ・エンティティからのデータのサンプリングを表示してデータに変換を適用したり、属性を選択してデータ・プロファイルを表示できます。

データ変換についてさらに学習します。
(オプション)「検証」タブでは、ソース演算子の構成で警告またはエラーを確認できます。
データ・フローを保存して編集を続行するには、「保存」をクリックします。

3. データのフィルタリングおよび変換

データのフィルタリング

フィルタ演算子は、条件に基づいてアップストリーム演算子からデータのサブセットを生成します。

「演算子」パネルから、フィルタ演算子をキャンバスにドロップします。
REVENUEをFILTER_1に接続します:
- カーソルを「REVENUE」に配置します。
- 「REVENUE」の横にあるコネクタの円をドラッグします。
- FILTER_1にコネクタの円をドロップします。
FILTER_1を選択します。
「プロパティ」パネルで、「フィルタ条件」の横にある「作成」をクリックします。
「フィルタ条件の作成」パネルで、「名前による検索」フィールドにSTAと入力します。
ORDER_STATUSをダブルクリックまたはドロップして、フィルタ条件エディタに追加します。
条件エディタで、='1-Booked'を入力します。条件は次のとおりです:

FILTER_1.REVENUE_CSV.ORDER_STATUS='1-Booked'

ノート

引用符の問題を回避するため、コピー・アンド・ペーストはしないでください。
「作成」をクリックします。
「演算子」パネルから、フィルタ演算子をキャンバスにドロップし、「CUSTOMERS」の後ろに配置します。
CUSTOMERSをFILTER_2に接続します:
- 「CUSTOMERS」にカーソルを移動します。
- 「CUSTOMERS」の右側にあるコネクタの円をドラッグします。
- FILTER_2にコネクタの円をドロップします。
FILTER_2の「プロパティ」パネルで、「フィルタ条件」の横にある「作成」をクリックします。
「フィルタ条件の作成」パネルで、「名前による検索」フィールドにCOUと入力します。
COUNTRY_CODEをダブルクリックして、条件エディタに追加します。
条件が次のようになるように、='US'と入力します:

FILTER_2.CUSTOMERS_JSON.COUNTRY_CODE='US'
「作成」をクリックします。
データ・フローを保存して編集を続行するには、「保存」をクリックします。

データの変換

データ・エクスプローラを使用すると、データ・サンプルを探索したり、プロファイリング・メタデータを確認したり、「プロパティ」パネルの「データ」タブで変換を適用することができます。変換が適用されるたびに、式演算子がキャンバスに追加されます。

FILTER_2の「プロパティ」パネルで、「データ」タブを選択します。
すべてのデータ行および属性が表示されます。行をスクロールするには垂直スクロールバーを使用し、属性をスクロールするには水平スクロールバーを使用できます。
「パターンで検索」フィールドに、STATE*と入力します。
表内の属性の数がフィルタされます。パターンに一致する属性のみが表示されます。
FILTER_2.CUSTOMERS_JSON.STATE_PROVINCEの変換メニュー()をクリックし、「大/小文字の変更」を選択します。
「大/小文字の変更」ダイアログで、「タイプ」メニューから「大/小文字」を選択します。
「ソース属性の保持」チェック・ボックスを選択しないでください。
「名前」はそのままにします。
「適用」をクリックします。

式演算子がデータ・フローに追加されます。「プロパティ」パネルで、「詳細」タブがフォーカスされ、式演算子の詳細が表示されます。

「式」表で、生成された式UPPER(EXPRESSION_1.CUSTOMERS_JSON.STATE_PROVINCE)を確認できます。
式演算子の「プロパティ」パネルで、「識別子」フィールドの名前をCHANGE_CASEに変更します。
「データ」タブをクリックし、水平スクロールバーを使用して端までスクロールします。

CHANGE_CASE.STATE_PROVINCEがデータセットの最後に追加されます。「データ」タブで、CHANGE_CASE.STATE_PROVINCEの変換済データをプレビューできます。
「演算子」パネルから、式演算子をキャンバスにドロップし、CHANGE_CASEの後ろに配置します。
CHANGE_CASEを新しいEXPRESSION_1演算子に接続します。
EXPRESSION_1の「プロパティ」パネルで、「式」表で「式の追加」を選択します。
「式の追加」パネルで:
1. 「識別子」フィールドで、式の名前をFULLNAMEに変更します。
2. 「データ型」はVARCHARのままにします。
3. 「長さ」を200に設定します。
4. 「式ビルダー」で、「受信」リストを「関数」リストに切り替えます。
5. 「名前による検索」フィールドに、CONと入力します。次に、Stringの下のCONCATを探します。
6. CONCAT(CONCAT(EXPRESSION_1.CUSTOMERS_JSON.FIRST_NAME, ' '),EXPRESSION_1.CUSTOMERS_JSON.LAST_NAME)と入力します
  
  関数のプレースホルダを強調表示し、「受信」リストから属性をダブルクリックまたは削除して、式を作成することもできます。
7. 「追加」をクリックします。
これで、EXPRESSION_1演算子の式表にCONCAT式がリストされます。式は必要な数だけ追加できます。
データ・フローを保存して編集を続行するには、「保存」をクリックします。

4. データの結合

フィルタおよび変換を適用した後は、一意の顧客識別子を使用してソース・データ・エンティティを結合してから、データをターゲット・データ・エンティティにロードできます。

EXPRESSION_1のデータをFILTER_1のデータと結合するには、「演算子」パネルから「結合」演算子をキャンバスにドロップし、EXPRESSION_1およびFILTER_1の後ろに配置します。
EXPRESSION_1をJOIN_1に接続します。次に、FILTER_1をJOIN_1に接続します。
JOIN_1を選択した場合、「プロパティ」パネルの「詳細」タブで、「結合条件」の横にある「作成」をクリックします。
「結合条件の作成」パネルで、「名前による検索」フィールドにCUSTと入力します。
CUST_IDおよびCUST_KEYを使用してエンティティを結合します。
エディタで、JOIN_1_1.CUSTOMERS_JSON.CUST_ID=JOIN_1_2.REVENUE_CSV.CUST_KEYと入力します。
「作成」をクリックします。
データ・フローを保存して編集を続行するには、「保存」をクリックします。

5. ターゲット演算子の追加

「演算子」パネルから、ターゲット演算子をキャンバスにドロップします。
JOIN_1をTARGET_1に接続します。
キャンバスでTARGET_1を選択します(まだ選択されていない場合)。
「プロパティ」パネルの「詳細」タブで、「新規データ・エンティティの作成」チェック・ボックスを選択しないでください。
次に、次の各オプションで「選択」をクリックして、データ・エンティティの選択を行います:
- 「データ・アセット」で、「Data_Warehouse」を選択します。
- 「接続」で、「デフォルトの接続」を選択します。
- 「スキーマ」で、「BETA」を選択します。
- 「データ・エンティティ」で、「CUSTOMERS_TARGET」を選択します。
「ステージングの場所」で、「データ・レイク」データ・アセット、そのデフォルト接続およびコンパートメントを選択します。次に、「スキーマ」で、サンプル・データのインポートに使用したオブジェクト・ストレージ・バケットを選択します。
「統合戦略」の値は「挿入」のままにします。
「属性」マッピングを確認するには、「マップ」タブをクリックします。

デフォルトでは、すべての属性が名前でマップされます。

たとえば、JOIN_1のCUST_IDは、ターゲット・データ・エンティティのCUST_IDにマップされます。
まだマップされていない属性を手動でマップするには、「ターゲット属性」表で「すべて」メニューを選択し、「マップされていない属性」を選択します。

(受信フィールド用の)「ソース属性」表でも同じことができます。
ここで、「ターゲット属性」の「ソース属性」からFULL_NAMEにFULLNAMEを削除します。
「ターゲット属性」表で、「マップされていない属性」を選択し、「すべて」を選択します。

これで、すべての属性がマップされました。
(オプション)「アクション」をクリックして、使用可能な様々なマッピング・オプションを確認します。
(オプション)「ルールの表示」をクリックして、適用されたルールを表示します。
データ・フローを保存してデザイナを終了するには、「保存してクローズ」をクリックします。

その他のリソース

さらに学習するには、次のURLを参照してください。

次の手順

データ・フローを使用してデータを収集および変換した後は、データ・フローを構成および実行するための統合タスクを作成する必要があります。

Oracle Cloud Infrastructureドキュメント