チュートリアル: マッピングデータフローを使用したデータの変換

2025-06-23

適用対象: Azure Data Factory Azure Synapse Analytics

ヒント

企業向けのオールインワン分析ソリューション、Microsoft Fabric の Data Factory をお試しください。 Microsoft Fabric は、データ移動からデータサイエンス、リアルタイム分析、ビジネスインテリジェンス、レポートまで、あらゆるものをカバーしています。無料で新しい試用版を開始する方法について説明します。

このチュートリアルでは、Azure Data Factory ユーザーインターフェイス (UX) を使用して、マッピングデータフローを使用して Azure Data Lake Storage (ADLS) Gen2 ソースから ADLS Gen2 シンクにデータをコピーおよび変換するパイプラインを作成します。このチュートリアルの構成パターンは、マッピングデータフローを使用してデータを変換するときに拡張することができます

このチュートリアルでは、一般にデータフローをマップすることを目的としています。データフローは、Azure Data Factory および Synapse パイプラインの両方で使用できます。 Azure Synapse Pipelines のデータフローを初めて使用する場合は、 Azure Synapse パイプラインを使用したデータフローに従ってください。

このチュートリアルでは、次の手順を実行します。

データファクトリを作成します。
Data Flow アクティビティを含むパイプラインを作成します。
4 つの変換を使用して、マッピングデータフローを構築します。
パイプラインをテスト実行します。
Data Flow アクティビティを監視します。

前提条件

Azure サブスクリプション。 Azure サブスクリプションをお持ちでない場合は、開始する前に無料の Azure アカウントを作成してください。
Azure Data Lake Storage Gen2 アカウント。 ADLS ストレージを、ソースとシンクのデータストアとして使用します。ストレージアカウントがない場合の作成手順については、Azure のストレージアカウントの作成に関するページを参照してください。
ここから MoviesDB.csv をダウンロードしてください。 GitHub からファイルを取得するには、コンテンツを任意のテキストエディターにコピーして、.csv ファイルとしてローカルに保存します。ファイルをストレージアカウントの 'sample-data' という名前のコンテナーにアップロードします。

Data Factory の作成

この手順では、データファクトリを作成し、Data Factory UX を開いて、データファクトリにパイプラインを作成します。

Microsoft Edge または Google Chrome を開きます。現在、Data Factory の UI がサポートされる Web ブラウザーは Microsoft Edge と Google Chrome だけです。
上部のメニューで、 リソースの作成>Analytics>Data Factory を選択します。
[新しいデータファクトリ] ページで、 [名前] に「ADFTutorialDataFactory」と入力します。

Azure データファクトリの名前は グローバルに一意にする必要があります。データファクトリの名前の値に関するエラーメッセージが表示された場合は、別の名前を入力してください。 (yournameADFTutorialDataFactory など)。 Data Factory アーティファクトの名前付け規則については、Data Factory の名前付け規則に関するページを参照してください。
データファクトリを作成する Azure サブスクリプションを選択します。
[リソースグループ] で、次の手順のいずれかを行います。
1. [Use existing (既存のものを使用)] を選択し、ドロップダウンリストから既存のリソースグループを選択します。
2. [新規作成] を選択し、リソースグループの名前を入力します。
リソースグループの詳細については、リソースグループを使用した Azure のリソースの管理に関するページを参照してください。
[バージョン] で、 [V2] を選択します。
[リージョン] で、データファクトリの場所を選択します。サポートされている場所のみがドロップダウンリストに表示されます。データファクトリによって使用されるデータストア (Azure Storage、SQL Database など) やコンピューティング (Azure HDInsight など) は、他のリージョンに存在していてもかまいません。
[確認および作成] を選択し、次に [作成] を選択します。
作成が完了すると、その旨が通知センターに表示されます。 [リソースに移動] を選択して、Data factory ページに移動します。
[ スタジオの起動 ] を選択して、別のタブで Data Factory スタジオを起動します。

Data Flow アクティビティを含むパイプラインの作成

