Cloud Data Fusion を使用して CSV データを BigQuery に取り込む - バッチ取り込み

1. はじめに

最終更新日: 2020 年 2 月 28 日

この Codelab では、CSV 形式の医療データを BigQuery に一括で取り込むデータ取り込みパターンを示します。このラボでは、Cloud Data Fusion バッチデータパイプラインを使用します。現実的な医療テストデータが生成され、Google Cloud Storage バケット（gs://hcls_testing_data_fhir_10_patients/csv/）で利用できるようになっています。

この Codelab では、次のことを学びます。

Cloud Data Fusion を使用して、GCS から BigQuery に CSV データ（バッチスケジュールされた読み込み）を取り込む方法。
Cloud Data Fusion でデータ統合パイプラインを視覚的に構築して、ヘルスケアデータの読み込み、変換、マスキングを一括で行う方法。

この Codelab を実行するには何が必要ですか？

GCP プロジェクトへのアクセス権が必要です。
GCP プロジェクトのオーナーロールが割り当てられている必要があります。
ヘッダーを含む CSV 形式の医療データ。

GCP プロジェクトがない場合は、こちらの手順に沿って新しい GCP プロジェクトを作成します。

CSV 形式の医療データは、gs://hcls_testing_data_fhir_10_patients/csv/ の GCS バケットにプリロードされています。各リソース CSV ファイルには、固有のスキーマ構造があります。たとえば、Patients.csv のスキーマは Providers.csv のスキーマとは異なります。事前に読み込まれたスキーマファイルは、gs://hcls_testing_data_fhir_10_patients/csv_schemas にあります。

新しいデータセットが必要な場合は、SyntheaTM を使用していつでも生成できます。次に、[入力データをコピー] ステップでバケットからコピーするのではなく、GCS にアップロードします。

2. GCP プロジェクトの設定

環境のシェル変数を初期化します。

PROJECT_ID を見つけるには、プロジェクトの識別をご覧ください。

<!-- CODELAB: Initialize shell variables ->
<!-- Your current GCP Project ID ->
export PROJECT_ID=<PROJECT_ID>
<!-- A new GCS Bucket in your current Project  - INPUT ->
export BUCKET_NAME=<BUCKET_NAME>
<!-- A new BQ Dataset ID - OUTPUT ->
export DATASET_ID=<DATASET_ID>

gsutil ツールを使用して、入力データとエラーログを保存する GCS バケットを作成します。

gsutil mb -l us gs://$BUCKET_NAME

合成データセットにアクセスします。

Cloud コンソールへのログインに使用しているメールアドレスから、hcls-solutions-external+subscribe@google.com に参加リクエストを送信します。
操作を確認する手順が記載されたメールが届きます。
メールに返信してグループに参加するオプションを使用します。ボタンをクリックしないでください。
確認メールが届いたら、Codelab の次のステップに進みます。

入力データをコピーします。

gsutil -m cp -r gs://hcls_testing_data_fhir_10_patients/csv gs://$BUCKET_NAME

BigQuery データセットを作成します。

bq mk --location=us --dataset $PROJECT_ID:$DATASET_ID

3. Cloud Data Fusion 環境の設定

Cloud Data Fusion API を有効にして、必要な権限を付与する手順は次のとおりです。

API を有効にします。

GCP Console API ライブラリに移動します。
プロジェクトリストからプロジェクトを選択します。
API ライブラリで、有効にする API を選択します。API を検索する場合は検索フィールドやフィルタを使用します。
[API] ページで、[有効にする] をクリックします。

Cloud Data Fusion インスタンスを作成する。

GCP Console で、ProjectID を選択します。
左側のメニューから [Data Fusion] を選択し、ページの中央にある [インスタンスを作成] ボタン（初回作成）をクリックするか、上部のメニューにある [インスタンスを作成] ボタン（追加作成）をクリックします。

インスタンス名を指定します。[Enterprise] を選択します。

[作成] ボタンをクリックします。

インスタンスの権限を設定します。

