G-gen の杉村です。BigQuery の継続的クエリ（Continuous queries）機能を使うと、事前定義した SQL ステートメントが継続的に実行され、リアルタイムなデータ変換やリバース ETL が容易に実現できます。当記事では継続的クエリの使い方を紹介します。

概要

継続的クエリとは

継続的クエリ（Continuous queries）は事前定義した SQL を BigQuery で継続的に実行し、ニアリアルタイムなデータ変換やリバース ETL を実現するための機能です。

同機能では、BigQuery テーブルに追加されたレコードに対して、数秒〜数十秒の遅延で、ほぼリアルタイムに SQL による加工を施すことができます。加工されたデータは、以下のような宛先に格納できます。

BigQuery テーブル
Pub/Sub トピック
Bigtable テーブル

BigQuery ML 機能を利用して生成 AI モデルや各種 Cloud AI API を呼び出すことも可能なため、BigQuery に到着したデータに対して即時に推論を実施することも可能です。

参考 : Introduction to continuous queries

なお当機能は2024年8月現在、承認ベースの Preview 段階であり、前掲の公式ドキュメントからリンクされている申込フォーム経由で Google に対して利用申請する必要があります。また Preview 段階機能の本番環境での利用は推奨されません。Preview 段階のサービスの注意点については、以下の記事もご参照ください。

blog.g-gen.co.jp

ユースケース

継続的クエリは、以下のような用途で用いることが想定されています。

イベントドリブンなデータ加工
リアルタイムな AI/ML 利用（異常検知、感情分析、レコメンデーション、生成 AI アプリ等）
リアルタイムなリバース ETL

実行可能な SQL ステートメント

継続的クエリでは BigQuery で利用可能なすべての SQL が利用可能なわけではなく、一部の SQL ステートメントに限定されています。

INSERT（他の BigQuery テーブルに対するデータ挿入）
EXPORT DATA（Pub/Sub や Bigtable に対するデータ挿入）
ML.GENERATE_TEXT 、 ML.GENERATE_EMBEDDING（機械学習モデルによる推論）
ML.UNDERSTAND_TEXT（Cloud Natural Language API による推論）、ML.TRANSLATE（Cloud Translation API による推論）
ML.NORMALIZER（数値式の配列を正規化）
BigQuery 関数のうちステートレスなもの（他の行の値に依存しない関数）
APPENDS 関数（特定時刻からの継続的クエリの開始。後述）

詳細は以下の公式ドキュメントをご確認ください。

参考 : Supported operations

料金

BigQuery Editions

継続的クエリの利用には、BigQuery のスロット料金が発生します。

継続的クエリを使うには、BigQuery Editions が必須です。継続的クエリを実行するプロジェクトに、Enterprise エディション もしくは Enterprise Plus エディションの予約（Reservation）を割り当てる必要があります。ただし、コミットメント（1年または3年）を購入する必要はありませんので、一時的な利用や検証目的の利用の場合でも、すぐに利用を停止することができます。

BigQuery Editions については、以下の記事も参照してください。

blog.g-gen.co.jp

制約

制約として、当機能は Editions のオートスケーリングに対応していません。Editions で作成した予約（Reservation）のスロットの最大予約サイズ（Max）と最小値（Baseline）値を同一の値にして、スロットがスケーリングしないようにする必要があります。

なお、予約（Reservation）に割り当て可能な最小スロット数は50スロットです。この場合、Enterprise エディションでは1時間あたり $3 の料金が発生します（コミットメントなしの場合）。最大スロット数は500です。

予約（Reservation）を作成してプロジェクトに割り当てる際、ジョブタイプを CONTINUOUS に設定する必要があります。通常のクエリに用いる QUERY ジョブタイプとは別の割り当てが必要になります。

利用方法

シンプルな例

最も単純化した例を示します。

INSERT INTO
 `my_dataset.my_table_02`
SELECT
 id, data
FROM
 `my_dataset.my_table_01`

上記のようなクエリを継続的クエリとして実行すると、まず my_table_01 の既存の全レコードの id 列と data 列が my_table_02 に複製されます。その後は、新規に my_table_01 に到着したデータの id 列と data 列が、リアルタイムに my_table_02 に挿入されるようになります。

EXPORT DATA ステートメントを使用することで、Pub/Sub や Bigtable にもデータをエクスポートできるほか、SELECT 文の中で各種関数や BigQuery ML 関数を利用することが可能です。

ただし、後述の制約のとおり、集計関数や JOIN、CURRENT_DATE などの非決定論的な関数は利用できません。

その他の詳細な利用方法は、以下のドキュメントをご参照ください。

参考 : Create continuous queries

APPENDS 関数の利用

前掲のような継続的クエリを最初に実行するとき、my_table_01 の既存のレコードがすべて my_table_02 に複製されます。

FROM 句で APPENDS 関数を指定すると、特定時刻以降に追加された行だけが my_table_02 に複製されるようになります。詳細な使い方は、以下の公式ドキュメントをご参照ください。

参考 : Start a continuous query from a particular point in time
参考 : APPENDS

制約と注意点

利用できない関数

継続的クエリでは、集計関数や JOIN、CURRENT_DATE などの非決定論的な（実行条件や入力が同じでも実行するたびに値が変わる可能性のある）関数は利用できません。

その他にも SELECT DISTINCT、UDF（User-defined Functions）、ウインドウ関数などが利用できないほか、ワイルドカードテーブルや外部テーブルには対応していません。また、列レベル・行レベルセキュリティの設定されたテーブルには利用できません。

多くの制限がありますので、以下の公式ドキュメントを参照してください。

参考 : Limitations

認証・認可

継続的クエリを長期実行する際は、サービスアカウントを使って認証する必要があります。

コンソール画面等から継続的クエリを設定すると、デフォルトではログイン中のユーザーアカウントによって認証されてしまいますが、この場合、認証トークンの TTL（継続時間）が2日間となり、期限終了後はクエリが停止します。

サービスアカウントで認証した場合、継続時間に制限はありません。

参考 : Authorization

サービスアカウント認証でクエリを実行する方法は、以下をご参照ください。

参考 : Run a continuous query by using a service account

レコードの削除と更新

以下のような継続的クエリを実行する場合を例に取ります。

INSERT INTO
 `my_dataset.my_table_02`
SELECT
 id, data
FROM
 `my_dataset.my_table_01`

このようなクエリを継続的に実行しているとき、my_table_01 上のデータが削除されても、宛先テーブルである my_table_02 のデータは削除されません。

また、UPDATE ステートメントで my_table_01 上のデータを更新すると、 my_table_02 のデータが更新されるわけではなく、my_table_02 に新規レコードとして更新後のデータが挿入されます。

クエリの更新とキャンセル

一度実行を始めた継続的クエリは、編集することができません。動作中のクエリを一度キャンセルして、新しいクエリとして再実行する必要があります。

編集後にクエリを継続的クエリとして再度実行すると、FROM 句で指定されているテーブルの全レコードが改めてエクスポート先に複製されてしまいます。つまり、以前クエリを中止したときにどこまでレコードを転送したか、という状態は保存されず、初めてクエリを実行したときと同じ挙動になります。編集前のクエリをキャンセルした時刻以降のデータだけを対象として転送を再開したい場合、前述の APPEND 関数を利用して処理開始時刻を指定する必要があります。