【ITニュース解説】Ingest data with a pipeline in Microsoft Fabric

データの世界では、日々膨大な量の情報が生まれている。システムエンジニアを目指すあなたも、こうしたデータをどのように集め、整理し、活用するかが、現代のITシステム開発において非常に重要なスキルとなることを知っておくべきだろう。今回解説するMicrosoft Fabricを使ったデータ処理は、まさにその核となる部分を扱っている。

Microsoft Fabricとは、マイクロソフトが提供する、クラウド環境で大量のデータを分析するための統合プラットフォームだ。これは「データレイクハウス」と呼ばれる新しいデータ管理の概念を基盤としている。データレイクハウスは、大量の生データをそのまま保管できる「データレイク」の柔軟性と、構造化されたデータを効率的に管理できる「データウェアハウス」の利点を組み合わせたものだ。データエンジニアの主な仕事の一つは、様々な業務システムから発生するデータを、このデータレイクハウスに安全かつ効率的に取り込み、分析可能な形に整えることだ。この「データの取り込み」は、専門用語で「データインジェスト」と呼ばれる。

Microsoft Fabricでは、このデータインジェストを実現するための強力なツールが提供されている。その一つが「パイプライン」だ。パイプラインとは、データが流れる一連の経路と、その経路でデータに対して行われる様々な処理（アクティビティ）を定義したものだ。これにより、データの抽出（Extract）、変換（Transform）、ロード（Load）という、いわゆるETL（またはELT）のプロセスを自動化できる。さらにFabricは、大量データを高速に処理するための「Apache Spark」という技術もサポートしている。パイプラインとSparkを組み合わせることで、複雑なデータ加工ロジックを実装し、外部のデータソースからOneLakeというFabricのストレージにデータをコピーし、Sparkコードでカスタム変換を行った後、分析用のテーブルにロードすることが可能となる。

では、具体的な手順を見ていこう。まず、Fabricで作業を始めるには、「ワークスペース」を作成する必要がある。これは、あなたの作業プロジェクトやデータ、ツールをまとめて管理するための専用の作業環境だ。ブラウザからFabricのホームぺージにアクセスし、ワークスペースを新規作成する。その際、Fabricの機能を利用するためのライセンスモードを選択する。新しいワークスペースは、何もデータが入っていない空の状態から始まる。

次に、データを格納する場所として「レイクハウス」を作成する。レイクハウスは、データの「貯水池」のようなもので、様々な形式のデータを取り込み、管理するための中心的なハブとなる。ワークスペース内で「レイクハウス」を選択し、任意の名前をつけて作成する。作成直後のレイクハウスには、まだテーブルもファイルも存在しない。この中に、今後取り込むデータを整理するためのサブフォルダ（例えばnew_data）を作成しておくと良いだろう。

いよいよデータの取り込み、つまりパイプラインの作成だ。最も簡単なデータインジェストの方法は、パイプライン内の「Copy Data（データコピー）」アクティビティを使って、データソースからレイクハウスへデータをコピーすることである。レイクハウスのホームぺージから「新しいデータパイプライン」を選択し、例えば「Ingest Sales Data」という名前で作成する。パイプラインエディタが開いたら、「Copy Data」ウィザードを使って設定を進める。 データソースの選択では、今回はHTTPプロトコルを使ってWeb上にあるCSVファイルを取得する。具体的には、Microsoftが学習用として公開しているURL（https://raw.githubusercontent.com/MicrosoftLearning/dp-data/main/sales.csv）を指定する。接続情報を設定し、認証方法を「Anonymous（匿名）」とする。次に、データの読み込み方法に関する詳細設定を行う。例えば、相対URLは空白、リクエストメソッドはGETとする。 続いて、取得したデータの形式について指定する。今回はCSVファイルなので、「DelimitedText（区切り文字テキスト）」を選択し、列の区切り文字を「カンマ（,）」、行の区切り文字を「改行（\n）」、そしてファイルの1行目をヘッダーとして扱うように設定する。データプレビューで内容を確認後、データの保存先を設定する。レイクハウス内の「Files」フォルダ配下に、先に作成しておいたnew_dataフォルダにsales.csvという名前で保存するように指定する。これらの設定が完了したら、パイプラインを保存して実行する。パイプラインの実行状況は、出力ペインで監視でき、ステータスが「成功」になるまで待つ。レイクハウスに戻り、「Files」を展開してnew_dataフォルダにsales.csvがコピーされていることを確認できるだろう。

