GSP807

概要

ETL は、Extract（抽出）、Transform（変換）、Load（読み込み）の略語です。このコンセプトは、EL、ELT、ELTL などと表現される場合もあります。

このラボでは、Cloud Data Fusion の Pipeline Studio を使用して ETL パイプラインを構築する方法を学びます。Pipeline Studio に表示される構成要素と組み込みのプラグインを使用して、バッチ パイプラインをノード単位で構築します。また、Wrangler プラグインを使用して変換を構築し、パイプラインを通過するデータに適用します。

ETL アプリケーションでよく使用される一般的なデータソースは、CSV 形式のテキスト ファイルに保存されたデータです。これは、多くのデータベース システムにおいてこの形式でデータのエクスポートやインポートが行われるためです。このラボでは CSV ファイルを使用しますが、データベース ソースやその他の任意のデータソースに対しても、ここで使用する手法を適用できます。

出力は BigQuery テーブルに書き込まれます。また、このターゲット データセットを対象としたデータ分析には標準 SQL を使用します。

目標

このラボでは、次の方法について学びます。

• Cloud Data Fusion の Pipeline Studio でバッチ パイプラインを作成する。
• Wrangler を使用してデータをインタラクティブに変換する。
• 出力を BigQuery に書き込む。

設定と要件

各ラボでは、新しい Google Cloud プロジェクトとリソースセットを一定時間無料で利用できます。

1. シークレット ウィンドウを使用して Google Cloud Skills Boost にログインします。

2. ラボのアクセス時間（例: 02:00:00）に注意し、時間内に完了できるようにしてください。
   一時停止機能はありません。必要な場合はやり直せますが、最初からになります。

3. 準備ができたら、[ラボを開始] をクリックします。

   注: [ラボを開始] をクリックしてから、ラボが必要なリソースをプロビジョニングして Data Fusion インスタンスを作成するまで 15～20 分ほどかかります。 その間、ラボの目標を理解するために以下のステップをご確認ください。
   左側のパネルにラボの認証情報（ユーザー名とパスワード）が表示されたら、インスタンスの作成が完了したため、コンソールへのログインに進めるようになります。

4. ラボの認証情報（ユーザー名とパスワード）をメモしておきます。この情報は、Google Cloud コンソールにログインする際に使用します。

5. [Google Console を開く] をクリックします。

6. [別のアカウントを使用] をクリックし、このラボの認証情報をコピーしてプロンプトに貼り付けます。
   他の認証情報を使用すると、エラーが発生したり、料金の請求が発生したりします。

7. 利用規約に同意し、再設定用のリソースページをスキップします。

注: ラボを完了した場合と最初からやり直す場合以外は、[ラボを終了] をクリックしないでください。クリックすると、作業内容とプロジェクトが削除されます。

Google Cloud コンソールにログインする

1. このラボ セッションで使用しているブラウザタブまたはウィンドウで、[接続の詳細] パネルからユーザー名をコピーし、[Google Console を開く] ボタンをクリックします。

注: アカウントの選択を求められたら、[別のアカウントを使用] をクリックします。

2. ユーザー名を貼り付け、プロンプトが表示されたらパスワードを入力します。
3. [次へ] をクリックします。
4. 利用規約に同意します。

これは、このラボの間だけ有効な一時的なアカウントです。以下の点に注意してください。

• 復元オプションを追加しないでください。
• 無料トライアルに登録しないでください。

5. コンソールが開いたら、左上のナビゲーション メニュー（）をクリックしてサービスのリストを確認します。

Cloud Shell をアクティブにする

Cloud Shell は、開発ツールが組み込まれた仮想マシンです。5 GB の永続ホーム ディレクトリを提供し、Google Cloud 上で実行されます。Cloud Shell を使用すると、コマンドラインで Google Cloud リソースにアクセスできます。gcloud は Google Cloud のコマンドライン ツールで、Cloud Shell にプリインストールされており、Tab キーによる入力補完がサポートされています。

1. Google Cloud Console のナビゲーション パネルで、「Cloud Shell をアクティブにする」アイコン（）をクリックします。

2. [次へ] をクリックします。
   環境がプロビジョニングされ、接続されるまでしばらく待ちます。接続の際に認証も行われ、プロジェクトは現在のプロジェクト ID に設定されます。次に例を示します。

