データ分析

マルチステートメントトランザクションを使用した BigQuery における Mercury の発展

2021年7月9日

https://storage.googleapis.com/gweb-cloudblog-publish/images/GettyImages-1315531636_1.max-1700x1700.jpg

Google Cloud Japan Team

※この投稿は米国時間 2021 年 7 月 1 日に、Google Cloud blog に投稿されたものの抄訳です。

ローマ神話における商売の神である Mercury（メルクリウス）は、商取引の象徴である財布を持ち、翼の生えたサンダルを履いて疾走する姿がよく描かれています。運送会社にリアルタイムで追跡されながら世界中を移動する何百万個もの荷物から、個人ローン、証券取引、政府間取引などのグローバルな決済に至るまで、商品やサービスを世界中で流通させるトランザクションが、今日における世界のビジネスシステムを支えています。本日、BigQuery におけるマルチステートメントトランザクションの公開プレビュー版を発表いたします。

BigQuery は長年にわたり、トランザクションあたり 1 つのテーブルに適用される DML ステートメントによる単一ステートメントトランザクション（例: INSERT、UPDATE、DELETE、MERGE、TRUNCATE）をサポートしてきました。マルチステートメントトランザクションは、単一のトランザクションで複数のテーブルにまたがる、DML を含む複数の SQL ステートメントをサポートすることで、その対象範囲を拡大します。つまり、あるトランザクション内のすべてのステートメントに関連する複数のテーブルのデータに対する変更が、成功した場合はアトミックに（すべて一度に）commit され、失敗した場合はすべてアトミックにロールバックされることを意味します。これらの新しいマルチステートメントトランザクション機能は、Mercury がその速さによって人間界と神界をすばやく行き来できたように、オンライントランザクション処理（OLTP）システムと BigQuery をより緊密かつ高速に統合することで、両者の隔たりを埋めることができます。

概要

トランザクションとは、業務単位を表すデータベースやデータウェアハウスに対して実行される一連のタスクのことです。たとえば、注文の迅速化アクションは複数のテーブルに一連の変更をもたらす可能性があります（例: 注文の納期を更新し、迅速な貨物配送を作成するために貨物に新しい出荷ラインを追加し、以前の通常配送の出荷ラインをキャンセル（更新）する）。これに対応して、バッチジョブがデータウェアハウスの ETL 処理を実行する際には、複数のテーブルのデータを変換して保存することで、データラングリングやデータクレンジングの操作を行う場合があります。いずれの場合でも、障害の発生時にはトランザクション全体をロールバックして、記録システムに部分的に保存されたデータによってデータセットの不整合を防ぐ必要があります。

https://storage.googleapis.com/gweb-cloudblog-publish/images/image3_P7Defd5.max-600x600.png

トランザクションの ACID プロパティ

トランザクションは、原子性、整合性、独立性、永続性（ACID）プロパティを持つことで知られています。原子性は、すべてのデータ変更（メタデータの変更を含む）が単一の操作で適用または復帰されることを示します。たとえば、ある一揃いの行が拡張によってテーブルに追加されて対応する一揃いの行が削除されるとき、両方の操作が同時に行われます。整合性は、トランザクションが完了する前後におけるデータの整合性を指します。データをテーブルに追加またはテーブルで更新されるときに、テーブルのデータへの変更がテーブルまたはその列に指定された制約と一致していることを示しています。たとえば、データがテーブルに追加されるかテーブルで変更される場合、NOT NULL 制約がコンパイルされます。独立性は、複数のトランザクションが互いに干渉しない限り、テーブルに対して同時に変更を加えられることを示します。たとえば、異なる業務単位からテーブルにデータをロードする 2 つのトランザクションは同時に進行できます。最後の永続性は、おそらく最も重要なトランザクションのプロパティで、障害の発生時（トランザクションを完了した後）に、トランザクションによって commit されたすべての変更が、将来のトランザクションやクエリのために保存されることを保証します。

