こんにちは！クラシルリワードで開発責任者をしているfunzinです。 はじめにクラシルリワードのサービス概要について紹介させてください。
クラシルリワードは「日常のお買い物体験をお得に変える」アプリです。 買い物のためにお店に行く（移動する）、チラシを見る、商品を買う、レシートを受け取る......。これら日常の行動がポイントに変わり、そのポイントを使って様々な特典と交換することができます。
気になる方はサービスサイトをぜひご確認ください。 それではさっそく本題のデータ基盤の構成について説明していきます。 クラシルリワードのデータ基盤の技術スタックは下記になります。 全体像のみではそれぞれの役割が把握しにくいため、次の章より詳細を説明していきます。 クラシルリワードではメインのデータ基盤としてBigQueryを利用しています。
BigQueryを選定した理由としては下記となります。 サービスの立ち上げ当初、新規事業ということもあり最小人数で開発をしていたためデータエンジニアが不在でした。
しかし、施策結果を振り返るためにイベントログ実装は必須であり、データ基盤を構築する必要がありました。 クラシルリワードは 現在iOS / Androidアプリでサービスを提供しています。 iOSでのイベント送信例 インフラはAWSで構築しているためAWS Athena、クラシルで利用しているSnowflakeなども検討しましたが、Google Analytics for Firebaseを利用するメリットが大きいためBigQueryを採用しました。 この章ではクラシルリワードのデータ基盤において、重要な役割を抜粋して説明していきます。 BigQueryの選定理由でも説明しましたが、 アプリからイベントログを送信するためにGoogle Analytics for Firebaseを利用しています。 例えば起動イベントのみを抽出する場合、日付別テーブルを利用するとwhere句で条件を指定してもその範囲の日付全てのイベントが対象となり、パーティション分割テーブルと比較してscan量が多くなってしまい、結果的にコストに影響します。 日々分析をする上でのscan量の課題を解決するために、下記のような工夫をしています。 BigQueryにエクスポートされた日付別テーブルを直接利用するのではなく、パーティション分割テーブルとして変換して分析ではこちらを利用するようにしています。
下記がパーティション分割テーブルを作成するサンプルクエリです。 パーティション: Date パーティション分割テーブル作成後はクエリスケジューリングを利用し、毎日定期実行で指定範囲の日数を上書きするような形でイベントログをアップデートしています。 2. new_event_nameの導入 Google Analytics for Firebaseの仕様上、イベント名は最大500までしか定義できません。
カジュアルに新規イベントを定義しすぎると上限の500に到達してしまい、新規イベントの計測ができなくなります。 例えば下記のようなイベント名とIDの場合、 クラシルリワードではインフラ環境をAWSで構築しているため、ユーザーのデータはRDSやDynamoDBに保存されています。 多くのテーブルは全体更新をしていますが、一部テーブルはレコード数が多くなっているため差分更新をしています。
差分更新の対応については説明が長くなってしまうため、また別のテックブログで話せればと思います。 運用面に関しては、現状大きな課題はないもののバッチ処理が失敗した時の調査やリトライなどの運用工数もかかり始めてはきているので、自前のETLフローではなくtroccoやFivetranなどのSaaS ETLツールも検討しています。 メインのデータ基盤はBigQueryですが、一部でSnowflakeを利用しています。
Snowflakeを利用することになった背景は下記2つです。 現状dely社では各事業部ごとに技術選定をしています。
クラシルではAWS Athena, Snowflakeを利用してデータ分析をしています。クラシルリワードでは今まで説明してきましたがBigQueryをメインで利用しています。 2. 一部のクラシルのテーブルをBigQueryにエクスポートしたいため クラシルリワードではクラシルチラシにまつわる情報もユーザーに提供していますが、チラシ情報のテーブルがクラシル側にあるため、BigQueryで分析を完結することができません。 例えば、クラシルリワード側でどの店舗のチラシを見られたかを分析したい場合、どうしてもBigQueryに店舗のマスターデータが必要になってきます。 そのためクラシルチラシのデータ(e.g. 店舗情報)をクラシルリワード側のBigQueryに格納するために下記のようなフローを構築することで実現しています。 クラシルでのデータをBigQueryにエクスポートすることで、クラシルリワード側ではBigQueryのみで完結してチラシ情報を分析することが可能となっています。 最後にBIツールについて説明します。 現状はセルフホスティングでRedash用のEC2インスタンスを立てて、SQLクエリを書いて分析をすることが多いです。 フォルダ構造 csv CSVではクエリ名、データソースID, クエリIDのみを最低限管理しています。 PRがマージされたタイミングでGitHub Actionsを実行し差分があるクエリのみをRedashに反映させるようにしています。 GitHub Actions ステップ実行順 現状は差分があるクエリのみをRedashに反映するように影響範囲をとどめています。 RedashではSQLクエリを書くというハードルが残り続けるため、SQLクエリをかけない非エンジニアでも分析が可能な体制にするため、現在は並行してLookerにも移行中です。 クラシルリワードのデータ基盤について紹介しました。
データエンジニアが不在の中でも、分析基盤を構築しサービス改善につなげることができました。
引き続きデータを分析することでクラシルリワードを改善していきたいと思いますので、よろしくお願いします。
はじめに
この記事ではクラシルリワードのデータ基盤の構成について紹介していきます。クラシルリワードについて
データ基盤の全体構成
BigQueryの選定理由
また自社データ基盤の場合、イベントログのスキーマ設計、エラー時の対応、データパイプラインの修正対応などに時間がかかることが懸念としてありました。
そのためフルマネージドなBigQueryを採用することで、データ基盤の構築にかかる時間を短縮しサービス開発にリソースを割くことができると判断しました。
2. Google Analytics for Firebaseとの相性が良い
モバイルアプリではFirebaseを利用することが一般的であり、クラシルリワードでもFirebaseを積極的に利用しています。
Google Analytics for Firebaseを利用することで下記のメリットがあります。
event_name, event_paramsなど最低限のものだけを意識すれば良い// event_name: test_event
// event_params: ["index": 0]
Analytics.logEvent("test_event", parameters: ["index": 0])
データ基盤における重要な役割
アプリのイベントログのscan量削減
日付別テーブルがBigQueryにエクスポートされるため、これだけでもイベントログ分析は可能になります。
しかし、そのまま日付別テーブルを利用する場合、DAUが増えるにつれて、BigQueryのscan量が増加しコストに影響してきます。
CREATE OR REPLACE TABLE
`product_firebase_analytics.events`
PARTITION BY
DATE(event_timestamp_micros)
CLUSTER BY
event_name,
new_event_name,
platform AS
SELECT
event_date,
TIMESTAMP_MICROS(event_timestamp) AS event_timestamp_micros,
ARRAY_TO_STRING(
[event_name,
(
SELECT value.string_value
FROM UNNEST(event_params)
WHERE KEY = 'id'
)],
"_"
) AS new_event_name,
platform,
... -- カラムが続く
FROM
`project.events_*`
クラスタ: event_name, new_event_name(後述), platform-- 直近3日以内のイベントログを上書きする
DECLARE start_date_interval INT64 DEFAULT 3;
DECLARE end_date_interval INT64 DEFAULT 0;
DELETE FROM
`product_firebase_analytics.events`
WHERE
DATE(event_time, 'Asia/Tokyo') BETWEEN DATE_SUB(
CURRENT_DATE('Asia/Tokyo'),
INTERVAL start_date_interval DAY
)
AND DATE_SUB(
CURRENT_DATE('Asia/Tokyo'),
INTERVAL end_date_interval DAY
);
INSERT INTO
`product_firebase_analytics.events`
SELECT
event_date,
TIMESTAMP_MICROS(event_timestamp) AS event_timestamp_micros,
ARRAY_TO_STRING(
[event_name,
(
SELECT value.string_value
FROM UNNEST(event_params)
WHERE KEY = 'id'
)],
"_"
) AS new_event_name,
platform,
...-- カラムが続く
FROM
`project.events_*`
WHERE
REGEXP_EXTRACT(_TABLE_SUFFIX, r '[0-9]+') BETWEEN FORMAT_DATE(
"%Y%m%d",
DATE_SUB(
CURRENT_DATE('Asia/Tokyo'),
INTERVAL start_date_interval DAY
)
)
AND FORMAT_DATE(
"%Y%m%d",
DATE_SUB(
CURRENT_DATE('Asia/Tokyo'),
INTERVAL end_date_interval DAY
)
)
イベント数の上限が500を超えないために、イベント名を一意にするのではなく、event_paramsにイベントを識別できるid(STRING)を含めて、パーティション分割テーブル作成時にnew_event_nameとして定義しています。new_event_nameはtest_hogeになります
new_event_nameはパーティション分割テーブル作成時にevent_nameとidを結合することで、新しいイベント名として定義することで実現しています。 ARRAY_TO_STRING(
[event_name,
(
SELECT value.string_value
FROM UNNEST(event_params)
WHERE KEY = 'id'
)],
"_"
) AS new_event_name
new_event_nameによって、下記のようなメリットが得られます。
event_nameはtest, imp, tapなど汎用的なイベントのみを定義して使い回すため、イベント名の定義数で500を超えることがなくなるnew_event_nameを指定しているため、分析する側はnew_event_nameを条件に指定することでscan量を削減することが可能DynamoDBやRDSをBigQueryにSync
これらのデータをアプリから送信したイベントログと結合してBigQueryで分析をしたいニーズがでてきます。
現在はDigdagとEmbulkを利用して、1日1回定期実行でBigQueryにSyncをしています。
これらはAWS ECSのタスクスケジューリングで実現しています。Snowflakeの活用
しかし、事業部が分かれているため、クラシル側のデータを分析するときに何か問題が発生するとクラシルのデータエンジニアに都度依頼する運用コストが発生していました。
この課題を解消するために、クラシルリワード側でSnowflakeをアカウントを作成し、Snowflakeのデータシェアリングを利用することで、クラシルリワード開発部でクラシルのデータを自由に分析することができるようにしています。
Snowflake(クラシル) -> Snowflake(クラシルリワード) -> Google Storage(クラシルリワード) -> BQ(クラシルリワード)
BIツール(Redash, Looker)
Redash
しかしSQLクエリは人によって書き方が異なる、数値出しの判断が間違える可能性があるため、SQLクエリをGitHubのRepositoryで管理してレビューをする体制をとっています。
こちらはバイセルさんのRedashのクエリ管理方法を参考にしています。Redash repositoryのフォルダ・ファイル構成
├── README.md
├── csv // クエリ名などをcsv管理
├── queries // クエリ管理
├── requirements.txt
└── scripts // Redashに反映用のスクリプト
name,data_source_id,query_id
query_test_name,1,29
git管理下のSQLクエリをRedashに反映
write_diff_query_only.sh)main.py)write_diff_query_only.sh#!/bin/bash
# write_diff_query_only.sh
# PR差分のみをcsv_query_list.csvに反映するスクリプト
header=`head -n 1 csv/query_list.csv`
current_commit=$GITHUB_SHA
previous_commit=$(git rev-parse "${current_commit}"\^1)
query_ids=`git diff --name-only "$current_commit" "$previous_commit" | grep queries | grep -o '[0-9]\+' | tr '\n' '|'`
query_ids=${query_ids%|}
echo "$header" > csv/diff_query_list.csv
if [ ! -z "$query_ids" ]; then
grep -E "($query_ids)$" csv/query_list.csv >> csv/diff_query_list.csv
fi
# main.pyで差分のみを上書きして実行するため
cat csv/diff_query_list.csv > csv/query_list.csv
main.py# main.py
import os.path
from redash_client.client import RedashClient
import csv
import os
class Query:
def __init__(self, base_path, name, data_source_id, query_id, schedule=None, description='generated by retail-redash-query'):
self.query_id = query_id
self.name = name
self.data_source_id = data_source_id
self.schedule = schedule
self.description = description
path = 'queries/{0}.sql'.format(query_id)
self.load_query(base_path, path)
def load_query(self, base_path, path):
with open('{0}/{1}'.format(base_path, path), 'r') as f:
self.query = f.read()
# Read query list
def get_queries():
reader = csv.reader(open('csv/query_list.csv', 'r'))
queries = []
# Skip header
header = next(reader)
for row in reader:
queries.append(Query("./", row[0], row[1], row[2]))
return queries
RedashClient.BASE_URL = os.environ['REDASH_HOST']
RedashClient.API_BASE_URL = os.environ['REDASH_HOST'] + "/api/"
api_key = os.environ['REDASH_API_KEY']
redash_client = RedashClient(api_key)
# Update queries
for query in get_queries():
print('QueryID: ' + query.query_id)
try:
redash_client.update_query(
query.query_id, query.name, query.query, query.data_source_id, query.description
)
except RedashClient.RedashClientException:
print('Error: ' + query.query_id)
Lookerについて
さいごに
クラシルリワードのデータ基盤
