---
title: "Data Fusionでデータプラットフォームを作成する | grasys blog"
url: "https://blog.grasys.io/post/yamada-k/data-fusion"
description: "こんにちは。エンジニアの山田です。 昨今ではAI（人工知能）の技術も進歩し、徐々に活用され始めていますが、例えば画像認識を使用したAIでは大量に蓄積された画像データから人の特徴や動作を学習（ディープラーニング）し、高い精度で個人を特定することができますし、ECサイトや動画配信サイトで使用されているAIでは個人の閲覧履…"
---

# Data Fusionでデータプラットフォームを作成する

-   ![](/_astro/noicon.CTHOhNiB_1HMs6A.webp)[yamada.k](/authors/yamada-k/)
-   公開日：2023年1月23日
-   カテゴリー：[Tech](/categories/tech/)
-   タグ：[#BigQuery](/tags/bigquery/)[#CSV](/tags/csv/)[#datafusion](/tags/datafusion/)[#Google Cloud](/tags/google-cloud/)[#Cloud Storage](/tags/cloud-storage/)

![Data Fusionでデータプラットフォームを作成する](/_astro/hero._mTIZ_Ah_Z2aWm3i.webp)

こんにちは。エンジニアの山田です。

昨今ではAI（人工知能）の技術も進歩し、徐々に活用され始めていますが、例えば画像認識を使用したAIでは大量に蓄積された画像データから人の特徴や動作を学習（ディープラーニング）し、高い精度で個人を特定することができますし、ECサイトや動画配信サイトで使用されているAIでは個人の閲覧履歴を蓄積、分析し、おすすめを提示してくれます。

また、従来から自社の経営状況をデータとして蓄積し、分析することにより、既存事業の改善や新しい事業の開拓なども行われてきました。

上記の通りデータを分析し、活用することは大半の業務にとってプラスになり得ます。長々と蛇足を書きましたが、今回はそんなデータを分析するためのプラットフォームとして、Google Cloud (GCP)からフルマネージドで難しいコーディングが不要なETLツールが提供されていますので、そちらの紹介をしたいと思います。

## Data Fusionとは？

Google Cloud (GCP)で提供されているフルマネージドのETL（Extract （抽出）、Transform （変換）、 Load （格納）サービスです。

主な用途は各種データソースから情報を収集、分析し、結果を必要とするシステム（そのシステムを閲覧する部署や顧客）に連携することになります。

データソースには各種クラウド（Google Cloud (GCP)、AWS、Azure）のストレージサービスに格納されたExcelやCSVといったファイル、データベースサービス（BigQuery、CloudSQL、Datastore）、またはオンプレミスに構築されているデータウェアハウスなどを指定することができます。

ETLサービスは他にもいくつかあると思いますが、Data Fusionを使用するメリットとしては以下があります。

-   GUIで設定することができることから、データフローを視覚化でき、データ加工を行うための複雑なコーディングが不要になる。
    
-   ETL処理はHadoop、Apache SparkなどのOSSが採用されたフルマネージドのDataprocサービスが自動で起動して処理を実行するため、実行環境の考慮が不要になる。
    
-   プラグインを追加することにより、Salesforceなどのサードベンダーが提供している外部サービスとのデータ連携が可能になる。
    

## Data Fusionの構築

まずはGCPのコンソール上で `data fusion`を検索します。

![Data Fusionの構築の画面（1）](/_astro/01-image-02d2d328.BZ-rEtl1_14C8n5.webp)

以下のように検索結果に `Data Fusion` が表示されますのでクリックします。

![Data Fusionの構築の画面（2）](/_astro/02-image-1-f3783ebe.BmzL_b_j_ZKFwIP.webp)

初回アクセスの場合はAPI（Data Fusion API）の有効化画面が表示されますので有効にします。

有効化が完了するとインスタンスが作成できるようになりますので `インスタンスを作成` をクリックします。

![Data Fusionの構築の画面（3）](/_astro/03-image-4-1024x328-331e9e88.cffFCEHn_ZIxw5q.webp)

インスタンスの作成に必要なパラメータを入力欄に入力し、`作成` ボタンをクリックします。

-   インスタンス名：（任意のインスタンス名を入力）
    
-   説明：任意（Data Fusionインスタンスの用途などを入力）
    
-   リージョン：asia-northeast1（国内利用の場合はasia-northeastを選択）
    
-   バージョン：（デフォルトで Current relase が選択済）
    
-   エディション：（使用用途に応じたエディションを選択 ※）
    
-   Dataprocサービスアカウント：（デフォルトのサービスアカウントが選択済）
    

※ 詳細オプションを設定する場合は上記以外にも設定が必要になります。（後述）

※ エディションに関する説明は以下の公式ドキュメントに記載されています。

[Data Fusion の料金](https://cloud.google.com/data-fusion/pricing?hl=ja)

エディションは主に使用人数、環境（開発 or 本番）、高可用性の要否、パイプライン（Data Fusionの処理フロー）の最大同時実行数あたりとコスト感（料金）を考慮して選択する必要があります。

| エディション | ユーザー数 | ワークロード | 高可用性 | 最大同時実行数 | 料金 |
| --- | --- | --- | --- | --- | --- |
| Developer | 2（推奨） | 開発 | ゾーン | 5 | 約 $250/月 |
| Basic | 無制限 | テスト | リージョン(小容量) | 40 | 約 $1,100/月 |
| Enterprise | 無制限 | 本番 | リージョン(大容量) | 200 | 約 $3,000/月 |

![Data Fusionの構築の画面（4）](/_astro/04-image-2-1-766x1024-7082481a.CHjxi1tK_Z2eILUA.webp)

詳細オプションは必要に応じて入力する必要があります。

外部システムとの連携が不要な場合（GCP内部でのみの利用の場合）は `プライベートIPを有効化` にチェックを入れて内部接続のみに制限したり、監視（アラート通知）を強化したい場合は `Stackdriver Logging サービスを有効にする` `Stackdriver Monitoring サービスを有効にする` のチェックを入れてStackdriver（監視ツール）を有効にしたりすることができます。

![Data Fusionの構築の画面（5）](/_astro/05-image-5-808x1024-cbae6f63._V_ldtNj_ZgHywE.webp)

## パイプライン（データフロー）の作成

インスタンスの作成が完了すると以下の画面が表示されますので `インスタンスの表示` のリンクをクリックしてData Fusionの操作画面（GUI）を表示させます。

![パイプライン（データフロー）の作成の画面（1）](/_astro/06-image-7-1024x254-e27dd150.DXqgRDX7_Z2rWcKz.webp)

画面が表示されたら左上のハンバーガーメニューをクリックしてメニューを表示させ、`Studio` をクリックします。

![パイプライン（データフロー）の作成の画面（2）](/_astro/07-image-8-1024x615-3d94adae.DgxAFAwp_Z1DCoL8.webp)

今回はシンプルに以下のフローを作成します。

-   GCS上のCSVを読み込む。
    
-   データベースに取り込めるようにCSVのカラムをデータ型に変更する。
    
-   変換したデータをBigQueryに取り込む。
    

まず左ペインの`Source` から`GCS` をクリックします。

※ 以下の画面がデータフローを作成するGUIの操作画面になります。

![パイプライン（データフロー）の作成の画面（3）](/_astro/08-image-9-1024x636-84ccf5c5.CS4H2GIt_1SXFvQ.webp)

GCSのノード（パイプライン ノード）が生成されますので、ノード内の `Properties` をクリックします。

![パイプライン（データフロー）の作成の画面（4）](/_astro/09-image-10-1024x455-47048b05.BcTACPpV_Z1mzIkK.webp)

`Use Connection` をクリックして `YES` に変更し、`BROWSE CONNECTIONS` ボタンをクリックします。

![パイプライン（データフロー）の作成の画面（5）](/_astro/10-image-11-1024x305-a19b62da.C8oF8atP_Z1KVycv.webp)

`Name` 列の `Cloud Storage Default` をクリックします。

![パイプライン（データフロー）の作成の画面（6）](/_astro/11-image-12-1024x218-f3ff8dcc.BfpmCR4b_29DyIK.webp)

`BROWSE` ボタンをクリックします。

![パイプライン（データフロー）の作成の画面（7）](/_astro/12-image-13-1024x116-7da089d0.C1L-HU2K_Z1lgtdI.webp)

GCSのバケット一覧からファイルが格納されているバケットをクリックし、取り込む対象のcsvを選択します。

※ 今回は事前にバケットを作成し、データ取り込み用のcsvを格納してあります。（csvファイルの内容は以下）

![パイプライン（データフロー）の作成の画面（8）](/_astro/13-image-18-232be8bd.COFrzwwR_Z2bcpGU.webp)

![パイプライン（データフロー）の作成の画面（9）](/_astro/14-image-14-1024x29-8d1930f2.Dcu7Lpmr_1K6O1p.webp)

`Format` を `csv` にすると `Enable Quoted Values` と `Use First Row as Header` が表示されるのでそれぞれクリックして `True` に変更します。

※ `Enable Quoted Values` はcsvの値がクォートで囲まれている場合、`Use First Row as Header` はcsvの1行目をヘッダーとして利用する場合に `True` にする必要があります。

![パイプライン（データフロー）の作成の画面（10）](/_astro/15-image-15-1024x515-a2211fa6.zptPKKy6_Zg9Mke.webp)

上の画面の `GET SCHEMA` ボタンをクリックすると画面右がCSVのヘッダー名に更新されるのですべての項目にチェックを入れます。

※ 右上の `Validate` ボタンをクリックすることで設定に問題がないかチェックすることができます。

![パイプライン（データフロー）の作成の画面（11）](/_astro/16-image-16-1024x570-26715426.Cz_BU-t9_1JpAou.webp)

右上の `×` でGCSのProperties画面を閉じ、Studioの画面に戻ります。

次は左ペインの `Transform` から `Wrangler` をクリックします。

`Wrangler` のノードが生成されますのでGCSノードの右端から矢印を引っ張り連結した上でWranglerノード内の `properties` をクリックします。

![パイプライン（データフロー）の作成の画面（12）](/_astro/17-image-20-1024x770-b534a22a.Mq1eKTBe_kDRcF.webp)

Wranglerではデータベースにデータとして取り込むためのデータ型変換を行います。

WranglerのProperties画面中央に `Directives` 項目があるので `Recipe` の入力欄に以下を入力します。

Plain textcontent\_copy

```
parse-as-datetime :register_date "yyyy-MM-dd HH:mm:ss.SSS"
```

※ `Directives` で実行可能なコマンドはCDAPで提供されています。

[CDAP](https://cdap.atlassian.net/wiki/spaces/DOCS/pages/1128988735/Parse+as+Datetime+directive)

![パイプライン（データフロー）の作成の画面（13）](/_astro/18-image-21-1024x311-1bad1d8e.nUFNqENp_Z23bR0g.webp)

日付型についてはDirectivesでの明示的な型変換が必要になりますが、文字列型から整数型への変換については画面右の `Outpu Schema` のプルダウンでの変更のみで変換されます。

![パイプライン（データフロー）の作成の画面（14）](/_astro/19-image-23-068bbfbb.C_8xmEy3_Z1JGxrc.webp)

念のため、GCS Propertiesのときと同様に `Validate` ボタンで設定をチェックした上で `×` ボタンでProperties画面を閉じます。

最後に左ペインの `Sink` から `BigQuery` をクリックします。

`BigQuery` のノードが生成されますのでWranglerノードの右端から矢印を引っ張り連結した上でBigQueryノードの `properties` をクリックします。

![パイプライン（データフロー）の作成の画面（15）](/_astro/20-image-24-1024x570-b272ed6d.B95O2D1u_Z1m4uTQ.webp)

`Use Connection` をクリックして `Yes` に変更し、`BROWSE CONNECTIONS` ボタンをクリックします。

![パイプライン（データフロー）の作成の画面（16）](/_astro/21-image-25-1024x153-71d0e117.D0IFG9Hl_ZFCFfk.webp)

`Name` 列の `BigQuery Default` をクリックします。

![パイプライン（データフロー）の作成の画面（17）](/_astro/22-image-26-1024x193-41fa0a84.Dwrkg3hb_Z1bEBnh.webp)

`BROWSE` ボタンをクリックします。

![パイプライン（データフロー）の作成の画面（18）](/_astro/23-image-27-1024x134-bc6b2e6e.Do_GbP-j_Z1KG9BN.webp)

表示されているBigQueryのデータセット一覧から対象のデータセットをクリックし、データを投入するテーブルを選択します。

![パイプライン（データフロー）の作成の画面（19）](/_astro/24-image-28-1024x140-f9752c80.pvmhTlOC_LfG7s.webp)

※ 今回は事前にBigQuery上にテーブルを作成してあります。（テーブルの内容は以下）

![パイプライン（データフロー）の作成の画面（20）](/_astro/25-image-37-120b11b6.C9ICksGf_WSxJ4.webp)

`Operation` で `Insert` を選択し、`Truncate Tables` を `True` にクリックして変更します。

※ 取り込み元のcsvの更新の仕方によっては、`Truncate` せず `Update` や `Upsert` の方が適している場合もあります。

![パイプライン（データフロー）の作成の画面（21）](/_astro/26-image-31-1024x442-e7380bdd.CwoVZ-Kx_2pivLc.webp)

画面最下部の `Output Schema` のすべてにチェックを入れます。

![パイプライン（データフロー）の作成の画面（22）](/_astro/27-image-32-1024x154-fee97668.BA3oQaR__auFOI.webp)

念のため、他のPropertiesのときと同様に `Validate` ボタンで設定をチェックした上で `×` ボタンでProperties画面を閉じます。

Studioの左上をクリックして入力欄を表示させ、上部にパイプライン名、下部にパイプラインの説明を記載し、`Save` ボタンをクリックします。

![パイプライン（データフロー）の作成の画面（23）](/_astro/28-image-34-1024x540-713e46be.BjLvONoL_1zawHX.webp)

Studioの右上の `Deploy` をクリックします。（パイプラインとして保存されます）

![パイプライン（データフロー）の作成の画面（24）](/_astro/29-image-35-e6b9ce40.Bu7H4ekl_Z1LhnvN.webp)

画面上部の `Run` ボタンをクリックしてパイプライン処理を実行します。

![パイプライン（データフロー）の作成の画面（25）](/_astro/30-image-36-1024x766-be2da3e9.DwDU6eux_Z15kAHg.webp)

左上の `Status` が `Succeeded` になることを確認します。  
※ `Failed` になる場合は `Logs` にてログを確認します。

![パイプライン（データフロー）の作成の画面（26）](/_astro/31-image-38-1024x125-2705c97f.Cv520q17_yGrfG.webp)

以上でデータフローの処理は完了ですので結果としてBigQuery上にデータが投入されていることを確認します。

![パイプライン（データフロー）の作成の画面（27）](/_astro/32-image-39-1024x303-3f71357e.Y9WbLn0y_GQlbz.webp)

以上でETL処理用のパイプラインの作成は完了になります。

別途、Webアプリケーションを作成し、BigQueryからデータを参照できるような仕組みを作ることによって、ユーザーが必要な情報を参照できるようにすることも可能です。

## 補足：SQLを使用しないテーブル操作

各種データソースからデータを収集してデータベースに登録した後、そのデータを加工、集計したいといったケースもあると思いますので、その方法の一例について記載します。

今回は以下のフローを作成し、これまでに作成した `employee` （社員）テーブルと `position` （役職）テーブルを結合、`position_count` （役職者数集計）テーブルに集計した結果を投入します。

-   BigQueryの `employee` テーブルと `position` テーブルを読み込む。
    
-   読み込んだ２つのテーブルを結合する。
    
-   結合したテーブルの `potision` カラムの項目（役職）ごとの数をカウントする。
    
-   カウントした結果をBigQueryの `position_count` テーブルに取り込む。
    

※ 求める結果を出すのに２つ目の結合処理は特に必要ありませんが、結合処理の紹介のためにあえて追加してあります。

まず以下のテーブルを作成します。

![補足：SQLを使用しないテーブル操作の画面（1）](/_astro/33-image-40-1867117a.BX3lN_A7_Z2tdDPT.webp)

![補足：SQLを使用しないテーブル操作の画面（2）](/_astro/34-image-41-fc94cd35.CF26bKeD_Z1HzbWX.webp)

`position` テーブルには実データも追加しておきます。

![補足：SQLを使用しないテーブル操作の画面（3）](/_astro/35-image-42-1024x426-adf244bc.CKHSrcxD_Bqj0j.webp)

Data FusionのGUIの操作方法は記載してきましたので、ここでは完成したパイプラインをお見せします。

![補足：SQLを使用しないテーブル操作の画面（4）](/_astro/36-image-43-1024x279-945d8a5f.CT1UvcLC_ZV2ryQ.webp)

ここで登場している `Joiner` はSQLで言うところの内部結合（INNER JOIN）、外部結合（OUTER JOIN）の処理を行うパイプラインノードになります。Group Byはその名の通りSQLで言うところのグループ化（GROUP BY）の処理を行うパイプラインノードになります。

`Joiner` の設定ですが `Join Type` で `inner` を選択し、`id`のカラムを使用して結合（INNNER JOIN）するように設定しています。

![補足：SQLを使用しないテーブル操作の画面（5）](/_astro/37-image-44-1024x385-e823c927.D5MYr7p0_Z1hHJ71.webp)

カラムは `employee` テーブルから `id` `name` `mail` `register_date` のカラムを抽出、`` `position` `` テーブルから `position` カラムを抽出して結合するように設定していますのでSQLで言うところの `LEFT INNNOR JOIN` の形式になります。

![補足：SQLを使用しないテーブル操作の画面（6）](/_astro/38-image-45-1024x928-6350f808.Bobr2JIW_h5tC2.webp)

結合されたテーブルのカラムは右端の `Output Schema` に表示されます。

![補足：SQLを使用しないテーブル操作の画面（7）](/_astro/39-image-46-a770ca76.BIv0VMyJ_ZoeF41.webp)

続いて `Group By` の設定ですが `position` のカラムで `Group by` しており、 `Aggregates` で `position_count` というカラム名を宣言して `count`（集計）を行なっています。

![補足：SQLを使用しないテーブル操作の画面（8）](/_astro/40-image-47-1024x328-1d39e1ff.Dp33bCsx_Z22jQNQ.webp)

集計結果のカラムは右端の `Output Schema` に表示されます。

![補足：SQLを使用しないテーブル操作の画面（9）](/_astro/41-image-48-480a995b.DRJd0GO__EAVbR.webp)

このパイプラインを実行した結果として `position_count` テーブルに集計結果が入ります。

![補足：SQLを使用しないテーブル操作の画面（10）](/_astro/42-image-49-35916c7b.DHLmexr4_ZpYYdI.webp)

以上がSQLを使用しないテーブルの操作方法になりますがいかがでしょうか？

SQLの考え方（結合、集計）自体は前提知識として必要かもしれませんが、SQL文自体を書く必要はありませんので、頭の中で構文が曖昧だったり、そもそもSQLが苦手だったりしても直感的にできてしまうと思います。

## この記事を書いた人

[![](/_astro/yamada-k.BijuJbe__ZJYP2S.webp)](/authors/yamada-k/)

### [yamada.k](/authors/yamada-k/)

yamada.kのプロフィールと執筆記事をご覧いただけます。

[プロフィールと記事一覧](/authors/yamada-k/)