分離レベル

マルチステートメントトランザクションを含む BigQuery のすべてのトランザクションは、スナップショット分離をサポートします。この分離レベルでは、トランザクション内のすべてのステートメントが、トランザクションの開始時点でのデータベースの一貫したスナップショットを参照します。BigQuery ではこの機能を、一定期間のデータベースへのミューテーション履歴を保持するマルチバージョンのプロパティによって実現します。進行中のトランザクションは、他の同時実行中のトランザクションからの変更（commit、uncommit を問わず）を参照しません。BigQuery は書込み内容の読み込みをサポートしており、トランザクション内のステートメントは、同じトランザクション内の以前のステートメントによって適用されたすべての変更を参照できます。

例

BigQuery が 1 度の操作で複数のテーブルに対して複数の変更を加えるマルチステートメントトランザクションのコンセプトをどのようにサポートするかを示す例を見てみましょう。この例では、100 個のパーティションを持つ 10 個のテーブルを作成することで、このデモを設定します。

lang-sql

読み込んでいます...

DECLARE num_tables DEFAULT 10;
DECLARE x INT64 DEFAULT 0;
# replace "demo" with your preferred dataset name
DECLARE ds STRING default "`00jathreya`";
DECLARE prefix STRING DEFAULT "mst_demo";
DECLARE table_name STRING;
DECLARE pre_table_name STRING;

/*
Sets up 10 tables, with schema (partition_id:integer,value:timestamp).
Each table has 100 partitions, with partition key 0, 10, 20, ..., 990.
*/

SET x = 0;
WHILE x < num_tables DO
 SET table_name = format("%s.%s_%d", ds, prefix, x); 
 EXECUTE IMMEDIATE format("""DROP TABLE IF EXISTS %s""", table_name); 
 EXECUTE IMMEDIATE format("""
 CREATE OR REPLACE TABLE %s (partition_id INT64, value TIMESTAMP)
 PARTITION BY RANGE_BUCKET(partition_id, GENERATE_ARRAY(0, 1000, 10))
 AS SELECT partition_id, CURRENT_TIMESTAMP() AS value FROM UNNEST(GENERATE_ARRAY(0, 990, 10)) AS partition_id
 """, table_name);
 SET x = x + 1;
END WHILE;
# Verify the partitioned tables created and the timestamp
execute immediate format("""select table_schema, table_name, count(partition_id), sum(total_rows), max(last_modified_time) from %s.INFORMATION_SCHEMA.PARTITIONS
where starts_with(table_name,'%s')
group by table_schema, table_name""", ds, prefix);
# Verify data in table 9 based on the timestamp
EXECUTE IMMEDIATE format("""
SELECT * FROM %s ORDER BY partition_id
""", table_name);

次に、BigQuery の新しい COMMIT コマンドと ROLLBACK コマンドによってトランザクションを横断する複数テーブルの変更を保存または逆行する方法を示します。

lang-sql

読み込んでいます...

DECLARE num_tables DEFAULT 10;
DECLARE x INT64 DEFAULT 0;
# Replace `mydataset` with your preferred dataset name
DECLARE ds STRING DEFAULT "`mydataset`";
DECLARE prefix STRING DEFAULT "mst_demo";
DECLARE table_name STRING;
DECLARE pre_table_name STRING;
 
/*
Steps of the transaction:
1. For table 0, updates the timestamp to CURRENT_TIMESTAMP.
2. FOR table 1 to 9, merge table x using data from table x-1. All MERGE should
see the updated data in table x-1, so as a result the change in table 0 should be
propagated all the way to table 9.
3. A total of 1000 partition changes on 10 tables.
*/
BEGIN TRANSACTION;
 