サンプル コマンド

• 有効なアカウント名前を一覧表示する:

gcloud auth list

（出力）

Credentialed accounts: - <myaccount>@<mydomain>.com (active)

（出力例）

Credentialed accounts: - google1623327_student@qwiklabs.net

• プロジェクト ID を一覧表示する:

gcloud config list project

（出力）

[core] project = <プロジェクト ID>

（出力例）

[core] project = qwiklabs-gcp-44776a13dea667a6

プロジェクトの権限を確認する

Google Cloud で作業を開始する前に、Identity and Access Management（IAM）内で適切な権限がプロジェクトに付与されていることを確認する必要があります。

1. Google Cloud コンソールのナビゲーション メニュー（）で、[IAM と管理] > [IAM] をクリックします。

2. Compute Engine のデフォルトのサービス アカウント {project-number}-compute@developer.gserviceaccount.com が存在し、編集者のロールが割り当てられていることを確認します。アカウントの接頭辞はプロジェクト番号で、ナビゲーション メニュー > [Cloud の概要] から確認できます。

アカウントが IAM に存在しない場合やアカウントに編集者のロールがない場合は、以下の手順に沿って必要なロールを割り当てます。

1. Google Cloud コンソールのナビゲーション メニューで、[Cloud の概要] をクリックします。

2. [プロジェクト情報] カードからプロジェクト番号をコピーします。

3. ナビゲーション メニューで、[IAM と管理] > [IAM] をクリックします。

4. IAM ページの上部にある [追加] をクリックします。

5. 新しいプリンシパルの場合は、次のように入力します。

{project-number}-compute@developer.gserviceaccount.com

{project-number} はプロジェクト番号に置き換えてください。

6. [ロールを選択] で、[基本]（または [Project]）> [編集者] を選択します。

7. [保存] をクリックします。

タスク 1. データを読み込む

次に、プロジェクトに GCS バケットを作成し、CSV ファイルをステージングします。Cloud Data Fusion はこの Storage バケットからデータを読み取ることになります。

1. Cloud Shell コンソールで次のコマンドを実行します。これにより、新しいバケットが作成され、関連するデータがそこにコピーされます。

export BUCKET=$GOOGLE_CLOUD_PROJECT gcloud storage buckets create gs://$BUCKET

作成されるバケットの名前は、現在のプロジェクト ID です。

2. 次のコマンドを実行して、データファイル（CSV ファイルと XML ファイル）をバケットにコピーします。

gsutil cp gs://cloud-training/OCBL163/titanic.csv gs://$BUCKET

[進行状況を確認] をクリックして、目標に沿って進んでいることを確認します。 データを読み込む

タスク 2. Cloud Data Fusion インスタンスに必要な権限を追加する

次に、以下の手順に沿って、インスタンスに関連付けられているサービス アカウントに権限を付与します。

1. Cloud コンソールのナビゲーション メニューで、[すべてのプロダクトを表示] をクリックし、[Data Fusion] > [インスタンス] を選択します。Cloud Data Fusion インスタンスがすでにセットアップされて使用できる状態になっているはずです。

注: インスタンスの作成には 10 分ほどかかります。完了するまでお待ちください。

2. Google Cloud コンソールで、[IAM と管理] > [IAM] に移動します。

3. Compute Engine のデフォルトのサービス アカウント {プロジェクト番号}-compute@developer.gserviceaccount.com が表示されていることを確認し、サービス アカウントをクリップボードにコピーします。

4. [IAM 権限] ページで、[+アクセス権を付与] をクリックします。

5. [新しいプリンシパル] フィールドに、サービス アカウントを貼り付けます。

6. [ロールを選択] フィールドをクリックし、「Cloud Data Fusion API サービス エージェント」と入力します。最初の数文字を入力すると [Cloud Data Fusion API サービス エージェント] が表示されるので、それを選択します。

7. [別のロールを追加] をクリックします。

8. [Dataproc 管理者] ロールを追加します。

9. [保存] をクリックします。

[進行状況を確認] をクリックして、目標に沿って進んでいることを確認します。 Cloud Data Fusion API サービス エージェントのロールをサービス アカウントに追加する