インスタンスを作成したら、次の手順で、インスタンスに関連付けられているサービスアカウントにプロジェクトの権限を付与します。

インスタンス名をクリックして、インスタンスの詳細ページに移動します。

サービスアカウントをコピーします。

プロジェクトの IAM ページに移動します。
IAM 権限ページで、サービスアカウントを新しいメンバーとして追加し、Cloud Data Fusion API サービスエージェントのロールを付与します。[追加] ボタンをクリックし、[新しいメンバー] フィールドに「サービスアカウント」を貼り付けて、[サービス管理] -> [Cloud Data Fusion API サーバーエージェント] ロールを選択します。
[保存] をクリックします。

これらの手順が完了したら、Cloud Data Fusion インスタンスページかインスタンスの詳細ページにある [インスタンスを表示] リンクをクリックして、Cloud Data Fusion の使用を開始できます。

ファイアウォールルールを設定します。

GCP Console -> [VPC ネットワーク] -> [ファイアウォールルール] に移動して、default-allow-ssh ルールが存在するかどうかを確認します。

そうでない場合は、すべての上り（内向き）SSH トラフィックをデフォルトネットワークに許可するファイアウォールルールを追加します。

コマンドラインを使用する場合:

gcloud beta compute --project={PROJECT_ID} firewall-rules create default-allow-ssh --direction=INGRESS --priority=1000 --network=default --action=ALLOW --rules=tcp:22 --source-ranges=0.0.0.0/0 --enable-logging

UI を使用する場合: [ファイアウォールルールを作成] をクリックして、情報を入力します。

4. 変換用のスキーマを構築する

GCP に Cloud Fusion 環境ができたので、スキーマを構築しましょう。このスキーマは CSV データの変換に必要です。

Cloud Data Fusion ウィンドウで、[操作] 列の [インスタンスを表示] リンクをクリックします。別のページにリダイレクトされます。表示された URL をクリックして、Cloud Data Fusion インスタンスを開きます。ようこそポップアップで [ツアーを開始] ボタンまたは [いいえ、結構です] ボタンをクリックしたかどうか。
ハンバーガーメニューを開き、[パイプライン] -> [Studio] を選択します。

左側のプラグインパレットの [変換] セクションで、Wrangler ノードをダブルクリックします。このノードは、データパイプライン UI に表示されます。

Wrangler ノードにカーソルを合わせ、[プロパティ] をクリックします。[Wrangle] ボタンをクリックし、目的のスキーマを構築するためにすべてのデータフィールドを含む .csv ソースファイル（patients.csv など）を選択します。
各列名（body など）の横にある下矢印（列の変換）をクリックします。
デフォルトでは、最初のインポートではデータファイルに 1 つの列しかないと想定されます。CSV として解析するには、[解析] → [CSV] を選択し、区切り文字を選択して、必要に応じて [最初の行をヘッダーとして設定] チェックボックスをオンにします。[適用] ボタンをクリックします。
[Body] フィールドの横にある下矢印をクリックし、[Delete Column] を選択して [Body] フィールドを削除します。また、列の削除、一部の列のデータ型の変更（デフォルトは「文字列」型）、列の分割、列名の設定など、他の変換を試すこともできます。

[Columns] タブと [Transformation steps] タブには、出力スキーマと Wrangler のレシピが表示されます。右上にある [適用] をクリックします。[検証] ボタンをクリックします。緑色の「エラーは見つかりませんでした」は成功を示します。

[Wrangler Properties] で、[Actions] プルダウンをクリックして、必要に応じて後でインポートできるように、目的のスキーマをローカルストレージにエクスポートします。
Wrangler レシピを保存して、後で使用できるようにします。

parse-as-csv :body ',' true
drop body

[Wrangler Properties] ウィンドウを閉じるには、[X] ボタンをクリックします。

5. パイプラインのビルドノード

このセクションでは、パイプラインコンポーネントを構築します。

Data Pipelines UI の左上に、パイプラインタイプとして [Data Pipeline - Batch] が選択されていることが表示されます。

