データ移動不要！Cloud SQL のマスタデータと BigQuery のトランザクションデータを結合して SQL で分析してみた。
こんにちは、データアナリティクス事業本部のみかみです。
本エントリは、クラスメソッド BigQuery Advent Calendar 2020 の 18 日目のエントリです。 25日のアドベントカレンダー終了まで、弊社クラスメソッド データアナリティクス事業本部のメンバーで、Google BigQuery に関する記事を紡いでいこうと思います。
やりたいこと
- BigQuery と Cloud SQL のテーブルデータを SQL で結合して集計したい
ある小売業の会社のシステムでは、商品マスタや店舗マスタなどのマスタデータは、システム管理画面でメンテナンスする必要があるため、リレーショナルデータベースの Cloud SQL で管理しています。 一方、実店舗での売上データは、いつかデータ分析で活用するために、デイリーバッチ処理で BigQuery にロードしています。
さて、そろそろ BigQuery にため込んだデータを活用し、売れ筋商品や売り上げの傾向などを分析したいと思っているのですが、 売上データでは販売店舗や販売商品がコードで記載されているため、人間が見て意味が分かる結果を取得するためには、BigQuery の売上データと Cloud SQL で管理しているマスタデータを結合する必要があります。
さてどうしましょう？ Cloud SQL からテーブルデータを出力して BigQuery にロードするバッチを実装する？ それとも Dataflow を使って、Cloud SQL から BigQuery にデータをロードするデータパイプラインを構築？
BigQuery なら、データを移動することなく、BigQuery からそのまま Cloud SQL のテーブルデータを SQL で参照できるんです！
動作確認用データを準備
Cloud SQL で管理している商品と店舗のマスタデータと、BigQuery に保存している売上データを作成します。 実在の商品名、店舗名を使用させていただいていますが、動作確認用の架空のデータです。
下記サイトからいただいてきた「鬼滅の刃コラボ商品」のデータの一部をもとに、商品マスタを作成しました。
また、沖縄のスーパーの名前をお借りして、店舗コードを乱数で付与し、店舗マスタを作成しました。
Cloud SQL に MySQL 5.7 のインスタンスを建て、作成したデータをロードして商品マスタテーブル（ mst_products ）と、店舗マスタテーブル（ mst_shops ）を作成しました。
それぞれのテーブルデータはこんな感じです。
mysql> select * from mst_shops order by shop_cd; +---------+--------------------------+ | shop_cd | shop_name | +---------+--------------------------+ | 779 | Jimmys | | 941 | ナカハラストアー | | 2640 | 丸大 | | 5394 | Aコープ | | 6427 | サンエー | | 6532 | ユニオン | | 7746 | りうぼう | | 9066 | かねひで | | 9824 | イオン琉球 | +---------+--------------------------+ 9 rows in set (0.08 sec)
mysql> select * from mst_prod order by jancd limit 10; +---------------+-----------------------------------------------------------------------------+--------------------------------------+-----------------------------------+--------+ | jancd | name_prod | name_bender | category | price | +---------------+-----------------------------------------------------------------------------+--------------------------------------+-----------------------------------+--------+ | 4514062284934 | ＵＨＡピピン ｅ－ｍａのど飴鬼滅の刃 袋 ４０ｇ | ユーハピピン | キャンディ・キャラメル | 150.64 | | 4549660503620 | バンダイ 鬼滅の刃ディフォルメシールウエハース１枚 | バンダイ | 玩具菓子 | 98.67 | | 4549660503637 | バンダイ 鬼滅の刃ウエハース２ １枚 | バンダイ | 玩具菓子 | 116.92 | | 4549660504306 | バンダイ 鬼滅の刃ＣＡＮＤＹ缶２ １９ｇ | バンダイ | キャンディ・キャラメル | 381.33 | | 4549660504757 | バンダイ 鬼滅の刃名場面チョコスナック ２０ｇ | バンダイ | 玩具菓子 | 177.23 | | 4549660504764 | バンダイ 鬼滅の刃禰豆子のチョコバー １個 | バンダイ | 玩具菓子 | 158.46 | | 4549660504771 | バンダイ 鬼滅の刃全員集中アクリルチャーム １個 | バンダイ | 玩具菓子 | 477.98 | | 4580036210100 | フォルテ 鬼滅の刃コレクターズＣ １枚 | フォルテ | 玩具菓子 | 116.83 | | 4580036210162 | フォルテ 鬼滅の刃コレクターズＣ２ １枚 | フォルテ | 玩具菓子 | 118.3 | | 4582138604269 | エフトイズ 鬼滅の刃日輪刀コレクション １粒 | エフトイズ・コンフェクト | 玩具菓子 | 498 | +---------------+-----------------------------------------------------------------------------+--------------------------------------+-----------------------------------+--------+ 10 rows in set (0.07 sec)
さらに、以下の Python コードを準備して、商品マスタと店舗マスタファイルから「店舗コード」「商品コード」「販売個数」「販売日時」の項目を持つ 10 日分の売上データをランダムで作成しました。
import pandas as pd
import random
import time
import datetime
def get_jancds():
path = './advent/mst_prod.txt'
df = pd.read_csv(path)
return df['商品コード']
def get_shopcds():
path = './advent/mst_shop.txt'
df = pd.read_csv(path, dtype = 'object')
return df['店舗コード']
def random_date(start, end, format, prop):
stime = time.mktime(time.strptime(start, format))
etime = time.mktime(time.strptime(end, format))
ptime = stime + prop * (etime - stime)
return time.strftime('%Y-%m-%dT%H:%M:%S%z', time.localtime(ptime))
dt_start = '2020-12-01 00:00:00'
day_cnt = 10
df = pd.DataFrame()
for i in range(day_cnt):
row_cnt = random.randint(1, 20)
start = (datetime.datetime.strptime(dt_start, '%Y-%m-%d %H:%M:%S')
+ datetime.timedelta(days=i)).strftime('%Y-%m-%d %H:%M:%S')
end = (datetime.datetime.strptime(start, '%Y-%m-%d %H:%M:%S')
+ datetime.timedelta(hours=23, minutes=59, seconds=59)).strftime('%Y-%m-%d %H:%M:%S')
df_day = pd.DataFrame({
'shopcd':[random.choice(get_shopcds()) for _ in range(row_cnt)],
'jancd':[random.choice(get_jancds()) for _ in range(row_cnt)],
'count':[random.randint(1, 5) for _ in range(row_cnt)],
'pos_timestamp':[random_date(start, end, '%Y-%m-%d %H:%M:%S', random.random()) for _ in range(row_cnt)],
})
df = df.append(df_day, ignore_index=True)
df = df[['shopcd', "jancd", 'count', 'pos_timestamp']]
df.to_csv('./advent/pos.csv', index=False)
作成した売上データを BigQuery のテーブルに格納します。
BigQuery に作成した pos_kimetsu テーブルには、こんな感じの売上データが格納できました。
Cloud SQL への外部接続を設定
BigQuery の外部リソースとして、Cloud SQL への接続を設定します。
BigQuery 管理画面の「データを追加」リンクから、「外部データソース」を追加します。
外部データソース入力画面で、「接続タイプ」に「Cloud SQL- MySQL」を選択し、「接続 ID」に任意の名称を入力。「接続ロケーション」は「東京（asia-northeast1）」を選択しました。
「Cloud SQL インスタンス ID」には project-id:location-id:instance-id の形式で Cloud SQL のインスタンス ID を指定しますが、この ID は Cloud SQL 管理画面からコピーすることができます。
接続する Cloud SQL の「データベース名」と「ユーザー名」「パスワード」を入力して「接続を作成」ボタンをクリックすれば完了です。
リソース一覧に Cloud SQL への「外部接続」が追加されました。 「接続 ID」は、SQL を実行する時に指定する必要があるのでコピーしておきます。
本当に Cloud SQL のデータが参照できるようになったかどうか、SQL を実行して確認してみます。
無事、Cloud SQL のテーブルデータが BigQuery から参照できるようになりました。
なお、今回は、BigQuery、Cloud SQL ともに管理権限があるユーザーで確認しています。 別ユーザーで実行する場合は権限付与が必要になる場合もあるのでご注意ください。
また、公式ドキュメントにトラブルシューティングの記載もあるので、もしうまくいかない場合は合わせてご確認ください。
BigQuery テーブルと Cloud SQL のテーブルを JOIN して集計
BigQuery から Cloud SQL のテーブルを参照するには、SQL で EXTERNAL_QUERY を使います。
EXTERNAL_QUERY の引数に、BigQuery 外部接続の「接続 ID」と SQL 文を渡すことで、Cloud SQL のテーブルデータを参照できます。
では、BigQuery の売上テーブルと Cloud SQL の商品マスタテーブルを結合して、各日付ごとに最も多く売れた商品と個数、売上げを確認してみます。 実行する SQL は以下です。
WITH target AS (
SELECT
pos.pos_date,
prod.name_prod,
category,
SUM(pos.count) AS count,
FLOOR(SUM(pos.count)*(SUM(prod.price)/COUNT(*))) AS sales
FROM (
SELECT
jancd,
count,
FORMAT_TIMESTAMP("%Y-%m-%d", pos_timestamp, "Asia/Tokyo") AS pos_date
FROM
dataset_advent_calendar.pos_kimetsu
) pos
INNER JOIN EXTERNAL_QUERY(
'cm-da-mikami-yuki-258308.asia-northeast1.con-advent-calendar',
'SELECT jancd, name_prod, category, price FROM mst_prod'
) prod ON pos.jancd = prod.jancd
GROUP BY
pos_date, name_prod, category
)
SELECT
pos_date,
name_prod,
category,
count,
sales
FROM (
SELECT
pos_date,
name_prod,
category,
count,
sales,
ROW_NUMBER() OVER(PARTITION BY pos_date ORDER BY count DESC) row_num
FROM
target
)
WHERE
row_num = 1
ORDER BY
pos_date
なお、使用しているデータは動作確認のために作成したダミーデータで、実際の店舗や売上とは全く関係ありません。
結果が取得できました！
先ほどは売上個数別に見てみましたが、売上別で見るとどんな商品が売れているのでしょうか？
WITH target AS (
SELECT
pos.pos_date,
prod.name_prod,
category,
SUM(pos.count) AS count,
FLOOR(SUM(pos.count)*(SUM(prod.price)/COUNT(*))) AS sales
FROM (
SELECT
jancd,
count,
FORMAT_TIMESTAMP("%Y-%m-%d", pos_timestamp, "Asia/Tokyo") AS pos_date
FROM
dataset_advent_calendar.pos_kimetsu
) pos
INNER JOIN EXTERNAL_QUERY(
'cm-da-mikami-yuki-258308.asia-northeast1.con-advent-calendar',
'SELECT jancd, name_prod, category, price FROM mst_prod'
) prod ON pos.jancd = prod.jancd
GROUP BY
pos_date, name_prod, category
)
SELECT
pos_date,
name_prod,
category,
count,
sales
FROM (
SELECT
pos_date,
name_prod,
category,
count,
sales,
ROW_NUMBER() OVER(PARTITION BY pos_date ORDER BY sales DESC) row_num
FROM
target
)
WHERE
row_num = 1
ORDER BY
pos_date
販売個数で見た時はガムやウエハースなどのお菓子がよく売れていましたが、売上高で見るとカップ麺が好調なことが分かりました。
注：参照しているのは動作確認のために勝手に作成したダミーデータで、実際の売上とは全く関係ありませんｍｍ
さらに、Cloud SQL の店舗マスタテーブルも一緒に結合して、各日付ごとに最も売り上げの多かった店舗で一番売れた商品を確認してみます。 以下の SQL を実行しました。
WITH target AS (
SELECT
pos.pos_date,
shop.shop_name,
prod.name_prod,
SUM(pos.count) AS count,
FLOOR(SUM(pos.count)*(SUM(prod.price)/COUNT(*))) AS sales
FROM (
SELECT
shopcd,
jancd,
count,
FORMAT_TIMESTAMP("%Y-%m-%d", pos_timestamp, "Asia/Tokyo") AS pos_date
FROM
dataset_advent_calendar.pos_kimetsu
) pos
INNER JOIN EXTERNAL_QUERY(
'cm-da-mikami-yuki-258308.asia-northeast1.con-advent-calendar',
'SELECT jancd, name_prod, price FROM mst_prod'
) prod ON pos.jancd = prod.jancd
INNER JOIN EXTERNAL_QUERY(
'cm-da-mikami-yuki-258308.asia-northeast1.con-advent-calendar',
'SELECT shop_cd, shop_name FROM mst_shops'
) shop ON pos.shopcd = shop.shop_cd
GROUP BY
pos_date, shop_name, name_prod
)
SELECT
pos_date,
shop_name,
name_prod,
sales
FROM (
SELECT
pos_date,
shop_name,
name_prod,
sales,
ROW_NUMBER() OVER(PARTITION BY pos_date ORDER BY sales DESC) row_num
FROM
target
)
WHERE
row_num = 1
ORDER BY
pos_date
Cloud SQL のテーブルが複数でも、問題なく結合できました。
ただ、やはり Cloud SQL からのデータ取得にはそれなりに時間がかかります。 クエリプランで確認してみると、BigQuery テーブルの 103 行を 84 msで読み込んでいるのに対して、Cloud SQL の 9 行の店舗マスタテーブルを読み込むのに 316 ms かかっていました。
定期的な SQL 実行等パフォーマンスが気になる場合は、CREATE TABLE AS などで BigQuery に実テーブルを作成してしまった方がよさそうです。
まとめ（所感）
思っていたより簡単に、Cloud SQL のデータを BigQuery から参照することができました。 BigQuery から Cloud SQL のデータを参照したい時にやることは、以下の 2 つだけです。（設定やデータ参照に関する権限は問題ないことが前提です）
- Cloud SQL への接続設定（初回のみ）
- SQL で
EXTERNAL_QUERY関数を使う
接続設定はテーブル単位ではなくデータベースに対して行うので、GCS や Google ドライブのデータを参照するときのように、ファイルごとにテーブルを定義する必要はありません。 また、接続設定で入力する ID などの入力値はテキストボックスにプレースホルダーが表示されるので、どんなフォーマットで入力すべきなのか迷うこともありませんでした。
画面や Web API などを提供するシステムの DB には、トランザクションがサポートされている Cloud SQL が利用されることが多いのではないかと思います。 一方、データ分析は BigQuery で行うケースがほとんどで、分析に利用したいデータが Cloud SQL にある場合もあるのではないかと思います。 BigQuery 管理画面の UI 操作だけで簡単に接続設定できて、また接続設定さえしておけば BigQuery からシームレスに Cloud SQL のテーブルが参照できるこの機能は、SQL を使ってシステムデータを分析したいユーザーにとって非常にありがたいのではないかと思いました。 ぜひ一度お試しあれ！ｖ
明日 19 日目の BigQuery Advent Calendar 2020 は甲木からお送り予定です。 また、今後のアドベントカレンダーの予定はこちらにも掲載しておりますので、引き続きお楽しみいただけますと幸いです。
