📅

BigQuery で日本の祝日データを活用する

に公開

こんにちは、Sally 株式会社 CTO の @aitaro です。普段はマーダーミステリーアプリ「ウズ」とマダミス制作ツール「ウズスタジオ」、マダミス情報サイト「マダミス.jp」を開発しています。

はじめに

データ分析をする際、日本の祝日データが必要になる場面が多いのではないでしょうか。例えば、売上分析で「平日と休日で売上がどう変わるか」を調べたい場合や、ユーザー行動の分析で「祝日の影響を除外したい」、事業の KPI を平日/祝日別に設定したいといったケースです。

弊社では BigQuery でデータ分析基盤を構築しております。その時に参照できる祝日データセットがあると色々と便利だったので、BigQuery にメンテされている祝日マスターテーブルを用意することにしました。

今回は、内閣府が公開している祝日データを BigQuery に取り込む方法をまとめてみました。実際に弊社で運用しているスクリプトをベースに、実装の詳細から運用まで紹介していきます。

日本の祝日データについて

日本の祝日データは、内閣府が CSV 形式で公開しています。このデータは「国民の祝日について」のページから取得でき、毎年更新されています。

https://www8.cao.go.jp/chosei/shukujitsu/gaiyou.html

祝日データを取得する方法は 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 を取得してパースする処理は次のようになります。

update_japan_holidays.py
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 で格納する形式に変換します。

update_japan_holidays.py
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 ステートメントを使っています。これで既存データを保持しながら新しい祝日だけを追加できます。

update_japan_holidays.py
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 算出にも使えます。

祝日情報は地味ですがデータ分析の基礎となるマスターデータです。一度セットアップしておけば色々な分析で使えるので、ぜひ試してみてください。

UZU テックブログ

Discussion