データがレイクハウスに取り込まれたので、次はそれを加工・整形するステップだ。ここで「ノートブック」が登場する。ノートブックとは、コードと説明文を組み合わせた対話型のドキュメントで、データ分析や加工によく使われる。レイクハウスのホームぺージから「新しいノートブック」を開く。ノートブックには「セル」と呼ばれるコードやテキストのブロックが含まれる。既存のセルに、後でパイプラインから値を渡せるようにするためのパラメータとしてtable_name = "sales"という変数を宣言し、これを「パラメータセル」として設定する。 その下に新しいコードセルを追加し、PySparkというPythonベースのSpark用言語でデータ加工のコードを記述する。このコードは、まず先にコピーしたsales.csvファイルを読み込む。次に、OrderDate列から「年」と「月」の情報を抽出し、それぞれYearとMonthという新しい列として追加する。また、CustomerName列から空白で区切って「名」と「姓」を抽出し、FirstNameとLastName列を作成する。最後に、必要な列だけを選択し、順序を整理して、加工されたデータをsalesという名前のテーブルとしてレイクハウスに保存する。この時、もしsalesテーブルが既に存在すれば、新しいデータを追記（append）する設定になっている。ノートブックのすべてのセルを実行すると、Sparkプールが起動し、コードが実行される。完了後、レイクハウスの「Tables」セクションを更新すると、新しくsalesテーブルが作成されていることを確認できる。ノートブックの名前も、後から識別しやすいように「Load Sales」に変更しておこう。

ここまでの作業で、データのコピーと加工・テーブルへのロードという二つのステップができた。次は、これらを一つにまとめて、より堅牢で再利用可能な自動処理フローにするために、最初に作成したパイプラインを修正する。 既存の「Ingest Sales Data」パイプラインを開き、そこに新しいアクティビティを追加する。まず「Delete data（データ削除）」アクティビティを「Copy data」アクティビティの左に配置し、接続する。この「Delete data」アクティビティは、Copy dataの前に実行され、レイクハウスのFiles/new_dataフォルダにある既存の.csvファイルをすべて削除する役割を担う。これは、常に最新のデータだけを取り込みたい場合に、前回の古いデータが残らないようにするための重要な前処理だ。 次に、「Copy data」アクティビティの右に「Notebook（ノートブック）」アクティビティを追加し、接続する。この「Notebook」アクティビティは、先に作成した「Load Sales」ノートブックを実行するように設定する。さらに重要な点として、ノートブックに渡すパラメータを設定する。ここではtable_nameというパラメータにnew_salesという値を設定する。これにより、ノートブック内で定義したtable_name = "sales"というデフォルト値が上書きされ、結果としてデータはnew_salesという新しいテーブルにロードされることになる。 これらの変更をパイプラインに保存し、再度実行する。パイプライン全体が順に実行され、「Delete data」が古いファイルを削除し、「Copy data」が新しいsales.csvをコピーし、最後に「Notebook」がそのsales.csvを読み込んで加工し、new_salesテーブルにデータを保存する。パイプラインの実行完了後、レイクハウスに戻り、「Tables」セクションを展開すると、新しくnew_salesテーブルが作成され、データが格納されていることを確認できるだろう。

この一連の作業を通じて、Microsoft Fabricを使い、外部データソースからデータを取り込み（インジェスト）、それをSparkノートブックで必要な形に加工・変換し、最終的に構造化されたテーブルとしてレイクハウスにロードするデータパイプラインを設計・実装する方法を学んだことになる。このプロセスは、まさにデータエンジニアリングの基本であり、データ活用の基盤を築く上で不可欠なスキルだ。システムエンジニアとして、このようなデータ処理の自動化と効率化の知識は、あなたのキャリアにおいて大きな武器となるだろう。