BigQuery で日本の祝日データを活用する
こんにちは、Sally 株式会社 CTO の @aitaro です。普段はマーダーミステリーアプリ「ウズ」とマダミス制作ツール「ウズスタジオ」、マダミス情報サイト「マダミス.jp」を開発しています。
はじめに
データ分析をする際、日本の祝日データが必要になる場面が多いのではないでしょうか。例えば、売上分析で「平日と休日で売上がどう変わるか」を調べたい場合や、ユーザー行動の分析で「祝日の影響を除外したい」、事業の KPI を平日/祝日別に設定したいといったケースです。
弊社では BigQuery でデータ分析基盤を構築しております。その時に参照できる祝日データセットがあると色々と便利だったので、BigQuery にメンテされている祝日マスターテーブルを用意することにしました。
今回は、内閣府が公開している祝日データを BigQuery に取り込む方法をまとめてみました。実際に弊社で運用しているスクリプトをベースに、実装の詳細から運用まで紹介していきます。
日本の祝日データについて
日本の祝日データは、内閣府が CSV 形式で公開しています。このデータは「国民の祝日について」のページから取得でき、毎年更新されています。
祝日データを取得する方法は Google Calendar API や Python の祝日パッケージなどいくつかありますが、弊社では内閣府の公式 CSV を使っています。
内閣府のデータであれば政府の公式情報源なので信頼性が高く、CSV 形式で公開されているため追加での API 認証も不要で、シンプルな HTTP リクエストで取得できます。
CSV の形式
公開されている CSV の形式は以下のようになっています。
国民の祝日・休日月日,国民の祝日・休日名称
1955/1/1,元日
1955/1/15,成人の日
1955/3/21,春分の日
...
このデータは、Shift-JIS エンコーディングで提供されてます。また、日付形式は YYYY/M/D でゼロパディングがないので、そのままでは BigQuery に入れられません。データを加工して追加します。
BigQuery テーブルの設計
まず、祝日データを格納する BigQuery テーブルのスキーマを設計します。弊社では以下のような構造で運用しています。
date の他に year, month, day など、AI のおすすめで追加しておきましたが、date さえあればデータ分析には十分です。
CREATE TABLE `project.dataset.japan_holidays` (
date DATE NOT NULL, -- 祝日の日付
name_ja STRING NOT NULL, -- 祝日名(日本語)
year INT64 NOT NULL, -- 年
month INT64 NOT NULL, -- 月
day INT64 NOT NULL, -- 日
day_of_week STRING, -- 曜日(英語)
updated_at TIMESTAMP NOT NULL -- 最終更新日時
)
CLUSTER BY date
OPTIONS(
description="Japanese national holidays from Cabinet Office"
);
Python スクリプトの実装
次に祝日データを取得して BigQuery に更新するスクリプトをいくつかの関数に分けて実装します。
今回は、Google API のライブラリサポートが充実しており、依存関係の管理が不要でシングルファイルで動く Python を選定しました。
祝日データの取得
内閣府の CSV を取得してパースする処理は次のようになります。
import csv
import io
import requests
from datetime import datetime
HOLIDAYS_CSV_URL = "https://www8.cao.go.jp/chosei/shukujitsu/syukujitsu.csv"
def fetch_holidays_from_csv() -> List[Dict[str, str]]:
"""
内閣府のCSVから祝日データを取得する
"""
response = requests.get(HOLIDAYS_CSV_URL, timeout=30)
response.raise_for_status()
# Shift-JISからUTF-8にデコード
content = response.content.decode('shift-jis')
# CSVをパース
csv_reader = csv.DictReader(io.StringIO(content))
holidays = []
for row in csv_reader:
date_str = row['国民の祝日・休日月日']
name_ja = row['国民の祝日・休日名称']
# 日付形式をYYYY/M/D から YYYY-MM-DD に変換
date_obj = datetime.strptime(date_str, "%Y/%m/%d")
formatted_date = date_obj.strftime("%Y-%m-%d")
holidays.append({
'date': formatted_date,
'name_ja': name_ja,
'date_obj': date_obj
})
return holidays
ここでハマったのは、内閣府の CSV が Shift-JIS だったことです(官公庁であること考えればむしろ当然ですが)。また、CSV の日付形式は YYYY/M/D でなぜかゼロパディングがないので、BigQuery の DATE 型に合わせて YYYY-MM-DD 形式に変換しています。
データの構造化
取得した祝日データを先ほど示した BigQuery で格納する形式に変換します。
DAYS_OF_WEEK = {
0: "Monday", 1: "Tuesday", 2: "Wednesday",
3: "Thursday", 4: "Friday", 5: "Saturday", 6: "Sunday"
}
def parse_holidays_data(holidays_list: List[Dict]) -> List[Dict]:
"""
祝日リストをBigQuery用の構造に変換する
"""
records = []
current_timestamp = datetime.now(timezone.utc).isoformat()
for holiday in holidays_list:
date_obj = holiday['date_obj']
record = {
"date": holiday['date'],
"name_ja": holiday['name_ja'],
"year": date_obj.year,
"month": date_obj.month,
"day": date_obj.day,
"day_of_week": DAYS_OF_WEEK.get(date_obj.weekday()),
"updated_at": current_timestamp
}
records.append(record)
return records
MERGE による差分更新
BigQuery への更新では MERGE ステートメントを使っています。これで既存データを保持しながら新しい祝日だけを追加できます。
def update_bigquery_table(
project_id: str,
dataset_id: str,
table_id: str,
records: List[Dict],
) -> None:
"""
MERGEを使用してBigQueryテーブルを更新する
"""
client = bigquery.Client(project=project_id)
table_ref = f"{project_id}.{dataset_id}.{table_id}"
temp_table_ref = f"{table_ref}_temp"
# DataFrameに変換
df = pd.DataFrame(records)
df['date'] = pd.to_datetime(df['date']).dt.date
df['updated_at'] = pd.to_datetime(df['updated_at'])
# 一時テーブルにロード
schema = [
bigquery.SchemaField("date", "DATE", mode="REQUIRED"),
bigquery.SchemaField("name_ja", "STRING", mode="REQUIRED"),
bigquery.SchemaField("year", "INTEGER", mode="REQUIRED"),
bigquery.SchemaField("month", "INTEGER", mode="REQUIRED"),
bigquery.SchemaField("day", "INTEGER", mode="REQUIRED"),
bigquery.SchemaField("day_of_week", "STRING", mode="NULLABLE"),
bigquery.SchemaField("updated_at", "TIMESTAMP", mode="REQUIRED"),
]
job_config = bigquery.LoadJobConfig(
schema=schema,
write_disposition=bigquery.WriteDisposition.WRITE_TRUNCATE,
)
temp_job = client.load_table_from_dataframe(
df, temp_table_ref, job_config=job_config
)
temp_job.result()
# MERGEで本テーブルを更新
merge_query = f"""
MERGE `{table_ref}` AS target
USING `{temp_table_ref}` AS source
ON target.date = source.date
WHEN MATCHED AND (
target.name_ja != source.name_ja OR
target.year != source.year OR
target.month != source.month OR
target.day != source.day OR
target.day_of_week != source.day_of_week
) THEN
UPDATE SET
name_ja = source.name_ja,
year = source.year,
month = source.month,
day = source.day,
day_of_week = source.day_of_week,
updated_at = source.updated_at
WHEN NOT MATCHED THEN
INSERT (date, name_ja, year, month, day, day_of_week, updated_at)
VALUES (source.date, source.name_ja, source.year, source.month,
source.day, source.day_of_week, source.updated_at)
"""
merge_job = client.query(merge_query)
merge_job.result()
# 一時テーブルをクリーンアップ
client.delete_table(temp_table_ref, not_found_ok=True)
WRITE_TRUNCATE でテーブル全体を置き換えるのではなく、MERGE で差分だけを更新しています。祝日名の変更があった場合も適切に反映されますし、何度実行しても同じ結果になり、冪等性があるので、スケジュール実行でも安心です。
スクリプトの実行方法
dev, stg, prd と各環境でスクリプトを実行したいので、project-id を引数で渡せるようにしています。 table-id 等は各環境で等しいので、引数で渡せる必要はないのですが、AI が柔軟性高く実装してくれたので、そのまま利用しています。
# 必要なパッケージをインストール
pip install google-cloud-bigquery pandas requests
# スクリプトを実行
python update_japan_holidays.py \
--project-id your-project-id \
--dataset-id your_dataset \
--table-id japan_holidays
スクリプトの実行には、環境変数 GOOGLE_APPLICATION_CREDENTIALS を設定し、gcloud の認証を通す必要があります。
定期実行の設定
祝日データは年に数回更新される可能性があるため、定期的に実行するのが推奨です。弊社では Cloud Scheduler と Cloud Run Jobs を組み合わせて月 1 回の定期実行を設定しています。本筋ではないので、定期実行の設定の仕方は割愛します。
活用例
実際のデータ分析の活用として、平日・休日の売上比較が挙げられます。
売り上げデータテーブルと祝日テーブルを join して分析します。
WITH daily_sales AS (
SELECT
DATE(order_timestamp) as order_date,
SUM(amount) as total_sales
FROM `project.dataset.orders`
GROUP BY order_date
)
SELECT
CASE
WHEN h.date IS NOT NULL THEN '祝日'
WHEN EXTRACT(DAYOFWEEK FROM ds.order_date) IN (1, 7) THEN '週末'
ELSE '平日'
END as day_type,
AVG(ds.total_sales) as avg_sales,
COUNT(*) as days_count
FROM daily_sales ds
LEFT JOIN `project.dataset.japan_holidays` h
ON ds.order_date = h.date
WHERE ds.order_date >= '2024-01-01'
GROUP BY day_type
まとめ
内閣府が公開している祝日データを BigQuery に取り込んで、データ分析で活用する方法を紹介しました。
この仕組みを入れることで、平日・休日での売上やユーザー行動の違いを分析できるようになります。連休の影響を考慮した需要予測や、営業日ベースでの KPI 算出にも使えます。
祝日情報は地味ですがデータ分析の基礎となるマスターデータです。一度セットアップしておけば色々な分析で使えるので、ぜひ試してみてください。
Discussion