この手順では、Data Flow アクティビティを含むパイプラインを作成します。

Azure Data Factory のホームページで、 [Orchestrate](調整) を選択します。
これで、新しいパイプラインのウィンドウが開いています。パイプラインのプロパティの [全般] タブで、パイプラインの名前に「TransformMovies」と入力します。
[アクティビティ] ウィンドウで、 [移動と変換] アコーディオンを展開します。ウィンドウから Data Flow アクティビティをパイプラインキャンバスにドラッグアンドドロップします。
データフローアクティビティに DataFlow1 という名前を付けます。
パイプラインキャンバスの上部のバーで、 [Data Flow のデバッグ] スライダーをオンにスライドします。デバッグモードを使用すると、ライブ Spark クラスターに対する変換ロジックの対話型テストが可能になります。 Data Flow クラスターのウォームアップには 5 から 7 分かかるため、ユーザーが Data Flow の開発を計画している場合は、最初にデバッグを有効にすることをお勧めします。詳細については、デバッグモードに関するページを参照してください。

データフローキャンバスでの変換ロジックの作成

この手順では、ADLS ストレージの moviesDB.csv を取得し、1910 年から 2000 年までのコメディの平均評価を集計するデータフローを構築します。その後、このファイルを ADLS ストレージに書き戻します。

キャンバスの下のパネルで、データフローアクティビティの [設定 ] に移動し、データフローフィールドの横にある [新規] を選択します。これにより、データフローキャンバスが開きます。
[プロパティ] ペインの [全般] で、データフローに TransformMovies という名前を付けます。
データフローキャンバスで [Add Source](ソースの追加) ボックスを選択して、ソースを追加します。
ソースに MoviesDB という名前を付けます。 [新規] を選択して、新しいソースデータセットを作成します。
[Azure Data Lake Storage Gen2] を選択します。 [続行] を選択します。
[DelimitedText] を選択します。 [続行] を選択します。
データセットに MoviesDB という名前を付けます。リンクされたサービスのドロップダウンで、 [新規] を選択します。
リンクされたサービスの作成画面で、ADLS gen2 のリンクされたサービスに ADLSGen2 という名前を付けて、使用する認証方法を指定します。次に、接続の資格情報を入力します。このチュートリアルでは、アカウントキーを使用してストレージアカウントに接続しています。 [テスト接続] を選択すると、資格情報が正しく入力されたことを確認できます。完了したら、[作成] を選択します。
データセットの作成画面に戻ったら、 [ファイルパス] フィールドの下でファイルが配置されている場所を入力します。このチュートリアルでは、moviesDB.csv ファイルはコンテナー sample-data に配置されています。ファイルにはヘッダーが含まれているため、 [First row as header](最初の行をヘッダーにする) をオンにします。ストレージ内のファイルからヘッダースキーマを直接インポートするには、 [From connection/store](接続/ストアから) を選択します。完了したら OK を選択します。
デバッグクラスターが起動している場合は、ソース変換の [データのプレビュー] タブに移動し、 [更新] を選択して、データのスナップショットを取得します。データプレビューを使用すると、変換が正しく構成されていることを確認できます。
データフローキャンバス上のソースノードの横にあるプラスアイコンを選択して、新しい変換を追加します。最初に追加する変換は、フィルターです。
フィルター変換に FilterYears という名前を付けます。 [フィルター] の横にある式ボックスを選択し、[式ビルダーを開く] を選択します。ここでは、フィルタリング条件を指定します。
データフローの式ビルダーでは、さまざまな変換で使用する式を対話形式で作成できます。式には、組み込み関数、入力スキーマの列、ユーザー定義のパラメーターを含めることができます。式の作成方法の詳細については、Data Flow の式ビルダーに関するページを参照してください。

このチュートリアルでは、1910 年から 2000 年の間に公開された、ジャンルがコメディの映画をフィルター処理します。現在、年は文字列になっているため、toInteger() 関数を使用して整数に変換する必要があります。以上演算子 (>=) と以下演算子 (<=) を使用して、年のリテラル値 1910 と 2000 に対する比較を行います。これらの式を and (&&) 演算子を使用して結合します。式は次のようになります。