サービス アカウントにユーザー権限を付与する

1. コンソールのナビゲーション メニューで、[IAM と管理] > [IAM] をクリックします。

2. [Google 提供のロール付与を含める] チェックボックスをオンにします。

3. リストを下にスクロールして、Google が管理する service-{project-number}@gcp-sa-datafusion.iam.gserviceaccount.com という表示形式の Cloud Data Fusion サービス アカウントを探し、サービス アカウント名をクリップボードにコピーします。

4. 次に、[IAM と管理] > [サービス アカウント] に移動します。

5. {project-number}-compute@developer.gserviceaccount.com という表示形式のデフォルトの Compute Engine アカウントをクリックし、上部のナビゲーション メニューの [アクセス権を持つプリンシパル] タブを選択します。

6. [アクセスを許可] ボタンをクリックします。

7. [新しいプリンシパル] フィールドに、前の手順でコピーしたサービスアカウントを貼り付けます。

8. [ロール] プルダウン メニューで、[サービス アカウント ユーザー] を選択します。

9. [保存] をクリックします。

タスク 3. バッチ パイプラインを構築する

データを処理する際、元データがどのようなものであるかを確認できると、その元データを変換の出発点として使用できるので便利です。そのために、Data Fusion の Wrangler コンポーネントを使用してデータの準備とクリーニングを行います。このデータファーストのアプローチでは、変換を簡単に可視化できます。また、リアルタイムのフィードバックを得られるため、正しい方向に進んでいることを確認できます。

1. コンソールで再びナビゲーション メニュー > [Data Fusion] > [インスタンス] に移動し、Data Fusion インスタンスの横にある [インスタンスを表示] リンクをクリックします。ラボの認証情報を選択してログインします。サービスのガイドに進むダイアログが表示された場合は [No, Thanks] をクリックします。これで Cloud Data Fusion UI が表示されるようになります。

2. Cloud Data Fusion ウェブ UI には固有のナビゲーション パネル（左側）があり、そこから必要なページに移動できます。Cloud Data Fusion UI で、左上のナビゲーション メニューをクリックしてナビゲーション パネルを表示し、[Wrangler] を選択します。

3. 左側のパネルに、事前構成されたデータ接続（Cloud Storage 接続など）が表示されます。[GCS]（Google Cloud Storage）で、[Cloud Storage Default] を選択します。

4. プロジェクト名に対応するバケットをクリックします。

5. titanic.csv をクリックします。

   

6. [Parsing Options] ダイアログ ボックスが表示されたら、[Format] プルダウン リストの [text] を選択します。

7. [CONFIRM] をクリックします。

データが Wrangler の画面に読み込まれます。これで、データ変換を繰り返し適用できるようになりました。

8. 最初の操作として、CSV の元データを解析し、行と列に分かれた表形式に変換します。これを行うには、最初の列見出し（body）にあるプルダウン アイコンを選択し、[Parse] メニュー項目のサブメニューから [CSV] を選択します。

   

9. 元データでは最初の行が列見出しになっているので、この行をヘッダーとして設定する必要があります。そこで、表示された [Parse as CSV] ダイアログ ボックスで [Set first row as header] オプションを選択します。

10. 続けて [Apply] をクリックします。

注: [Set first row as header] チェックボックスの横に表示される非推奨の警告は無視してかまいません。

11. この段階で元データが解析され、この操作によって生成された列が表示されます（[body] 列の右側の列）。右端には、すべての列名がリスト表示されます。

12. 次に、クリーンアップを行います。元の CSV データを表す [body] 列は不要になったので、メニュー オプションを使用して削除します。最初の列見出し [body] のプルダウン アイコンを選択し、[Delete column] メニュー項目を選択します。

注: 変換を適用するときに、コマンドライン インターフェース（CLI）を使用することもできます。CLI は画面下部にある黒いバーです（その中の緑色の $ はプロンプトです）。コマンドの入力を開始すると自動入力機能が起動して、一致するオプションが表示されます。たとえば、[body] 列を削除するために、ディレクティブ drop: body を使用することもできます。

13. Wrangler UI の右端にある [Transformation steps] タブをクリックします。これまでに適用した 2 つの変換が表示されます。