左側のパネルには、[フィルタ]、[ソース]、[変換]、[分析]、[シンク]、[条件とアクション]、[エラーハンドラとアラート] などのさまざまなセクションがあり、パイプラインのノードを選択できます。

ソースノード

ソースノードを選択します。
左側のプラグインパレットの [Source] セクションで、Data Pipelines UI に表示される [Google Cloud Storage] ノードをダブルクリックします。
GCS ソースノードにカーソルを合わせ、[Properties] をクリックします。

必須項目を入力します。次のフィールドを設定します。

Label = {任意のテキスト}
参照名 = {任意のテキスト}
プロジェクト ID = 自動検出
パス = 現在のプロジェクトのバケットの GCS URL。例: gs://$BUCKET_NAME/csv/
Format = text
Path Field = filename
Path Filename Only = true
Read Files Recursively = true

[+] ボタンをクリックして、GCS 出力スキーマに「filename」フィールドを追加します。
詳細な説明については、[ドキュメント] をクリックしてください。[検証] ボタンをクリックします。緑色の「エラーは見つかりませんでした」は成功を示します。
[GCS Properties] を閉じるには、[X] ボタンをクリックします。

ノードを変換する

変換ノードを選択します。
左側のプラグインパレットの [変換] セクションで、Data Pipelines UI に表示される [Wrangler] ノードをダブルクリックします。GCS ソースノードを Wrangler 変換ノードに接続します。
Wrangler ノードにカーソルを合わせ、[プロパティ] をクリックします。
[アクション] プルダウンをクリックして [インポート] を選択し、保存したスキーマ（例: gs://hcls_testing_data_fhir_10_patients/csv_schemas/ schema (Patients).json）をインポートします。前のセクションで保存したレシピを貼り付けます。
または、セクション 変換のスキーマを構築するの Wrangler ノードを再利用します。
必須項目を入力します。次のフィールドを設定します。

Label = {任意のテキスト}
入力フィールド名 = {*}
前提条件 = {filename != "patients.csv"}。各入力ファイル（例: patients.csv、providers.csv、allergies.csv など）を Source ノードと区別します。

JavaScript ノードを追加して、ユーザー提供の JavaScript を実行し、レコードをさらに変換します。この Codelab では、JavaScript ノードを使用して、各レコード更新のタイムスタンプを取得します。Wrangler 変換ノードを JavaScript 変換ノードに接続します。JavaScript の [プロパティ] を開き、次の関数を追加します。

function transform(input, emitter, context) {
  input.TIMESTAMP = (new Date()).getTime()*1000;
  emitter.emit(input);
}

[+] 記号をクリックして、TIMESTAMP という名前のフィールドを出力スキーマに追加します（存在しない場合）。データ型としてタイムスタンプを選択します。

詳細については、ドキュメントをクリックしてください。[検証] ボタンをクリックして、入力したすべての情報を検証します。緑色の「エラーは見つかりませんでした」は成功を示します。
[変換のプロパティ] ウィンドウを閉じるには、[X] ボタンをクリックします。

データのマスキングと匿名化

列のプルダウンをクリックし、[データをマスク] で要件に応じてマスキングルールを適用することで、個々のデータ列を選択できます（例: SSN 列）。

Wrangler ノードの [レシピ] ウィンドウで、ディレクティブを追加できます。たとえば、次の構文に従ってハッシュディレクティブとハッシュアルゴリズムを使用して、匿名化を行います。

hash <column> <algorithm> <encode>

<column>: name of the column
<algorithm>: Hashing algorithm (i.e. MD5, SHA-1, etc.)
<encode>: default is true (hashed digest is encoded as hex with left-padding zeros). To disable hex encoding, set <encode> to false.

シンクノード

シンクノードを選択します。
左側のプラグインパレットの [Sink] セクションで、BigQuery ノードをダブルクリックします。このノードがデータパイプライン UI に表示されます。
BigQuery シンクノードにカーソルを合わせ、[プロパティ] をクリックします。

