Databricksレイクハウス上でFivetranとdbtを用いてどのようにマーケティング分析ソリューションを構築するのか

[Marketing Analytics on the Databricks Lakehouse using Fivetran and dbt \- The Databricks Blog](https://www.databricks.com/blog/2022/08/03/how-to-build-a-marketing-analytics-solution-using-fivetran-and-dbt-on-the-databricks-lakehouse.html)の翻訳です。

:::note warn
本書は抄訳であり内容の正確性を保証するものではありません。正確な内容に関しては原文を参照ください。
:::

マーケティングチームは、マーケティングとセールスキャンペーンをドライブするために多くの異なるプラットフォームを用いており、膨大な量の価値を生み出しますが、データは分断されています。これらのデータとすべてまとめることで、キャンペーンの収益を最大50%増加させた[Publicis Groupe](https://www.databricks.com/customers/publicis-groupe)のように、大きな投資対効果をもたらす役に立ちます。

単一のプラットフォームでデータウェアハウスとAIユースケースを統合するDatabricksレイクハウスは、マーケティング分析ソリューションを構築するには理想的な場所となります。信頼できる唯一の情報源を保持し、AI/MLユースケースを解放します。また、[解約分析、生涯価値分析](https://www.databricks.com/solutions/accelerators/survival-analysis-for-churn-and-lifetime-value)、[顧客セグメンテーション](https://www.databricks.com/solutions/accelerators/customer-segmentation)、[広告効果測定](https://www.databricks.com/solutions/accelerators/sales-forecasting)を含む広範なマーケティング分析ユースケースを実現可能にする2つのDatabricksのパートナーソリューション、Fivetranとdbtを活用します。
![](https://www.databricks.com/wp-content/uploads/2022/07/db-262-blog-img-1.png)
*DatabricksクラスターとDatabricks SQLを用いて、FivetranとdbtはDelta Lakeに対して読み書きを行えます*

Fivetranを用いることで、複雑なパイプラインを構築、管理する必要なしに50以上のマーケティングプラットフォームをDelta Lakeに容易にデータを取り込むことができます。マーケティングプラットフォームのAPIの変更があった場合、Fivetranはインテグレーションをアップデートし、修正するので、マーケティングデータはあなたのデータパイプラインに流れ込み続けます。

dbtは、シンプルなSQLを用いてレイクハウスのユーザーはデータパイプラインを構築できる人気のあるオープンソースフレームワークです。すべてのものは、テキストとしてディレクトリで整理されるので、バージョン管理、デプロイメント、テスト可能性が容易になります。データがDelta Lakeに取り込まれると、データを変換、テスト、ドキュメント化するためにdbtを活用することができます。そして、取り込まれたデータに基づいて構築された変換済みマーケティング分析データマートは、新たなマーケティングキャンペーンや取り組みを推進するために活用できる準備が整ったことになります。

Fivetranとdbtの両方は、Databricksプラットフォーム内でデータ、分析、AIツールを検索し、セキュアに直接接続できるワンストップポータルである[Partner Connect](https://www.databricks.com/jp/partnerconnect)に組み込まれています。数クリックだけで、お使いのDatabricksワークスペースから直接これらのツール(それ以外のツールも)を設定し、接続することができます。

# マーケティング分析ソリューションの構築方法

このハンズオンデモでは、Fivetranを用いてMarketoとSalesforceのデータをDatabricksに取り込み、マーケティング分析データモデルを変換、テスト、ドキュメント化するためにdbtを使用する方法を説明します。

このデモのすべてのコードは[workflows\-examples repository](https://github.com/databricks/workflows-examples/tree/main/marketing_analytics_demo)のGithubから利用できます。
![](https://www.databricks.com/wp-content/uploads/2022/07/db-262-blog-img-2.png)
*データソースとモデルを示すdbtのリネージュグラフ*

最終的なdbtモデルのリネージュグラフはこのようになります。Fivetranのソーステーブルは左側の緑であり、最終的なマーケティング分析モデルは右側となります。モデルを選択することで、それぞれのモデルのに応する依存関係が紫でハイライトされます。

# Fivetranを用いたデータ取り込み

![](https://www.databricks.com/wp-content/uploads/2022/07/db-262-blog-img-3.png)
*Fivetranには数多くのマーケティング分析データソースのコネクターがあります*

Delta Lakeへのマーケティングデータの取り込みをスタートするために、Fivetranで新規のSalesforceとMarketoのコネクションを作成します。また、Fivetranは、自動でDelta Lakeにそれぞれのデータソースの[スキーマを作成し、管理](https://docs.databricks.com/sql/language-manual/sql-ref-syntax-ddl-create-schema.html)します。このデータを変換、クレンジング、集計するために、あとでdbtを活用します。
![](https://www.databricks.com/wp-content/uploads/2022/07/db-262-blog-img-4.png)
*Salesforceデータソースに対するスキーマの定義*

デモのために、Delta Lakeに作成されるスキーマを`marketing_salesforce`と`marketing_marketo`と名付けます。スキーマが存在しない際、Fivetranは最初の取り込みロード処理の一部としてこれらを作成します。
![](https://www.databricks.com/wp-content/uploads/2022/07/db-262-blog-img-5.png)
*Delta Lakeテーブルとして同期するデータソースオブジェクトを選択*

Delta Lakeにどのオブジェクトを同期するのかを選択することができ、これらのオブジェクトのそれぞれが別々のテーブルとして保存されます。また、Fivetranは、それぞれテーブルで何のカラムが同期されているのかの管理、参照をシンプルにします。
![](https://www.databricks.com/wp-content/uploads/2022/07/db-262-blog-img-6.png)
*同期された月間アクティブ行をモニタリングするためのFivetranダッシュボード*

さらに、それぞれのテーブルが日次、月次で何行同期されたのかを示す[アクティブ行](https://fivetran.com/docs/getting-started/consumption-based-pricing#monthlyactiverows)が何行あるのかを分析するために、Fivetranはモニタリングダッシュボードを提供します。

# dbtを用いたデータモデリング

すべてのマーケティングデータをDelta Lakeに取り込んだので、以下のステップに沿ってデータモデルを作成するためにdbtを活用します。

## ローカルでdbtプロジェクトをセットアップし、Databricks SQLに接続

[set\-up instructions for dbt Core and dbt\-databricks](https://docs.databricks.com/integrations/prep/dbt.html)の手順に従って、お好きなIDEでローカルdbt開発環境をセットアップします。

以下の情報を聞いてくる`dbt init`を用いて、新規dbtプロジェクトを作成し、[Databricks SQLウェアハウス](https://docs.databricks.com/sql/admin/sql-endpoints.html)に接続します。

```
$ dbt init
Enter a name for your project (letters, digits, underscore): 
Which database would you like to use?
[1] databricks
[2] spark

Enter a number: 1
host (yourorg.databricks.com): 
http_path (HTTP Path): 
token (dapiXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX): 
schema (default schema that dbt will build objects in): 
threads (1 or more) [1]: 
```

プロファイルを設定すると、以下を用いて接続をテストすることができます。

```
$ dbt debug
```

## ステージングのためにFivetran dbtモデルパッケージをインストール

MarketとSalesforceデータを用いた最初のステップは、我々のモデルのソースとしてテーブルを作成することです。幸運なことに、Fitetranでは、ビルド済みの[Fivetran dbtモデルパッケージ](https://hub.getdbt.com/fivetran/)を提供しており、容易にセットアップして実行できるようになっています。このデモでは、[marketo\_source](https://hub.getdbt.com/fivetran/marketo_source/latest/)と[salesforce\_source](https://hub.getdbt.com/fivetran/salesforce_source/latest/)を活用します。

パッケージをインストールするには、あなたのdbtプロジェクトのルートに** packages.yml**ファイルを追加し、`marketo-source`、`salesforce-source`、`fivetran-utils`パッケージを追加するだけです。

```yaml:packages.yml
packages:
  - package: dbt-labs/spark_utils
    version: 0.3.0
  - package: fivetran/marketo_source
    version: [">=0.7.0", "<0.8.0"]
  - package: fivetran/salesforce_source
    version: [">=0.4.0", "<0.5.0"]
```

パッケージをダウンロードし使用するには、以下を実行します。

```
$ dbt deps
```

packagesフォルダにFivetranパッケージがインストールされていることを確認できるはずです。

## Fivetran dbtモデルdbt_project.ymlの更新

Fivetranパッケージが適切にDatabrikcsで動作するようにするために、`dbt_project.yml`ファイルに幾つかの設定を行う必要があります。

`dbt_project.yml`ファイルは、あなたのdbtプロジェクトのルートフォルダにあるはずです。

### spark_utilsでdbt_utilsマクロを上書き

Fivetran dbtモデルは[dbt\_utils](https://hub.getdbt.com/dbt-labs/dbt_utils/latest/)パッケージのマクロを活用しますが、これらのマクロのいくつかをDatabricksで動作するように変更する必要がありますが、これは[spark\_utils](https://hub.getdbt.com/dbt-labs/spark_utils/latest/)パッケージを用いることで簡単に行えます。

これは、`dbt_project.yml`ファイルで[dispatch config](https://docs.getdbt.com/reference/dbt-jinja-functions/dispatch#overriding-package-macros)を用いて設定できる特定の`dbt_utils`マクロに対してシム(既存のコードの動作を修正するために使われるコード)を指定することで動作し、これによって`dbt_utils`名前空間からマクロを解決する際に、dbtは最初に`spark_utils`パッケージのマクロを検索します。

```yaml
dispatch:
 - macro_namespace: dbt_utils
   search_order: ['spark_utils', 'dbt_utils']
```

### marketo_sourceとsalesforce_sourceスキーマの変数

Fivetranパッケージでは、Fivetranによってデータが取り込まれた際に、どこに取り込むのかを指定するために、[カタログ](https://docs.databricks.com/sql/language-manual/sql-ref-syntax-ddl-create-catalog.html)(dbtではデータベースとして参照されます)と[スキーマ](https://docs.databricks.com/sql/language-manual/sql-ref-syntax-ddl-create-schema.html)を指定することが必要となります。

適切なカタログ名、スキーマ名としてこれらの変数を`dbt_project.yml`ファイルに追加します。`_database`が空の場合、デフォルトカタログの`hive_metastore`が使用されます。Fivetranでコネクションを作成する際に定義したものがスキーマ名となります。

```yaml
vars:
 marketo_source:
   marketo_database: # leave blank to use the default hive_metastore catalog
   marketo_schema: marketing_marketo
 salesforce_source:
   salesforce_database: # leave blank to use the default hive_metastore catalog
   salesforce_schema: marketing_salesforce
```

### Fivetranステージングモデルのターゲットスキーマ

Fivetranによって作成されるすべてのステージングテーブルが、デフォルトのターゲットスキーマに作成することを避けるには、別のステージングスキーマを定義することが有効です。

`dbt_project.yml`ファイルにステージングのスキーマ名を追加すると、デフォルトのスキーマ名の末尾に追加されます。

```yaml
models:
 marketo_source:
   +schema: your_staging_name # leave blank to use the default target_schema
 salesforce_source:
   +schema: your_staging_name # leave blank to use the default target_schema
```

上の内容に基づいて、`profiles.yml`で定義されているターゲットスキーマが**mkt_analytics**の場合、marketo_sourceとsalesforce_sourceテーブルで使用されるスキーマは、**mkt_analytics_your_staging_name**となります。

### 不在テーブルの無効化

このステージでは、適切に動作することをテストするためにFivetranのモデルパッケージを実行することができます。

```
dbt run –select marketo_source
```

```
dbt run –select salesforce_source
```

Fivetranで同期するものとして選択しなかったことで、テーブルが存在しないためにいずれかのモデルが失敗した場合、**dbt_project.yml**ファイルを更新することで、ソーススキーマでこれらのモデルを無効化することができます。

例えば、メールが未達となり、メールテンプレートテーブルがMarketソーススキーマに存在しない場合、モデルの設定に以下を追加することで、これらオンテーブルのモデルを無効化することができます。

```yaml
models:
 marketo_source:
   +schema: your_staging_name 
   tmp:
     stg_marketo__activity_email_bounced_tmp:
       +enabled: false
     stg_marketo__email_template_history_tmp:
       +enabled: false
   stg_marketo__activity_email_bounced:
     +enabled: false
   stg_marketo__email_template_history:
     +enabled: false
```

# マーケティング分析モデルの開発

![](https://www.databricks.com/wp-content/uploads/2022/07/db-262-blog-img-7.png)
*スタースキーマと集計テーブルのデータモデルを示すdbtのリネージュグラフ*

ここまででFivetranパッケージは、マテリアライズされた集計テーブルを伴うスタースキーマデータモデルとなるマーケティング分析ユースケースのデータモデルを開発し始めたモデルの作成、テスト、ステージングを行なってきました。

例えば、最初のマーケティング分析ダッシュボードとして、特定の企業や地域でメールのキャンペーンが開かれ、クリックされた数に基づいて、どの程度エンゲージメントできているのかを見たいと考えます。

このためには、Salesforceユーザーのメール、Salesforceのaccount_id、Marketoのlead_idを用いてSalesforceのテーブルとMarketoテーブルをjoinします。

以下のように、martフォルダ配下にモデルが整理されます。

```
marketing_analytics_demo
|-- dbt_project.yml
|-- packages.yml
|-- models
      |-- mart
             |-- core
             |-- intermediate
             |-- marketing_analytics
```

Githubの[/models/mart](https://github.com/databricks/workflows-examples/tree/main/marketing_analytics_demo/dbt/models/mart)ディレクトリにすべてのモデルのコードを確認することができ、以下ではサンプルとともにそれぞれのフォルダが何であるのかを説明します。

## コアモデル

コアモデルは、すべての後段のモデルの構築に用いられるファクトとディメンジョンテーブルとなります。

`dim_user`モデルのdbt SQLコードです。

```sql:SQL
with salesforce_users as (
   select
       account_id,
       email
   from {{ ref('stg_salesforce__user') }}
   where email is not null and account_id is not null
),
marketo_users as (
   select
       lead_id,
       email
   from {{ ref('stg_marketo__lead') }}
),
joined as (
   select
     lead_id,
     account_id
   from salesforce_users
     left join marketo_users
     on salesforce_users.email = marketo_users.email
)

select * from joined
```

また、フォルダのyamlファイルを用いてモデルに対するドキュメントとテストを追加することができます。

追加されている**core.yml**ファイルには2つのシンプルなテストが含まれています。

```yaml
version: 2

models:
 - name: dim_account
   description: "The Account Dimension Table"
   columns:
     - name: account_id
       description: "Primary key"
       tests:
         - not_null
 - name: dim_user
   description: "The User Dimension Table"
   columns:
     - name: lead_id
       description: "Primary key"
       tests:
         - not_null
```

## 中間モデル

最終的な後段のモデルのいくつかは同じ計算メトリクスに依存しており、SQLの繰り返しを避けるために、再利用可能な中間モデルを作成することができます。

`int_email_open_clicks_joined`モデルに対するdbt SQLコードです。

```sql
with opens as (
	select * 
	from {{ ref('fct_email_opens') }} 
), clicks as (
	select * 
	from {{ ref('fct_email_clicks') }} 
), opens_clicks_joined as (

    select 
      o.lead_id as lead_id,
      o.campaign_id as campaign_id,
      o.email_send_id as email_send_id,
      o.activity_timestamp as open_ts,
      c.activity_timestamp as click_ts
    from opens as o 
      left join clicks as c 
      on o.email_send_id = c.email_send_id
      and o.lead_id = c.lead_id

)

select * from opens_clicks_joined
```

## マーケティング分析モデル

これらが。マーケティング、セールスチームによって活用されるダッシュボードやレポートを強化するために用いられる最終的なマーケティング分析モデルです。

`country_email_engagement`モデルのdbt SQLコードです。

```sql
with accounts as (
	select 
        account_id,
        billing_country
	from {{ ref('dim_account') }}
), users as (
	select 
        lead_id,
        account_id
	from {{ ref('dim_user') }} 
), opens_clicks_joined as (

    select * from {{ ref('int_email_open_clicks_joined') }} 

), joined as (

	select * 
	from users as u
	left join accounts as a
	on u.account_id = a.account_id
	left join opens_clicks_joined as oc
	on u.lead_id = oc.lead_id

)

select 
	billing_country as country,
	count(open_ts) as opens,
	count(click_ts) as clicks,
	count(click_ts) / count(open_ts) as click_ratio
from joined
group by country
```

# dbtモデルの実行およびテスト

これでモデルの準備が整い、以下を用いてすべてのモデルを実行することができます。

```
$ dbt run
```

次に以下でテストを実行します。

```
$ dbt test
```

# dbtドキュメントのリネージュグラフの参照

![](https://www.databricks.com/wp-content/uploads/2022/07/db-262-blog-img-8.png)
*マーケティング分析モデルのdbtリネージュグラフ*

モデルの実行が成功すると、以下を用いてドキュメントとリネージュグラフを生成することができます。

```
$ dbt docs generate
```

ローカルで以下を実行して参照します。

```
$ dbt docs serve
```

# dbtモデルをプロダクションにデプロイする

ローカルでdbtモデルを開発してテストしたら、プロダクションにデプロイ選択肢は複数あり、そのうちの一つがDatabricksワークフローにおけるdbtタスクタイプ(プライベートプレビュー)です。

あなたのdbtプロジェクトはGitリポジトリ上でバージョン管理されている必要があります。Gitリポジトリをポイントすることで、Databricksワークフローでdbtタスクを作成することができます。
![](https://www.databricks.com/wp-content/uploads/2022/07/db-262-blog-img-9.png)
*dbtをオーケストレートするためにDatabricksワークフローでdbtタスクタイプを使用*

dbtプロジェクトでパッケージを使用しているので、最初のコマンドは最初のタスクとしての**dbt deps**に次いで**dbt run**であり、次のタスクとして**dbt test**を実行することになります。
![](https://www.databricks.com/wp-content/uploads/2022/07/db-262-blog-img-10.png)
*2つの依存関係を持つdbtタスクからなるDatabricksワークフローのジョブ*

そして、run nowを用いて即座にワークフローを実行するか、特定のスケジュールでdbtプロジェクトを実行するためにスケジュールをセットアップすることもできます。
![](https://www.databricks.com/wp-content/uploads/2022/07/db-262-blog-img-11.png)
*それぞれのdbt実行に対するdbtログの参照*

それぞれの処理実行ごとにdbtコマンドごとのログを参照できるので、デバッグや問題の修正に役立ちます。

# Fivetranとdbtを用いてマーケティング分析を強化する

上で説明したように、FivetranとdbtをDatabrikcsレイクハウスプラットフォームと組み合わせることで、セットアップ、管理が容易で、任意のデータモデリング要件に対応できる柔軟性を持つパワフルなマーケティング分析ソリューションを容易に構築できるようになります。

自分のソリューションの構築を始めるには、Databricksと[Fivetran](https://docs.databricks.com/integrations/ingestion/fivetran.html)、[dbt](https://docs.databricks.com/integrations/prep/dbt.html)とのインテグレーションに関するドキュメントを参照いただき、クイックにスタートできるように[marketing\_analytics\_demo project example](https://github.com/databricks/workflows-examples/tree/main/marketing_analytics_demo)を再利用してください。

Databricksワークフローにおけるdbtタスクタイプはプライベートプレビューです。dbtタスクタイプを試したい場合には、Databricks担当者にコンタクトしてください

### Databricks 無料トライアル

[Databricks 無料トライアル](https://databricks.com/jp/try-databricks)