toInteger(year) >= 1910 && toInteger(year) <= 2000

コメディ映画を見つけるには、rlike() 関数を使用して、ジャンル列でパターン 'Comedy' を検索します。 rlike 式を年の比較と結合すると、次の式が得られます。

toInteger(year) >= 1910 && toInteger(year) <= 2000 && rlike(genres, 'Comedy')

デバッグクラスターがアクティブな場合は、 [更新] を選択してロジックを確認し、使用された入力と比較した式の出力を確認できます。データフローの式言語を使用してこのロジックを実現する方法に対する正解は複数あります。

式の操作が完了したら、[Save and Finish]$保存して終了する$ を選択します。
フィルターが正しく機能していることを確認するには、データプレビューをフェッチします。
次に追加する変換は、 [Schema modifier](スキーマ修飾子) の下にある [集計] 変換です。
集計変換に AggregateComedyRatings という名前を付けます。 [グループ化] タブで、ドロップダウンから [year] を選択し、映画の公開年ごとに集計をグループ化します。

$[Aggregate Settings]$集計の設定$ の下にある [グループ化] タブの年オプションを示すスクリーンショット。$
[集計] タブに移動します。左側のテキストボックスで、集計列に AverageComedyRating という名前を付けます。式ビルダーを使用して集計式を入力するには、右側の式ボックスを選択します。

$[Aggregate Settings]$集計の設定$ の下にある [集計] タブの年オプションを示すスクリーンショット。$
列 [Rating] の平均値を取得するには、avg() 集計関数を使用します。 Rating は文字列で、avg() で受け入れられるのは数値入力なので、toInteger() 関数を使用して値を数値に変換する必要があります。式は次のようになります。

avg(toInteger(Rating))

完了したら、[Save and Finish]$保存して終了する$ を選択します。
変換出力を表示するには、 [Data Preview](データのプレビュー) タブに移動します。 year と AverageComedyRating の 2 つの列だけがあることに注目してください。
次に、 [Destination](変換先) の下で [シンク] 変換を追加します。

$[Destination]$変換先$ でシンク変換を追加する場所を示すスクリーンショット。$
シンクに Sink という名前を付けます。 [新規] を選択して、シンクデータセットを作成します。
[Azure Data Lake Storage Gen2] を選択します。 [続行] を選択します。
[DelimitedText] を選択します。 [続行] を選択します。
シンクデータセットに MoviesSink という名前を付けます。リンクされたサービスの場合、手順 6 で作成した ADLS gen2 のリンクされたサービスを選択します。データの書き込み先となる出力フォルダーを入力します。このチュートリアルでは、コンテナー 'sample-data' 内のフォルダー ' output ' に書き込んでいます。フォルダーは、事前に存在している必要はなく、動的に作成することができます。 [First row as header](最初の行をヘッダーにする) をオンに設定し、 [スキーマのインポート] で [なし] を選択します。完了を選択します。

これで、データフローの構築が完了しました。これをパイプラインで実行する準備ができました。

Data Flow を実行して監視する

パイプラインを発行する前にデバッグすることができます。この手順では、データフローパイプラインのデバッグ実行をトリガーします。データのプレビューではデータが書き込まれませんが、デバッグ実行によってシンクの変換先にデータが書き込まれます。

パイプラインキャンバスに移動します。 [デバッグ] を選択して、デバッグ実行をトリガーします。
Data Flow アクティビティのパイプラインデバッグでは、アクティブなデバッグクラスターが使用されますが、それでも初期化には少なくとも 1 分かかります。進行状況は [出力 ] タブで追跡できます。実行が成功したら、実行にカーソルを合わせ、眼鏡アイコンを選択して監視ウィンドウを開きます。
監視ウィンドウで、 [ステージ] ボタンを選択して、各変換ステップで費やされた行数と時間を確認します。
変換を選択すると、データの列とパーティション分割に関する詳細情報が表示されます。