この記事について
noteの方で、「境界線が見えていなかった」という気づき軸の話を書きました。あちらでは技術の細部にはあえて踏み込みませんでしたが、こちらでは実際にどう設計し、どう実装したかを、コードと判断理由込みで残しておきます。
構成としては、まずこの①設計編で「何を作ったか」を淡々と解説し、続く②トラブルシューティング編で「作る過程で何が起きたか」を扱います。どちらから読んでも成立するように書いているので、興味のある方から読んでもらえればと思います。
きっかけと前提
Visual Studioサブスクリプションに、実はAzureの使用権がついていました。せっかく権利があるならと、実務とは切り離した実験環境として組んでみることにしたのが今回のパイプラインです。
実務データはもちろん使えないので、疑似的な人事データセット(600名分・5テーブル)をAIに生成してもらい、それを題材にしました。着手前のスキルセットは、Azure完全未経験、Power BIは触った程度、DB設計はほぼ未経験、という状態でした。
全体構成
Blob Storageへのフラットなcsv出力処理は、SQL Database連携を追加した後もそのまま残しています。同じデータに対して2つの出力先がある構成にしておくと、比較や切り分けがしやすくなるためです。
DB設計
テーブル構成
正規化した5テーブルで構成しています。
| テーブル名 | 役割 | 主キー |
|---|---|---|
department_master | 部門マスタ(親) | department_id |
employee_master | 社員マスタ | employee_id |
evaluation_history | 評価履歴 | employee_id + fiscal_year(複合) |
attendance_summary | 勤怠サマリ | employee_id + year_month(複合) |
attrition | 離職情報 | employee_id |
_master・_history・_summaryと接尾辞で役割を分けているのは、テーブル名を見ただけで「これはマスタ(ディメンション)なのか、時系列で積み上がるファクトなのか」が分かるようにするためです。attritionのように、実質1:1の関係でありながらイベント記録としての性質も持つテーブルは、どちらに寄せるか少し悩みましたが、今回は独立テーブルとして切り出しています。
評価履歴と勤怠サマリは、社員×年度、社員×年月という複合主キーにしています。同一社員の複数年度・複数月のレコードを一意に特定するには、単一の自動採番IDよりもこちらの方が業務的な意味が素直に表現できると考えたためです。
文字コードの落とし穴
日本語を含むカラムはVARCHARではなくNVARCHARで定義する必要があります。VARCHARのままだと、日本語データが文字化けします。SQL ServerではVARCHARが非Unicode文字列型のため、日本語のようなマルチバイト文字を正しく扱うにはNVARCHAR(Unicode文字列型)を使う必要がある、という基本的な仕様なのですが、着手前は意識したことがありませんでした。
FK制約と操作順序
外部キー制約を貼っているため、テーブルの作成順序(親→子)と、データ削除時の順序(子→親)がそれぞれ逆向きに制約されます。この「順序を意識しないといけない」という感覚は、後述するFunction App側の書き込みロジックにもそのまま効いてきます。
Function App実装
pymssqlを選んだ理由
Azure SQL Databaseへの接続ライブラリとしては、pyodbcが定番です。ただし今回、Function AppはFlex従量課金プラン(Linux)を選択していたのですが、このプランはroot権限でのカスタム操作ができない制約があり、pyodbcが要求するODBC Driverを追加インストールできませんでした。
そこで、ODBC Driverに依存せず単体で動作するpymssqlを採用しました。プラン選択とライブラリ選択が独立した判断ではなく、片方が決まるともう片方の選択肢が絞られる、という依存関係になっていたのが実装前には見えていなかった点です。
書き込みロジック:洗い替え方式
def _load_to_sql_database(df_department, df_employee, df_evaluation, df_attendance, df_attrition):
conn = _get_sql_connection()
try:
cursor = conn.cursor()
# 既存データをクリア(FK制約があるため、子テーブル→親テーブルの順)
for table in ["evaluation_history", "attendance_summary", "attrition",
"employee_master", "department_master"]:
cursor.execute(f"DELETE FROM {table}")
# 親テーブルから順に投入
_write_df_to_sql(cursor, "department_master", df_department)
_write_df_to_sql(cursor, "employee_master", df_employee)
_write_df_to_sql(cursor, "evaluation_history", df_evaluation)
_write_df_to_sql(cursor, "attendance_summary", df_attendance)
_write_df_to_sql(cursor, "attrition", df_attrition)
conn.commit()
finally:
conn.close()実行のたびに既存データをDELETEしてから入れ直す、いわゆる「洗い替え(full refresh)」方式にしています。差分更新ではなくこの方式にしたのは、実行の冪等性(何度実行しても同じ結果になること)を確保するためです。差分更新はロジックが複雑になりやすく、今回のような検証目的のパイプラインでは、洗い替えの方がシンプルで事故が少ないと判断しました。
削除の順序が子テーブル→親テーブル、投入の順序が親テーブル→子テーブルと、DB設計の章で触れたFK制約の向きがそのままコードの構造に表れています。
②へ続く
ここまでが「何を作ったか」の設計部分です。実際に動かす過程では、この設計だけでは防ぎきれないトラブルにいくつも遭遇しました。接続まわりのハマりどころ、pandasのバージョン差、トランザクション設計が実際に効いた場面、Power BIのリレーション設計など、②トラブルシューティング編でまとめて扱います。なお、ここで触れたコードの全体はGitHubで公開しています。
コメント