自己紹介
こんにちは。データエンジニアのヨンドンです。
お酒とSQLが大好物です。
データエンジニアリング専門なので分析はしません。
データ成熟度カーストで表すと私が関わっている仕事は下記の部分です。下級データ民です。
背景
今までは主に物販業における売上データや在庫データの分析をするためのデータウェアハウス構築とか、Web解析のログデータ分析基盤構築を主にやってきました。
ところが今回は、今まで触った事がない「人事データ」を分析するためのデータウェアハウスを Snowflake で構築する事になりました。
人事データウェアハウスなんてそんなにしょっちゅう作る事はないので、どんな事をやったのかメモ代わりに書き記します。
この記事で説明するツールは以下の2つです。
-
Snowflake ... データストレージとして使ってます。
-
dbt ... データの変換・モデリングで使ってます。
Snowflake に入れたデータを dbt で変換しながら ディメンショナルモデル を構築するという ELT を用いました。
ディメンショナルモデルとは
Wikipedia にはこう書いてあります。(原文を拙訳)
ディメンショナルモデリングは、ラルフ・キンボール氏が開発したビジネス次元ライフサイクル手法の一部で、データウェアハウス設計に使用する一連の手法、技術、概念を含んでいます。この手法は、ビジネス内の主要なビジネスプロセスを特定し、ボトムアップアプローチとしてモデリングし、実装することに重点を置いています。
平たく言うと、ディメンショナルモデルとは、いわゆるスタースキーマ型の非正規化モデルを採用したデータウェアハウスの事です。
中心にビジネスデータとなるファクトテーブルがあり、その周りに各データを説明するためのディメンションテーブルを星のような型に配置したモデルです。
ディメンショナルモデルの事をスタースキーマと呼ぶ人もいるので、もはやディメンショナルモデル＝スタースキーマです。
このモデルでは「データウェアハウスとはデータマートの集合体であるべき」という考えから、分析アウトプットから逆算してデータウェアハウスをモデリングするというボトムアップ型のアプローチを使います。
ディメンショナルモデルについては本が出ており、前述のラルフ・キンボールが書いた本が有名です。
英語の本なんですけど、私はこの本を読んでディメンショナルモデルを実践しました。
まさにディメンショナルモデル本の決定版です。
私の中ではこの本を「聖書」と呼んでます。私の中ではな。
何を分析するのか
前述のキンボール本こと The Data Warehouse Toolkit から、 "Chapter 9 - Humran Resources Management" に記載の内容に従いました。
主な項目は下記のものです。
- Employee Profile Change
- Perisodic Headcount Snapshots
- Recursive Employee Hierarchies
これらを日本語に意訳するとこうです。
従業員のプロフィール変更
給与情報や異動、昇進、などなど社員のステータスが変わるたびに、変更理由と変更日時などと共にプロフィール変更をトラッキングします。
これは従業員ディメンションテーブルを作って、さらにそこで変更履歴を残すような設計にします。
毎月の在籍人数
会社が数百人以上の規模になると会社に何人のどんな種類の社員がいて、それが毎月どのように推移するのか見えにくくなります。
というわけで、過去データも含めて、毎月何人の社員がいるのか、月ごとのスナップショットレコードを記録するためのファクトテーブルを作ります。
組織階層
会社には CEO がいれば平社員もいます。
各マネージャー毎に何人の部下がいるのか、そしてマネージャーから各社員まで何人の中間管理職がいるのかという組織構成分析のためのディメンションテーブルを作ります。
なお、他にも社員のパフォーマンス分析やスキルセット分析もやりますが長くなるので割愛。
データソース
人事データは SAP の SuccessFacotrs という人事管理ツールの中に入っていて、まずそれを分析のために取り出す必要がありました。
で、どうやらOData APIを使ってデータを取り出せるんだけど、わざわざAPIを実装してクライアントプログラムを作るのも大変です。
というわけでちょうど社内で使っている Workato というローコード／ノーコードで実装出来る RPA業務自動化サービスを使ってデータ抽出をしました。
ただ今回のテーマは ディメンショナルモデルなので、ここでは Workato そのものについては説明しません。
Workatoで定期的に SuccessFactors からデータをエクスポートして、それを Snowflake に UPSERT (MERGE) するというレシピを作って定期的に実行してます。
データストレージ領域
今回のデータは構造化データばかりなので、ストレージは Snowflake だけで構成してます。
中は4つのデータベース領域に分けてあります。
- RAW ... 加工なしの生データを保存する領域
- STAGING ... 未使用カラムの除外、NULLの除去、フォーマットの統一、表記の統一 などのクレンジングや変換を適用した、分析における最小単位のデータ定義をあてる領域
- WAREHOSE ... ディメンショナルモデル(スタースキーマ) でモデリングしたデータを格納するデータウェアハウス領域
- MARTS ... PowerBIから直接参照するためのデータマート領域。スタースキーマから、分析テーマ別に作った各テーブルを置いてある
SuccessFactors からエクスポートしたデータは未加工のまま RAW データベースに格納されます。
それから dbt を使って、クレンジング・変換を経て STAGING データベースに格納されます。
さらに dbt によって WAREHOUSE へ移されて、最後にこれまた dbt で MART を作ります。
dbt 大活躍です。
dbt でモデリングする
とにかく今回の構成は dbt が鍵です。
dbt のフォルダ構成は主要なものだけ見るとこうなってます。
corporate_dwh
├── macros
│ └── udfs
├── models
│ ├── marts
│ │ └── hr
│ ├── staging
│ │ └── successfactors
│ └── warehouse
│ └── hr
└── snapshots
└── hr
いちばん重要なのは models です。
この下にはそれぞれ staging, warehouse, martsのフォルダを作ってあります。
各フォルダの中には今後他のビジネスドメインにも拡張する予定があるので、ドメイン毎のフォルダを作ってます。
staging だけはデータソース別にフォルダを分けます。
maros フォルダには繰り返し使う SQL をパーツ化して格納します。
その配下の udfs フォルダには ユーザー定義関数(UDF)を格納しておいて、各モデルの実行直前にpre_hookで呼び出します。
snapshots フォルダにはディメンショナルモデルで構築した各ディメンションテーブルの変更履歴をトラックするためのモデルです。
後述しますが、ディメンショナルモデルでいう Slowly Changing Dimesion Type2 (SCD2) を実装するために使います。
ER図
ここからが本題です。
人事データ分析のために下記のようなER図を書きました。
このスタースキーマがつまりディメンショナルモデルです。
- DIM_DATE ... 日付ディメンション
- DIM_EMPLOYEE ... 従業員ディメンション
- DIM_EMP_HIER ... 組織階層ディメンション
- FCT_HEADCOUNT ... ファクトテーブル。毎月の在籍人数を記録した指標テーブル
これらのテーブルについてはこのあと説明します。
ディメンションテーブルの作成
ディメンションテーブルは WAREHOSE 領域に格納します。
dim_date
まずは日付ディメンションの dim_date.
これは dbt_date パッケージで前後12年分作りました。
サンプルも公開されていて便利です。
dim_employee
従業員ディメンションの dim_employee.
SuccessFactors から取り出したオブジェクトをそれぞれ Snowflake の RAW データベースにいれて、そこからクレンジングなどの整理や前処理をしてから STAGING 領域に格納されたテーブルから作ります。
ディメンショナルモデルではサロゲートキーを使うのですが、そのキー生成に dbt_utils パッケージの surrogate_key を使いました。
インクリメンタルで番号を生成しなくて良いので便利です。
dbt 素晴らしい。
-- dim_employee.sql
with stg_employee as (
select
-- ステージングの処理が長すぎるので省略
from {{ ref('stg_employee') }}
)
select
{{
dbt_utils.surrogate_key([
'employee_id','employee_name',
'last_date_worked','cost_center_name',
'dept_name', 'event', 'job_title',
'manager_id', 'pay_grade'
])
}} as employee_key
,employee_id
,employee_name
,first_date_worked
,last_date_worked
,cost_center_name
,dept_name
,event
,event_reason
,job_title
,job_type
,job_grade
,job_level
,manager_id
,pay_grade
,created_date_time
,last_modified_date_time
from stg_employee
dim_employee_hier
組織階層ディメンションの dim_emp_hier.
CEOから各社員までの階層構造を記録するディメンションテーブルです。
SQLでツリー構造を扱うのはメンドクサイです。
ここでは CONNECT BY メソッドを使って CEO を起点として全従業員の社員番号による階層構造を抽出しました。
参考にしたのは↓のページです。
サロゲートキーはこれまた dbt_utils.surrogate_key で生成します。
-- dim_employee_hier.sql
WITH de as (
select * from {{ ref('dim_employee') }}
)
,base as (
select
employee_id
,sys_connect_by_path(employee_id, ' -> ') as emp_hier
,manager_id
,job_title
,created_date_time
,last_modified_date_time
from
de start with job_title = 'CEO' connect by manager_id = prior employee_id
order by
user_id
)
select
{{
dbt_utils.surrogate_key([
'employee_id',
'emp_hier',
'manager_id',
'job_title'
])
}} as emp_hier_key
,employee_id
,emp_hier
,(array_size(split(emp_hier, ' -> ')) - 1) as emp_hier_depth
,created_date_time
,last_modified_date_time
from base
結果こういうテーブルが出来上がります。
CONNECT BYはOracle方言なのですが、これがSnowflakeにも実装されてたので使いました。
Recursive CTE使えば似たような事も可能ですけどね。
Slowly Changing Dimension Type 2 の実装
ディメンションの変更履歴をトラックするための実装です。
dbt の snapshot 機能を使うとなんとも簡単に実装出来てしまいます。
dim_employee に snapshot を設定した scd_dim_employee.sql を作成しました。
-- scd_dim_employee.sql
{% snapshot scd_dim_employee %}
{{
config(
unique_key='employee_key',
strategy='timestamp',
updated_at='last_modified_date_time',
)
}}
select * from {{ ref('dim_employee') }}
{% endsnapshot %}
出来上がったテーブルを見ると、 dbt_valid_from と dbt_valid_to という有効期限を示すタイムスタンプカラムが追加されます。
これでレポート集計期間に応じてこれらのタイムスタンプをチェックすれば、変更履歴を反映したレポートを抽出できます。
同様に dim_employee_hier にも snapshot を使って SCD2 を実装しました。
これだけで変更履歴も追えるなんて、dbt はとても便利です。
ファクトテーブルの作成
ディメンションテーブルが揃ったらファクトテーブルを作ります。
fct_headcount
ファクトテーブルも WAREHOSE 領域に格納します。
毎月月初に集計をするので、毎月1日を snapshot_date として扱います。
dim_date には 2010年1月1日 からの日付情報が入ってます。
そこから month_start_date カラムだけを持ってきます。
そして dim_employee に入っている「 first_date_worked の翌月の1日」を snapshot_date との比較用に使います。
あとは各社員ごとに、毎月の集計日に在籍してるかどうかを比較しながら毎月の在籍人数指標を集計して行きます。
ついでに毎月ごとに在籍月数指標も出してます。
あと、このテーブルは毎月データが追記されていくインクリメンタル型です。
dbt の incremental materialization を使ってます。
SQL の最後の方に書いてありますが、上書きしないために日付をチェックして insert する仕組みになってます。
リトライによってデータがダブる事もありません。
冪等性を担保してる訳です。
dbt は便利ですね。
-- fct_headcount.sql
{{
config(
materialized='incremental'
)
}}
with month as (
select
distinct month_start_date as snapshot_date
from {{ ref('dim_date') }}
)
,base as (
select *
,dateadd(day, 1, last_day(first_date_worked)) as snapshot_date
from {{ ref('dim_employment') }}
)
SELECT
b.employee_key
,b.empployee_id
,m.snapshot_date
,b.first_date_worked
,b.last_date_worked
,CASE
WHEN m.snapshot_date <= COALESCE(b.last_date_worked, current_date()) THEN 1
ELSE 0
END AS count_employees
,datediff(month, b.first_date_worked, coalesce(b.last_date_worked, m.snapshot_date)) as months_worked
FROM month as m
INNER JOIN base as b ON(m.snapshot_date >= b.snapshot_date)
where m.snapshot_date <= current_date()
{% if is_incremental() %}
and m.snapshot_date >= (select max(this.snapshot_date) from {{ this }} as this)
{% endif %}
order by m.snapshot_date, b.first_date_worked
これで毎月の在籍人数を集計したファクトテーブルの完成です。
2010年1月1日以降の過去12年に渡る月別の社員数の推移、そして社員別の在席月数が指標として集計されています。
このファクトテーブルを軸にして、レポート要件に応じて dim_employee と dim_employee_hier を JOIN して分析をして行きます。
さらにやるべき事
だいぶ話が長くなってしまったのでこの先は詳細は書きませんが、以下の事を考える必要があります。
- データの整合性 ... dbt test を使って制約を実装する必要があります。
- データマート ... ファクトテーブルとディメンションテーブルを JOIN して必要なカラムだけ抽出したサブセットテーブルを MARTS 領域に複数作成して行きます。
- データのマスキング ... 人事データは個人情報を含むので、Tag-based Masking Policies を実装します。
- パイプラインの自動化 ... 今後他のデータソースも取り込む予定なので、Airflowでオーケストレーションを実装します。
ま、今回の本題はディメンショナルモデルDWHの構築なので、これらはまた別の機会にでも (ぉぃ
まとめ
dbt を使えばディメンショナルモデルのDWH構築もだいぶシンプルに出来ます。
いやー、 dbt って本当にいいもんですね。