SET x = 0;
WHILE x < num_tables DO
 SET table_name = format("%s.%s_%d", ds, prefix, x);
 
 IF x = 0 THEN
   EXECUTE IMMEDIATE format("""
   UPDATE %s SET value = CURRENT_TIMESTAMP WHERE true
   """, table_name);
 ELSE
   SET pre_table_name = format("%s.%s_%d", ds, prefix, x-1);
   EXECUTE IMMEDIATE format("""
   MERGE %s t2 USING %s t1 ON t2.partition_id = t1.partition_id
   WHEN MATCHED THEN UPDATE SET value = t1.value
   """, table_name, pre_table_name);
 END IF;
 
 SET x = x + 1;
END WHILE;
 
# Verify data in table 9 inside transaction shows recently updated data
SET table_name = format("%s.%s_%d", ds, prefix, num_tables-1);
EXECUTE IMMEDIATE format("""
SELECT * FROM %s ORDER BY partition_id
""", table_name);
 
IF x = 9 THEN
 COMMIT TRANSACTION;
ELSE
 ROLLBACK TRANSACTION;
END IF;
 
# Verify data in table 9 outside transaction shows timestamp of original data
EXECUTE IMMEDIATE format("""
SELECT * FROM %s ORDER BY partition_id
""", table_name);

モニタリングおよび管理

単一の DML コマンドにマッピングされる単一ステートメントトランザクションとは異なり、マルチステートメントトランザクションは、複数テーブルを横断する複数の DML オペレーションにまたがる複雑なものです。これらの複雑なジョブを監視するために、BigQuery JOBS INFORMATION_SCHEMA ビューが拡張され、BigQuery の各トランザクションの一意の識別子を含む新しい列 TRANSACTION_ID が追加されました。これらのビューを照会すると、進行中のトランザクションもしくは完了したトランザクションを照会できます。

lang-sql

読み込んでいます...

#Returns the list of transactions running under a parent job ID .;
SELECT
DISTINCT transaction_id
FROM `region-us`.INFORMATION_SCHEMA.JOBS_BY_PROJECT
WHERE parent_job_id = "job_id" AND state = "RUNNING";
 
# Returns the active transactions and parent job id that are running and affect a named table.;
WITH 
  running_transactions AS (
  SELECT DISTINCT transaction_id 
  FROM `region-us`.INFORMATION_SCHEMA.JOBS_BY_PROJECT 
  WHERE state = "RUNNING")
SELECT 
  transaction_id, parent_job_id, query 
FROM `region-us`.INFORMATION_SCHEMA.JOBS_BY_PROJECT, running_transactions
WHERE destination_table = "mytable" 
  AND transaction_id = running_transactions.transaction_id;
 
# Returns all successfully committed transactions
SELECT 
  transaction_id, parent_job_id, query 
FROM `region-us`.INFORMATION_SCHEMA.JOBS_BY_PROJECT 
WHERE statement_type = "COMMIT_TRANSACTION" AND error_result IS NULL;
 
# Returns all rolled back transactions
SELECT 
  transaction_id, parent_job_id, query 
FROM `region-us`.INFORMATION_SCHEMA.JOBS_BY_PROJECT
WHERE statement_type = "ROLLBACK_TRANSACTION" AND error_result IS NULL;

トランザクションを終了させるには、トランザクションを含むスクリプトの親ジョブ ID を BQ.JOBS.CANCEL システムプロシージャに渡すことで、トランザクションが実行されている親ジョブを終了させる必要があります。

まとめ

企業がよりリアルタイムな知見に基づく意思決定を加速させる中、マルチステートメントトランザクションは、トランザクションデータを BigQuery などの分析データウェアハウスに取り込むための重要な実現手段となります。強力な ACID プロパティ、commit とロールバックの完全なトランザクションサポート、実績のあるトランザクション処理のバックボーン上に構築された豊富なモニタリング API および管理 API により、BigQuery をご利用のお客様がこれらの新機能を通じてトランザクションの価値を引き出すお手伝いをします。

-Google Cloud プロダクトマネージャー Jagan R. Athreya

-Google Cloud シニアスタッフソフトウェアエンジニア Pavan Edara

投稿先