BigQuery

BigQuery Connectorは、BigQueryのデータセットからテーブルとビューを読み、レコードとテーブルを書き戻し、Google Cloud Storageからコンテンツをロードし、プロシージャとクエリジョブを実行します。プロダクトのイベント、セッション、請求のデータがすでにGoogle Cloudにあるときに使うDWHです。絞り込みと集計はBigQueryの中にとどまり、エンジンに届くのは結果だけです。

読み取り元Copy Anchor Link

設定したデータセットにあるすべてのテーブルとビューが読み取り元です。bigquery.sessions、bigquery.orders、bigquery.customers のように参照します。列と型はデータセットのメタデータから取ります。ネストした項目と繰り返し項目は、オブジェクトと配列として読みます。パーティション分割やクラスタ化されたテーブルも他と同じように読め、パーティション列での絞り込みはプッシュダウンされるので、BigQueryは必要なパーティションだけをスキャンします。

テーブルのキーは、宣言されている PRIMARY KEY です。BigQueryは主キーを強制しないので、Connectorはこの宣言を事実ではなく、書き込みの前に検証すべき主張として扱います（Effectと操作を参照）。キーを宣言していないテーブルも、接続側でキーになる列を宣言すれば更新できます。

別のデータセットを読むには、プログラムの中で名前を修飾するのではなく、bigquery_raw のような2つ目の接続を設定してください。

型Copy Anchor Link

書き込みは同じ規則で逆方向に変換します。対象の列に収まらない値は、静的に分かるなら check で、そうでなければその項目が失敗します。黙って切り詰めることはありません。

Effectと操作Copy Anchor Link

レコードのEffectはキーで書きます。

• insert — EFFECT INSERT bigquery.account_scores (col, …)。小さなEffectにはストリーミング挿入、大きなEffectにはバッチロードをConnectorが選びます。どちらを選んだかはRunの記録に残ります。
• update — EFFECT UPDATE bigquery.account_scores ID account_id SET (…)。
• upsert — EFFECT UPSERT bigquery.account_scores ID account_id SET (…)。MERGE として適用します。
• delete — EFFECT DELETE bigquery.account_scores ID account_id。

ID で指定するキーは、ちょうど1行を識別しなければなりません。BigQueryは一意性を強制しないので、Connectorはプレビューを作るときに検証し、複数の行に一致するキーの項目は failed になります。単一行の変更として適用されることはありません。同じキーに対して元のレコードが2つの異なる値を持つ更新は、check で拒否されます。ストリーミング挿入で書かれた行は、短い間（ストリーミングバッファ）更新も削除もできません。その行を対象にした項目は failed（conflict）で終わり、あとのRunで再試行できます。

テーブルとコンテンツ：

• create table — |> CREATE [OR REPLACE] TABLE bigquery.events_30d は、レコードをデータセットの中のテーブル全体として書きます。
• store — EFFECT STORE bigquery.loads SET (table = 'raw_events', uri = 'gs://acme-landing/events/*.jsonl', source_format = 'JSONL') は、すでにGoogle Cloud Storageにあるオブジェクトからテーブルへのロードジョブを開始します。コンテンツを先にそこへ置くにはGCS Connectorを使います。

操作：

• call — EFFECT COMMAND bigquery.call ID key INPUT (procedure = 'product.rebuild_scores', args = [...]) は、ストアドプロシージャを呼びます。
• run_query_job — EFFECT COMMAND bigquery.run_query_job ID key INPUT (sql = @statement, destination = 'product.snapshot') は、結果を宛先のテーブルに書くクエリジョブを実行します。SQLはパラメーターで、Hostがアカウントの権限に照らして検査します。プログラムがレコードの値から組み立てることはできません。

論理的な項目と物理的な文は別のものです。Connectorは1,000件の UPSERT の項目を1つの MERGE で適用してもかまいませんが、各項目は自分の識別子、プレビューの行、結果を持ち続けます。

セットアップCopy Anchor Link

"bigquery": {
  "connector": "bigquery",
  "project": "acme-analytics",
  "dataset": "product",
  "location": "EU",
  "credentials": {"service_account": {"env": "BIGQUERY_SERVICE_ACCOUNT_JSON"}}
}

サービスアカウントには、読み取りにプロジェクトでの bigquery.tables.getData と bigquery.jobs.create、レコードの書き込みに bigquery.tables.updateData、CREATE TABLE とロードジョブに bigquery.tables.create、プロシージャの呼び出しに bigquery.routines.get が必要です。読み取りは roles/bigquery.dataViewer と roles/bigquery.jobUser で、書き込みは roles/bigquery.dataEditor でまかなえます。読み取り専用の接続と書き込み用の接続を、2つの接続名のもとで2つのサービスアカウントに分けることもできます。

