BigQueryで繰り返し処理を実行して、日付を一括更新する方法

目次

• はじめに
• バックアップを取得
• 元のデータと内容を比較
• fuga_atの日付を一括更新
• まとめ

はじめに

ファインディ株式会社、データソリューションチームの山家(@yamayafumiteru)です。
実務でBigQueryを使用しており、日付を含むテーブル（例: dataset.hoge_yyyymmdd）のデータに対して特定の処理（yyyymmddとfuga_atを同じ日付にする)をしたいケースに遭遇しました。更新するために調べていた所、BigQueryのコンソールでは、IF文やWHILE文を利用することができることがわかりました。
それらとCREATE TABLE等を併用することで作成できるか試して見ようと思います。

※Pythonを使用して更新する方法もありますが、この記事ではPythonの使用は含まれていません。

バックアップを取得

tableの再作成を始める前に、重要なデータを保護するためにバックアップを取ることが重要です。

誤って削除してしまったり、上書きしてしまった際に取り返しがつかないことになりかねません。
一応、BigQueryにはデータのバージョン管理とリカバリ機能が組み込まれており、誤ってデータを削除してしまっても7日以内であれば復元することができます。(今回とは、別のタイミングでこの機能にとても助けられました。。。）

消して復元できるからといっても、複数のtableを戻すのは手間なので、バックアップを取っておくに越したことはないと思います。
以下は、hoge.fuga.piyo_yyyymmddというデータに対して処理を行う場合の処理になります。

# 変数の宣言
DECLARE date_array ARRAY < STRING >;
DECLARE x INT64 DEFAULT 1;
DECLARE create_table_name STRING;

# 変数への代入
SET
  date_array = (
    SELECT
      ARRAY_AGG(
        REGEXP_EXTRACT(table_name, r'(\d{8})')
        ORDER BY
          table_name
      ) as table_names
    FROM
      `hoge.fuga.INFORMATION_SCHEMA.TABLES`
    WHERE
      table_catalog = 'hoge' -- プロジェクト名
      AND table_schema = 'fuga' -- データセット名
      AND REGEXP_CONTAINS(table_name, r'^piyo_[0-9]{8}$') -- テーブル名に含まれる文字列
  );

WHILE x <= array_length(date_array) DO
  SET
  create_table_name = CONCAT(
    '`hoge.fuga.tmp_piyo_',
    date_array [ORDINAL(x)],
    '`'
  );
  EXECUTE IMMEDIATE (
    CONCAT(
      'CREATE OR REPLACE TABLE ',
      create_table_name,
      ' AS (',
      'SELECT ',
      '  * ',
      'FROM ',
      '  `hoge.fuga.piyo_*` ',
      'WHERE ',
      '  _TABLE_SUFFIX = "',
      date_array [ORDINAL(x)],
      '"',
      ')'
    )
  );
  SET
    x = x + 1;

END WHILE;

これらを実行することでhoge.fuga.piyo_yyyymmddがhoge.fuga.tmp_piyo_yyyymmddに作成されます。

元のデータと内容を比較

EXCEPT DISTINCTを使用してpiyoとtmp_piyoが同じ内容であることを確認する。

SELECT count(*) AS count_diff
FROM (
  SELECT *
  FROM `hoge.fuga.piyo_yyyymmdd`
  EXCEPT DISTINCT
  SELECT *
  FROM `hoge.fuga.tmp_piyo_yyyymmdd`
) AS diff

fuga_atの日付を一括更新

今回、hoge.fuga.piyo_yyyymmddのyyyymmddとfuga_atが違う場合、yyyymmddの日付に合わせる処理を実行しようと思います。
先程作成したtmp_piyo_yyyymmddをもとにcreate_table_nameに対してfuga_atを修正した結果をREPLACEします。

# 変数の宣言
DECLARE date_array ARRAY < STRING >;
DECLARE x INT64 DEFAULT 1;
DECLARE create_table_name STRING;

# 変数への代入
SET
  date_array = (
    SELECT
      ARRAY_AGG(
        REGEXP_EXTRACT(table_name, r'(\d{8})')
        ORDER BY
          table_name
        LIMIT
          5
      ) as table_names
    FROM
      `hoge.fuga.INFORMATION_SCHEMA.TABLES`
    WHERE
      table_catalog = 'hoge' -- プロジェクト名
      AND table_schema = 'fuga' -- データセット名
      AND REGEXP_CONTAINS(table_name, r'^piyo_[0-9]{8}$') -- テーブル名に含まれる文字列
  );

WHILE x <= array_length(date_array) DO
  SET
  create_table_name = CONCAT(
    '`hoge.fuga.piyo_yyyymmdd',
    date_array [ORDINAL(x)],
    '`'
  );
  IF EXISTS (
    SELECT
      *
    FROM
      `hoge.fuga.piyo_*`
    WHERE
      _TABLE_SUFFIX = date_array [ORDINAL(x)]
      AND PARSE_DATE("%Y%m%d", date_array [ORDINAL(x)]) != fuga_at
  ) THEN EXECUTE IMMEDIATE (
    CONCAT(
      'CREATE OR REPLACE TABLE ',
      create_table_name,
      ' AS (',
      'SELECT ',
      'PARSE_DATE("%Y%m%d", "',
      date_array [ORDINAL(x)],
      '") AS fuga_at ', -- ここのfuga_at以外のカラム名を記述する
      'FROM ',
      '`hoge.fuga.tmp_piyo_*` ',
      'WHERE ',
      '_TABLE_SUFFIX = "',
      date_array [ORDINAL(x)],
      '"',
      ');'
    )
  );
  END IF;

  SET
    x = x + 1;

END WHILE;

まとめ

普段はPythonとBigQueryを使用することが多いのですが、BigQueryでもWHILEやIFを使用できることが知れてよかったです。
特定のtableに対して一括で更新をかけたいけど、Pythonの実行環境がなくてBigQueryで実行したいというニーズがある方のご参考になれば幸いです。