必須項目に入力します。次のフィールドを設定します。

Label = {任意のテキスト}
参照名 = {任意のテキスト}
プロジェクト ID = 自動検出
データセット = 現在のプロジェクトで使用されている BigQuery データセット（DATASET_ID など）
Table = {テーブル名}

詳細については、ドキュメントをクリックしてください。[検証] ボタンをクリックして、入力したすべての情報を検証します。緑色の「エラーは見つかりませんでした」は成功を示します。

[BigQuery Properties] を閉じるには、[X] ボタンをクリックします。

6. バッチデータパイプラインを構築する

パイプライン内のすべてのノードを接続する

ソースノードの右端から接続矢印 > をドラッグして宛先ノードの左端でドロップします。
パイプラインには、同じ GCS ソースノードから入力ファイルを取得する複数のブランチを設定できます。

パイプラインに名前を付けます。

以上です。初めてのバッチデータパイプラインを作成し、パイプラインのデプロイと実行ができる状態になりました。

メールでパイプラインアラートを送信する（省略可）

Pipeline Alert の SendEmail 機能を利用するには、仮想マシンインスタンスからメールを送信するようにメールサーバーを設定する必要があります。詳細については、以下のリファレンスリンクをご覧ください。

インスタンスからのメールの送信 | Compute Engine ドキュメント

この Codelab では、次の手順で Mailgun を介してメールリレーサービスを設定します。

Mailgun を使用したメールの送信 | Compute Engine ドキュメントの手順に沿って、Mailgun でアカウントを設定し、メールリレーサービスを構成します。その他の変更は以下のとおりです。
すべての受信者のメールアドレスを Mailgun の承認済みリストに追加します。このリストは、左側のパネルの [Mailgun] > [Sending] > [Overview] オプションで確認できます。

受信者が support@mailgun.net から送信されたメールで [同意する] をクリックすると、そのメールアドレスが承認済みリストに保存され、パイプラインアラートメールを受信できるようになります。

「始める前に」のセクションのステップ 3 - 次のようにファイアウォールルールを作成します。

「Postfix でメールリレーとして Mailgun を構成する」の手順 3。手順に記載されている [Local Only] ではなく、[Internet Site] または [Internet with smarthost] を選択します。

「Postfix でメールリレーとして Mailgun を構成する」の手順 4。vi /etc/postfix/main.cf を編集して、mynetworks の末尾に 10.128.0.0/9 を追加します。

vi /etc/postfix/master.cf を編集して、デフォルトの smtp（25）をポート 587 に変更します。

Data Fusion Studio の右上にある [構成] をクリックします。[パイプラインアラート] をクリックし、[+] ボタンをクリックして [アラート] ウィンドウを開きます。[SendEmail] を選択します。

[Email] 構成フォームに入力します。アラートのタイプごとに、[実行条件] プルダウンから [完了、成功] または [失敗] を選択します。[ワークフロートークンを含める] = false の場合、[メッセージ] フィールドの情報のみが送信されます。[Include Workflow Token] = true の場合、[Message] フィールドの情報とワークフロートークンの詳細情報が送信されます。[プロトコル] には小文字を使用する必要があります。[送信者] に、会社のメールアドレス以外の「偽の」メールアドレスを使用する。

7. パイプラインの構成、デプロイ、実行/スケジューリング

Data Fusion Studio の右上にある [構成] をクリックします。[Engine Config] で [Spark] を選択します。[Configure] ウィンドウで [Save] をクリックします。

[プレビュー] をクリックしてデータをプレビューし、もう一度 [プレビュー] をクリックして前のウィンドウに戻ります。プレビューモードでパイプラインを **実行** することもできます。

[ログ] をクリックしてログを表示します。
[保存] をクリックして、すべての変更を保存します。
新しいパイプラインを構築するときに保存したパイプライン構成をインポートするには、[インポート] をクリックします。
[エクスポート] をクリックして、パイプライン構成をエクスポートします。
[デプロイ] をクリックして、パイプラインをデプロイします。
デプロイしたら [Run] をクリックし、パイプラインの実行が完了するのを待ちます。