注: 画面右側の [Transformation steps] タブに表示されるディレクティブは、メニュー項目を選択した場合も、CLI を使用した場合も作成されます。ディレクティブは個別の変換であり、複数が集まって 1 つの「レシピ」と呼ばれます。

このラボでは、2 つの変換（レシピ）で問題なく ETL パイプラインを作成できます。次のステップでは、このレシピをパイプライン構築ステップに組み込みます。このレシピは ETL の「T」の役割を果たします。

14. [Create a Pipeline] ボタンをクリックして次のセクションに進み、パイプラインを作成します。ここで、ETL パイプラインがどのように構成されるかを確認できます。

    

15. 次に表示されるダイアログで、[Batch pipeline] を選択して続行します。

注: バッチ パイプラインは、インタラクティブに実行することも、5 分に 1 回や 1 年に 1 回など、任意の頻度で実行するようにスケジュール設定することもできます。

タスク 4. Pipeline Studio を利用する

パイプライン構築の残りのタスクは Pipeline Studio で行います。Pipeline Studio は、データ パイプラインを視覚的に構成できる UI です。このツールには ETL パイプラインの主要な構成要素が表示されます。

この時点で、パイプラインに 2 つのノードが表示されています。1 つは Google Cloud Storage から CSV ファイルを読み取る GCS File プラグイン、もう 1 つは複数の変換から構成されるレシピが含まれる Wrangler プラグインです。

注: パイプラインのノードは、順番に接続されて有向非巡回グラフを生成するオブジェクトです。たとえば、ソース、シンク、変換、アクションなどがあります。

これらの 2 つのプラグイン（ノード）は、ETL パイプラインの「E」と「T」を表しています。このパイプラインを完成させるには、ETL の「L」となる BigQuery シンクを追加します。

1. BigQuery シンクをパイプラインに追加するには、左側のパネルの [Sink] セクションに移動し、[BigQuery] アイコンをクリックしてキャンバス上に配置します。

2. BigQuery シンクがキャンバスに配置されたら、Wrangler ノードと BigQuery ノードを接続します。そうするには、以下の図に示すように Wrangler ノードの矢印を BigQuery ノードまでドラッグして接続します。あとは、必要なデータセットにデータを書き込めるように、いくつかの構成オプションを指定するだけです。

   

タスク 5. パイプラインを構成する

次に、パイプラインを構成します。これを行うには、各ノードのプロパティを開いて設定を確認し、必要に応じて変更を加えます。

1. GCS ノードにカーソルを合わせると、[Properties] ボタンが表示されます。このボタンをクリックして、構成設定を開きます。

各プラグインには、必ず存在しなければならない必須フィールドがいくつかあり、該当するフィールドにはアスタリスク（*）が付いています。使用するプラグインに応じて、左側に [Input Schema]、中央に [Configuration] セクション、右側に [Output Schema] が表示されます。

シンク プラグインには出力スキーマがなく、ソース プラグインには入力スキーマがありません。また、シンクとソースの両方のプラグインに、リネージのデータソース / シンクを識別するための必須フィールド [Reference Name] があります。

各プラグインに [Label] フィールドがあります。これは、パイプラインが表示されるキャンバスで確認できるノードのラベルです。

2. [Properties] ボックスの右上にある [X] をクリックして閉じます。

3. 次に、Wrangler ノードにカーソルを合わせ、[Properties] をクリックします。

注: Wrangler のようなプラグインには入力スキーマがあります。これらのフィールドは、プラグインに渡されて処理されます。プラグインで処理された後、出力データは出力スキーマでパイプラインの次のノードに送信されるか、シンクの場合はデータセットに書き込まれます。

4. [Properties] ボックスの右上にある [X] をクリックして閉じます。

5. BigQuery ノードにカーソルを合わせて [Properties] をクリックし、次の構成設定値を入力します。

   • [Reference Name] に「Titanic_BQ」と入力します。

   • [Dataset] に「demo」と入力します。

   • [Table] に「titanic」と入力します。

6. [Properties] ボックスの右上にある [X] をクリックして閉じます。

タスク 6. パイプラインをテストする

あとは、パイプラインをテストして、想定どおりに動作することを確認するだけです。その前に、作業内容が失われないように、ドラフトに名前を付けて保存してください。

