はじめに
BigQueryのデータのバックアップ・復元については、いくつかやり方があります。
参考↓
その中でも、今回はタイムトラベル機能を使用してデータの復元をするための手順・コードを共有させて頂きます。
BigQuery タイムトラベル機能
BigQueryには標準でタイムトラベルという機能が存在します。過去7日間であれば、任意のタイミングのデータを復元可能であるため、その機能を利用してBigQueryのデータを復元してみたいと思います。
なお、Google Cloudのコンソール画面からは削除されたテーブルを復元することはできず、bqコマンドラインツールかクライアントライブラリ（Python等）を使用する必要があります。今回はPythonのクライアントライブラリを使用して削除されたテーブルの復元をするためのコードを紹介します。
コンソール画面から削除されたテーブルを復元することはできませんが、存在するテーブルを過去の特定の時間のデータに戻すことはSQLを使用することで対応可能です。
サンプルプログラム
前提
今回はGoogle Colaboratoryで実行する想定としています。
また、BigQueryを操作するためのサービスアカウントは作成済みであり、JSONキーファイルがCloud Storageに格納されているものとします。
コード
コード全体は以下を展開してご確認ください。
コード全体
import time
import os
from datetime import datetime, timedelta
from google.cloud import bigquery
from google.colab import auth
from google.cloud import storage
auth.authenticate_user()
def calculate_snapshot_epoch(days_ago=0, hours_ago=0):
target_time = datetime.now() - timedelta(days=days_ago, hours=hours_ago)
return int(target_time.timestamp() * 1000)
# 認証情報の設定
# GCSクライアントを作成
client = storage.Client()
# 認証情報のJSONファイルをGCSからダウンロード
bucket_name = 'BUCKET_NAME' # ここにCloud Storageのバケット名を入力
blob_name = 'JSON_KEY_FILE' # ここにサービスアカウントのJSONキーファイルのパスを入力
destination_file_name = 'DOWNLOAD_FILE_NAME'
bucket = client.bucket(bucket_name)
blob = bucket.blob(blob_name)
blob.download_to_filename(destination_file_name)
# 環境変数に認証情報のパスを設定
os.environ['GOOGLE_APPLICATION_CREDENTIALS'] = destination_file_name
# テーブルの復元を実行
client = bigquery.Client()
# TODO: 復元させたいテーブルを指定する
table_id = "BIGQUERY_TABLE"
# TODO: 復元後のテーブル名を指定する
# 同じテーブル名で復旧させる場合、table_idと同じ文字列を入力する
recovered_table_id = "BIGQUERY_TABLE"
# TODO: どれくらい前の状態を復元させるか指定する
# Example: 2 days and 3 hours ago(days_ago=2, hours_ago=3)
snapshot_epoch = calculate_snapshot_epoch(days_ago=1, hours_ago=6)
# 必要に応じてテーブル削除してから復旧させる
# すでにテーブルが存在しない場合はコメントアウトすること
client.delete_table(table_id)
snapshot_table_id = "{}@{}".format(table_id, snapshot_epoch)
job = client.copy_table(
snapshot_table_id,
recovered_table_id,
location="asia-northeast1", # 適切なリージョンを指定する
)
job.result()
print(
"Copied data from deleted table {} to {}".format(table_id, recovered_table_id)
)
大きく以下の2つの処理となるので、それぞれ解説をします。
認証情報の設定
# 認証情報の設定
# GCSクライアントを作成
client = storage.Client()
# 認証情報のJSONファイルをGCSからダウンロード
bucket_name = 'BUCKET_NAME' # ここにCloud Storageのバケット名を入力
blob_name = 'JSON_KEY_FILE' # ここにサービスアカウントのJSONキーファイルのパスを入力
destination_file_name = 'DOWNLOAD_FILE_NAME'
bucket = client.bucket(bucket_name)
blob = bucket.blob(blob_name)
blob.download_to_filename(destination_file_name)
# 環境変数に認証情報のパスを設定
os.environ['GOOGLE_APPLICATION_CREDENTIALS'] = destination_file_name
Cloud Storageに格納しているサービスアカウントのJSONキーファイルにアクセスし、認証情報を設定しています。
バケット名やJSONキーファイル名は実際のものに置き換えてください。
Google Colaboratoryを実行するユーザーは、Cloud Storageのバケットにアクセス可能な権限を持っている前提です。
テーブルの復元
基本的にはこちらのサンプルスクリプトを参考にしています。
def calculate_snapshot_epoch(days_ago=0, hours_ago=0):
target_time = datetime.now() - timedelta(days=days_ago, hours=hours_ago)
return int(target_time.timestamp() * 1000)
# テーブルの復元を実行
client = bigquery.Client()
# TODO: 復元させたいテーブルを指定する
table_id = "BIGQUERY_TABLE"
# TODO: 復元後のテーブル名を指定する
# 同じテーブル名で復旧させる場合、table_idと同じ文字列を入力する
recovered_table_id = "BIGQUERY_TABLE"
# TODO: どれくらい前の状態を復元させるか指定する
# Example: 2 days and 3 hours ago(days_ago=2, hours_ago=3)
snapshot_epoch = calculate_snapshot_epoch(days_ago=1, hours_ago=6)
# 必要に応じてテーブル削除してから復旧させる
# すでにテーブルが存在しない場合はコメントアウトすること
client.delete_table(table_id)
snapshot_table_id = "{}@{}".format(table_id, snapshot_epoch)
job = client.copy_table(
snapshot_table_id,
recovered_table_id,
location="asia-northeast1", # 適切なリージョンを指定する
)
job.result()
print(
"Copied data from deleted table {} to {}".format(table_id, recovered_table_id)
)
プログラム実行前に修正する変数は以下の通りです。
- table_id
復元したいテーブルのテーブルIDを設定してください。 - recovered_table_id
復元後のテーブルのテーブルIDを設定してください。同じ名前で復元する場合には、table_idと同じ内容を設定します。
以下の1行でいつ時点のデータを復元するかを設定します。days_agoに遡る日数を、hours_agoに遡る時間を指定します。（以下の場合は1日と6時間前（30時間前）のデータに復元されます。）
snapshot_epoch = calculate_snapshot_epoch(days_ago=1, hours_ago=6)
以下の1行については、「テーブルは存在するが、過去のとある時間に戻したい」場合を想定して書いたものです。すでに削除されたテーブルを復旧させる場合にはコメントアウトして下さい。
client.delete_table(table_id)
以下のlocation引数は東京リージョンを想定して記述していますが、必要に応じて適切なリージョンを設定ください。
job = client.copy_table(
snapshot_table_id,
recovered_table_id,
location="asia-northeast1", # 適切なリージョンを指定する
)
解説は以上です。
まとめ
今回はBigQueryのタイムトラベル機能を使用したデータの復元を、Pythonを使用して実施してみました。
プログラムを書く手間はありますが、このようなプログラムを用意しておくといざ障害など発生した時に助かりますね。何か皆様のお役に立つことがあれば大変嬉しく思います。
ご覧頂きありがとうございました。