クイックスタート: マッピングデータフローを使用してデータを変換する

[アーティクル]
05/23/2023

このクイックスタートでは、Azure Synapse Analytics を使い、マッピングデータフローを使用して Azure Data Lake Storage Gen2 (ADLS Gen2) ソースから ADLS Gen2 シンクにデータを変換するパイプラインを作成します。このクイックスタートの構成パターンは、マッピングデータフローを使用してデータを変換するときに拡張することができます。

このクイックスタートでは、次の手順を実行します。

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

前提条件

Azure サブスクリプション:Azure サブスクリプションをお持ちでない場合は、開始する前に無料の Azure アカウントを作成してください。
Azure Synapse ワークスペース:「クイックスタート:Synapse ワークスペースを作成する」の手順に従い、Azure portal を使用して Synapse ワークスペースを作成します。
Azure ストレージアカウント:ADLS ストレージを、"ソース" と "シンク" のデータストアとして使用します。ストレージアカウントがない場合の作成手順については、Azure のストレージアカウントの作成に関するページを参照してください。

このチュートリアルで変換するファイルは MoviesDB です。このファイルは、こちらにあります。 GitHub からファイルを取得するには、コンテンツを任意のテキストエディターにコピーして、.csv ファイルとしてローカルに保存します。ファイルをご自分のストレージアカウントにアップロードするには、Azure portal を使用した BLOB のアップロードに関するページを参照してください。例では、'sample-data' という名前のコンテナーを参照しています。

Synapse Studio への移動

Azure Synapse ワークスペースが作成された後、Synapse Studio を開く方法は 2 つあります。

Azure portal で Synapse ワークスペースを開きます。 [はじめに] の下の [Synapse Studio を開く] カードで、[開く] を選択します。
Azure Synapse Analytics を開き、ワークスペースにサインインします。

このクイックスタートでは、例として "adftest2020" という名前のワークスペースを使用します。自動的に Synapse Studio のホームページに移動します。

Synapse Studio のホームページ

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

パイプラインには、一連のアクティビティを実行するための論理フローが含まれています。このセクションでは、Data Flow アクティビティが含まれるパイプラインを作成します。

[統合] タブに移動します。パイプラインヘッダーの横にあるプラス符号のアイコンを選択し、[パイプライン] を選択します。
パイプラインの [プロパティ] 設定ページで、名前として「TransformMovies」と入力します。
[アクティビティ] ペインの [Move and Transform](移動と変換) で、 [データフロー] をパイプラインキャンバス上にドラッグします。
[Adding data flow](データフローの追加) ページのポップアップで、[Create new data flow](新しいデータフローの作成) ->[データフロー] を選択します。完了したら、 [OK] をクリックします。
[プロパティ] ページで、データフローに TransformMovies という名前を付けます。

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

Data Flow を作成すると、データフローキャンバスが自動的に表示されます。この手順では、ADLS ストレージ内の MoviesDB.csv を取得し、1910 年から 2000 年までのコメディの平均評価を集計するデータフローを作成します。次に、このファイルを ADLS ストレージに書き戻します。

データフローキャンバスの上にある [Data flow debug](データフローのデバッグ) スライダーをオンにスライドします。デバッグモードを使用すると、ライブ Spark クラスターに対する変換ロジックの対話型テストが可能になります。 Data Flow クラスターのウォームアップには 5 から 7 分かかるため、ユーザーが Data Flow の開発を計画している場合は、最初にデバッグを有効にすることをお勧めします。詳細については、デバッグモードに関するページを参照してください。
データフローキャンバスで [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] をクリックします。
デバッグクラスターが起動している場合は、ソース変換の [Data Preview](データのプレビュー) タブに移動し、 [更新] をクリックして、データのスナップショットを取得します。データプレビューを使用すると、変換が正しく構成されていることを確認できます。
データフローキャンバスでソースノードの横にあるプラスアイコンをクリックして、新しい変換を追加します。最初に追加する変換は、フィルターです。
フィルター変換に 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] を選択し、映画の公開年ごとに集計をグループ化します。
[集計] タブに移動します。左側のテキストボックスで、集計列に AverageComedyRating という名前を付けます。式ビルダーを使用して集計式を入力するには、右側の式ボックスをクリックします。
列 [Rating] の平均値を取得するには、avg() 集計関数を使用します。 Rating は文字列で、avg() で受け入れられるのは数値入力なので、toInteger() 関数を使用して値を数値に変換する必要があります。式は次のようになります。

avg(toInteger(Rating))

完了したら、 [Save and Finish](保存して終了する) をクリックします。
変換出力を表示するには、 [Data Preview](データのプレビュー) タブに移動します。 year と AverageComedyRating の 2 つの列だけがあることに注目してください。
次に、 [Destination](変換先) の下で [シンク] 変換を追加します。
シンクに Sink という名前を付けます。 [新規] をクリックして、シンクデータセットを作成します。
[Azure Data Lake Storage Gen2] を選択します。 [続行] をクリックして続行します。
[DelimitedText] を選択します。 [続行] をクリックして続行します。
シンクデータセットに MoviesSink という名前を付けます。リンクされたサービスの場合、手順 7 で作成した ADLS Gen2 のリンクされたサービスを選択します。データの書き込み先となる出力フォルダーを入力します。このクイックスタートでは、コンテナー "sample-data" 内のフォルダー "output" に書き込んでいます。フォルダーは、事前に存在している必要はなく、動的に作成することができます。 [First row as header](最初の行をヘッダーにする) をオンに設定し、 [スキーマのインポート] で [なし] を選択します。完了したら、 [OK] をクリックします。

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

Data Flow を実行して監視する

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

パイプラインキャンバスに移動します。 [デバッグ] をクリックして、デバッグ実行をトリガーします。
Data Flow アクティビティのパイプラインデバッグでは、アクティブなデバッグクラスターが使用されますが、それでも初期化には少なくとも 1 分かかります。進行状況は [出力] タブで追跡することができます。実行が正常に完了したら、眼鏡のアイコンをクリックして [監視] ウィンドウを開きます。
[監視] ウィンドウには、各変換手順で使用した行数と所要時間が表示されます。
変換をクリックすると、データの列とパーティション分割に関する詳細情報が表示されます。