1. 右上のメニューで [Save] をクリックします。パイプラインの名前を入力し、説明を追加するよう求められます。

   • パイプラインの名前として「ETL-batch-pipeline」を入力します。
   • 説明として「ETL pipeline to parse CSV, transform and write output to BigQuery」と入力します。

2. [保存] をクリックします。

3. パイプラインをテストするために、[Preview] アイコンをクリックします。ボタンバーに [Run] アイコンが表示されます。このアイコンをクリックすると、パイプラインをプレビュー モードで実行できます。

4. [Run] アイコンをクリックします。パイプラインがプレビュー モードで実行されている間は、データが実際に BigQuery テーブルに書き込まれることはありませんが、データが正しく読み取られていることと、パイプラインがデプロイされたときにデータが想定どおりに書き込まれることを確認できます。[Preview] ボタンは切り替え式です。プレビュー モードを終了するときは、もう一度クリックしてください。

   

5. パイプラインの実行が完了したら、Wrangler ノードにカーソルを合わせて [Properties] をクリックします。続けて、[Preview] タブをクリックします。正常に処理されていれば、入力された元データが左側のノードに、出力された解析済みレコードが右側のノードに表示されます。[Properties] ボックスの右上にある [X] をクリックして閉じます。

注: データを操作する各ノードで、同様の出力が表示されます。これは、パイプラインをデプロイする前に、作業の方向性が正しいことを証明できる効果的な方法です。エラーが発生した場合も、ドラフトモードであれば簡単に修正できます。

6. [Preview] アイコンをもう一度クリックして、プレビュー モードをオフにします。

7. ここまで問題がなければ、パイプラインのデプロイに進みます。右上の [Deploy] アイコン（）をクリックして、パイプラインをデプロイします。

パイプラインがデプロイされていることを示す確認ダイアログが表示されます。

8. パイプラインが正常にデプロイされたら、ETL パイプラインを実行して BigQuery にデータを読み込むことができます。

9. [Run] アイコンをクリックして ETL ジョブを実行します。

10. 完了すると、パイプラインのステータスが [Succeeded] に変わり、パイプラインが問題なく実行されたことがわかります。

    

11. パイプラインでデータが処理されると、パイプラインの各ノードから、処理されたレコード数を示す指標が出力されます。 解析オペレーションでは 892 件のレコードが表示されますが、ソースには 893 件のレコードがありました。これは、解析オペレーションで最初の行が列見出しの設定に使用されたことにより、残りの 892 件のレコードが処理対象となったためです。

[進行状況を確認] をクリックして、目標に沿って進んでいることを確認します。 バッチ パイプラインをデプロイして実行する

タスク 7. 結果を見る

パイプラインは出力を BigQuery テーブルに書き込みます。次の手順で、この点を確認できます。

1. 新しいタブで Cloud コンソールの BigQuery UI を開くか、コンソールのタブを右クリックして [タブを複製] を選択し、ナビゲーション メニューを使用して [BigQuery] を選択します。メッセージが表示されたら、[完了] をクリックします。

2. 左側のペインの [エクスプローラ] セクションで、プロジェクト ID（qwiklabs で始まる ID）をクリックします。

3. プロジェクト内の demo データセットで、titanic テーブルをクリックし、[+]（SQL クエリ）をクリックして、次のような簡単なクエリを実行します。

SELECT * FROM `demo.titanic` LIMIT 10

[進行状況を確認] をクリックして、目標に沿って進んでいることを確認します。 結果を見る

お疲れさまでした

ここでは、Cloud Data Fusion の Pipeline Studio に表示される構成要素を使用してバッチ パイプラインを構築する方法を学びました。また、Wrangler を使用してデータの変換ステップを作成する方法も学びました。

次のラボを受講する

「Cloud Data Fusion で Wrangler を使用してデータ変換を作成し、データを準備する」に進んでください。

マニュアルの最終更新日: 2025 年 5 月 6 日

ラボの最終テスト日: 2025 年 5 月 6 日

Copyright 2020 Google LLC All rights reserved. Google および Google のロゴは Google LLC の商標です。その他すべての企業名および商品名はそれぞれ各社の商標または登録商標です。