ActQL Cloudでも項目は同じです。サービスアカウントの鍵はCloudのシークレットストアに保存するか、鍵の代わりにCloud自身のGoogleのIDにロールを付与します。

CapabilityとプッシュダウンCopy Anchor Link

explain は、プッシュダウンされた各ステップをSite connector:bigquery、Engine BigQuery SQL で表示し、ドライランで見積もれるときはジョブがスキャンするバイト数も示します。

プレビューと突き合わせCopy Anchor Link

プレビュー。 更新、UPSERT、削除は full_diff でプレビューします。Connectorは対象のレコードをキーで読み、現在の値を記録し、プレビューには行ごとに現在の値と新しい値が出ます。BigQueryには行のバージョンがないので、Connectorは対象の各行の現在の値のハッシュを記録し、MERGE や UPDATE をその条件付きにします。プレビューと適用の間に変わった行は更新されません。その項目は failed（conflict）で終わり、続けるよう求められればRunは新しいプレビューを作ります。追加、テーブル作成、ロードジョブは targeted でプレビューします。何をどこに書くかと、CREATE OR REPLACE なら置き換えられる件数です。プロシージャの呼び出しとクエリジョブは、名前、引数、宛先をプレビューします。

突き合わせ。 書き込み、ロード、クエリはすべてBigQueryのジョブです。ConnectorはジョブIDを送信前に自分で生成し、Effectの項目に対して記録します。応答が失われたら、そのジョブを取得して突き合わせます。エラーなしで完了していれば文はコミットされ、エラーで完了していれば failed、まだ実行中なら項目は進行中のままです。プロジェクトの履歴にあっても、EffectのジョブIDを持たないジョブは証跡になりません。対象のレコードが期待どおりの値を持っていることも同じです。ストリーミング挿入にはジョブがなく、BigQueryには insertId で照会する手段もないので、応答が失われた項目は unknown のままで、何も再送しません。突き合わせが必要な追加には、ジョブになるバッチロードを使います。操作は、ジョブが作られれば accepted、エラーなしで DONE に達すれば applied、どちらも確定できなければ unknown で終わります。

上限と費用Copy Anchor Link

• プッシュダウンされたクエリはBigQueryのジョブになり、スキャンしたバイト数（予約があればスロット時間）で課金されます。explain は、取れるときはドライランの見積もりを表示します。1回の読み取りがスキャンできる量は、プログラムのバイト数の上限で抑えられます。
• 結果はAPIのページ単位で取得します。上限に達したConnectorは、途中までの結果を返すのではなく、エラーで止まります。
• ストリーミング挿入はすぐに見えるようになりますが、最大90分間は変更できません。バッチロードは無料ですが、見えるまでに時間がかかります。Connectorはどちらを使ったかを報告します。
• ロードジョブとテーブル作成は、プロジェクトの1日あたりのロードとDMLの割り当てを消費します。超えそうなプログラムは送信前に失敗します。

例Copy Anchor Link

直近1週間の日次アクティブユーザー数です。

FROM bigquery.sessions
|> WHERE session_date >= DATE_SUB(CURRENT_DATE(), INTERVAL 7 DAY)
|> AGGREGATE COUNT(DISTINCT user_id) AS dau GROUP BY session_date
|> ORDER BY session_date;

セッションがなくなった取引先にSalesforce側で印を付けます。

SELECT a.id AS id, s.last_session
FROM salesforce.accounts AS a
JOIN bigquery.account_activity AS s ON a.id = s.account_id
|> WHERE s.last_session < DATE_SUB(CURRENT_DATE(), INTERVAL 30 DAY)
|> EFFECT UPDATE salesforce.Account
     ID id SET (Activity_Status__c = 'dormant');

スコアのテーブルを最新に保ちます。

SELECT a.id AS account_id, a.description AS notes, s.sessions_30d
FROM salesforce.accounts AS a
JOIN bigquery.account_activity AS s ON a.id = s.account_id
|> INFER `model-task:churn-risk.v1`
     INPUT (notes = notes, active_users = sessions_30d, trend = 'flat') AS risk
|> WHERE risk.outcome = 'ok'
|> SELECT account_id, risk.value.level AS level, risk.value.reason AS reason
|> EFFECT UPSERT bigquery.account_scores ID account_id SET (level = level, reason = reason);