[アクション] ボタンの [複製] を選択して、パイプラインを複製できます。
[アクション] ボタンの [エクスポート] を選択して、パイプライン構成をエクスポートできます。
必要に応じてパイプライントリガーを設定するには、Studio ウィンドウの左端または右端にある [インバウンドトリガー] または [アウトバウンドトリガー] をクリックします。
[スケジュール] をクリックして、パイプラインを定期的に実行してデータを読み込むようにスケジュールします。

[概要] には、実行履歴、レコード、エラーログ、警告のグラフが表示されます。

8. 検証

Validate パイプラインが正常に実行されました。

BigQuery データセットにすべてのテーブルがあるかどうかを検証します。

bq ls $PROJECT_ID:$DATASET_ID

     tableId       Type    Labels   Time Partitioning
----------------- ------- -------- -------------------
 Allergies         TABLE
 Careplans         TABLE
 Conditions        TABLE
 Encounters        TABLE
 Imaging_Studies   TABLE
 Immunizations     TABLE
 Medications       TABLE
 Observations      TABLE
 Organizations     TABLE
 Patients          TABLE
 Procedures        TABLE
 Providers         TABLE

アラートメールを受信します（設定されている場合）。

結果の表示

パイプラインの実行後、次の手順で結果を確認します。

BigQuery UI でテーブルに対してクエリを実行します。BigQuery UI に移動
次のクエリを、実際のプロジェクト名、データセット、テーブルに更新します。

9. クリーンアップ

このチュートリアルで使用したリソースについて、Google Cloud Platform アカウントに課金されないようにする手順は次のとおりです。

チュートリアルが終了したら、GCP で作成したリソースをクリーンアップして、割り当てを使い果たしたり、今後料金が発生しないようにします。次のセクションで、このようなリソースを削除または無効にする方法を説明します。

BigQuery データセットの削除

このチュートリアルで作成した BigQuery データセットを削除するには、次の手順を行います。

GCS バケットの削除

このチュートリアルで作成した GCS バケットを削除する手順は次のとおりです。

Cloud Data Fusion インスタンスの削除

手順に沿って Cloud Data Fusion インスタンスを削除します。

プロジェクトの削除

課金をなくす最も簡単な方法は、チュートリアル用に作成したプロジェクトを削除することです。

プロジェクトを削除するには、次の操作を行います。

注意: プロジェクトを削除すると、次のような影響があります。

プロジェクト内のすべてのものが削除されます。このチュートリアルで既存のプロジェクトを使用した場合、それを削除すると、そのプロジェクトで行った他の作業もすべて削除されます。
カスタムプロジェクト ID が失われます。このプロジェクトを作成したときに、将来使用するカスタムプロジェクト ID を作成した可能性があります。appspot.com URL など、プロジェクト ID を使用する URL を保持するには、プロジェクト全体を削除するのではなく、プロジェクト内の選択したリソースを削除します。

複数のチュートリアルとクイックスタートを実施する予定がある場合は、プロジェクトを再利用すると、プロジェクトの割り当て上限を超えないようにできます。

GCP Console で [プロジェクト] ページに移動します。プロジェクトページに移動
プロジェクトリストで、削除するプロジェクトを選択し、[削除] をクリックします。
ダイアログでプロジェクト ID を入力し、[シャットダウン] をクリックしてプロジェクトを削除します。

10. 完了

お疲れさまでした。Cloud Data Fusion を使用して BigQuery に医療データを取り込むコードラボを完了しました。

Google Cloud Storage から BigQuery に CSV データをインポートした。

ヘルスケアデータの読み込み、変換、マスキングを一括で行うためのデータ統合パイプラインを視覚的に構築しました。

これで、Google Cloud Platform の BigQuery でヘルスケアデータ分析を開始するために必要な主な手順を理解できました。