ココネBI室の紹介＆Cloud Composer導入

2023年6月20日

お疲れ様です。
開発本部BI室でデータエンジニアを担当しているHです。

1. 目次

BI室の紹介
KPI環境の紹介
データの流れ
Composer導入
最後に

2. BI室の紹介

BI室では、ココネにおけるすべてのデータを管理・集約して、すべてのココネメンバーに適切なデータや情報を提供できる環境を整えるべく努めています。

目指すべきBIメンバー像：
意思決定者にとって信頼される「参謀」へ

BIメンバーとして意識すること：
①「プロアクティブ」な動き
②「事業へ資する」動き
③「効率を高める」動き

「やること」と「やらないこと」：

やること	やらないこと
積極的なコニュニケーション（情報収集）	各方面からの分析依頼待ち
アナリスト同士のナレッジ共有	属人的なナレッジの蓄積
事業に資するかどうかの見極め	“単なる興味”に基づいた分析
ネクストアクションにつながる分析	ファクトのみの掲示
分析業務の効率化、周囲への協力依頼	「時間をかけて、なんとかする」

という方針のもと、
データアナリスト：5名 / データエンジニア：2名が働いています。

データエンジニアの主な役割としては

ETL環境の構築・保守管理
ETL実装
KPIページ作成

でしたが、直近のミッションでETL実装/KPIページ作成をデータアナリスト自身が行える環境づくりを目指すこととなり、非エンジニアでも簡単にETL実装ができるような環境を構築中です。

3. KPI環境の紹介

3-1. DWH

現在はBigQueryを使っています。私が入社した2014年当時はMSSQLでした。MongoDBのデータをMSSQLにETL処理し、MariaDB BI用SlaveとOPENQUERYを用いてJOINし集計を行っていました。
アプリタイトルが増えたことにより分析に要する各種データのデータサイズが増大し、DWHリプレースを検討することとなりました。具体的には、2019/03/28から検討・検証し、2019/06/06にBigQuery導入が決定し、アプリタイトル毎に順次移行し2019/07/04からBigQuery運用開始となり、2020/11/25にMSSQLからBigQueryへ移行完了しました。

3-2. ETL

現在はCloud Composer/Dataflowを使ってELTを行っています。ETLクラウド化する前はSSIS(SQL Server Integration Services)/SQL Server Agentを使ってETL処理を行っていました。
ETLクラウド化は2021/12からCloud Composer/Dataflow検討・検証・実装し、アプリタイトル毎に順次移行し2022/05から運用開始となり、2023/02/14にSSISからCloud Composer/Dataflowへ移行完了しました。

3-3. BIツール

現在はSSRS(SQL Server Reporting Services)を使ってKPIページを表示しています。BIツールのリプレースを検討中で、Tableau/Looker/Amazon QuickSightのPoCを行っています。

4. データの流れ

上記のように「各事業部が管理するBI用DB・共通システムが管理するBI用DB」から「分析やKPIページ表示に必要なデータだけ」をETLしてBigQueryに格納、KPIページ/Looker Studio/Connected SheetsからBigQueryを参照してKPIレポートを事業部側へ提供しています。

原則として1日1回の深夜バッチで前日分のデータをETL処理しています。従って、BigQueryにあるデータは前日分までのデータとなります。例外的に一部の事業部では4時間間隔で当日分のデータのETLを行いレポートしています。

5. Composer導入

これまで約１年ほどComposerを運用してきて
①ETL高速化
②ETLエラー時のAirflow UIでのスムーズな対応
③BI用DBへ接続安定化（ネットワーク設定）
の3点で劇的に改善できたと実感しています。

今回は①を簡単に紹介します。

タスク間の依存関係を設定することで無駄な待ち時間を作らずに処理でき、かつタスクの依存関係を可視化することができました。タスク間だけでなく異なるDAG間の依存関係もExternalTaskSensorを用いて設定しました。

具体的には、導入前後で以下のように変わりました。

▼タスクの依存関係

具体例：
A,B,C,Dの処理があり、
Dの処理がB/Cが成功したことを条件として開始する。

■導入前
SQL Server Agent ジョブにステップとしてSSISパッケージを登録していました。
依存関係の設定はなく、ステップに基づき順々に実行となっていました。

■導入後
タスク名で依存関係を指定しました。

▼異なるDAG間の依存関係

■導入前
SQL Server Agent ジョブ間の依存関係は「sp_start_job」を用いて設定していました。
※AirflowでいうところのTriggerDagRunOperatorに相当

■導入後
異なるDAG間の依存関係は「ExternalTaskSensor」を用いて設定しました。
導入後は複数のDAGを待てるようになりました。

waiting_for_DAG_x = ExternalTaskSensor(
	task_id='waiting_for_DAG_x',
	external_dag_id='DAG_x',
	execution_date_fn=lambda x: x + timedelta(hours=12),
	timeout=10800,
	allowed_states=['success'],
	mode='poke',
	dag=dag
)