Google Cloud Storage

Google Cloud Storage Connectorは、バケットのオブジェクトを一覧し、その内容を読み、オブジェクトを書き込み・置換し、メタデータを変更し、削除し、コピーと結合をします。他のシステムがバケットに置いたファイルをプログラムが拾い、エクスポートを残していくための手段です。

読み取り元Copy Anchor Link

• gcs.objects — バケットのオブジェクト。resource、version（generation）、name、size、content_type、generation、metageneration、storage_class、updated、md5、crc32c、metadata（カスタムメタデータのJSONオブジェクト）。キーは name です。プレフィックスで絞り込むと、1つの「フォルダ」を一覧できます。LIST_FILES(gcs, glob => r'inbound/*.csv') はパスによる同じ一覧で、resource、version、path を返します。
• gcs.contents — オブジェクトの内容。resource、version、name、media_type、そして content_ref の content。一覧から resource と version でJOINしてください。ConnectorはJOINで指定されたgenerationだけを読みます。CSV、Parquet、JSON Linesのオブジェクトのレコードを型付きのテーブルとして読むには、ローカルファイルConnectorのテーブルの読み取り元を gs:// のパスに向けます。

version はオブジェクトのgenerationです。プログラムは、その間に上書きされていても、一覧で見たとおりのバイト列を読みます。

型Copy Anchor Link

Effectと操作Copy Anchor Link

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

• update — EFFECT UPDATE gcs.objects ID name SET (content_type = …, storage_class = …, metadata = …) は、内容を変えずにオブジェクトのメタデータかストレージクラスを書き換えます。ストレージクラスの変更はその場での書き換えで、新しいgenerationになります。
• delete — EFFECT DELETE gcs.objects ID name は、オブジェクト（プレビューしたgeneration）を削除します。

コンテンツ：

• upload — EFFECT STORE gcs.objects SET (name = …, content = content [, content_type = …, overwrite = TRUE]) は、レコードの content（SERIALIZE または contents の読み取り元から来た content_ref）からオブジェクトを作ります。overwrite がなければ、その名前のオブジェクトが存在しないことが書き込みの条件です。あれば、プレビューで見たgenerationを置き換えます。

操作：

• copy — EFFECT COMMAND gcs.copy TARGET name ID key INPUT (destination = 'other-bucket/path/file.csv') は、オブジェクトをバケット内か、アカウントが書き込める別のバケットにコピーします。
• compose — EFFECT COMMAND gcs.compose TARGET destination ID key INPUT (sources = ['part-1.csv', 'part-2.csv']) は、最大32個のオブジェクトを1つに連結します。

セットアップCopy Anchor Link

"gcs": {
  "connector": "gcs",
  "bucket": "acme-exports",
  "prefix": "actql/",
  "credentials": {"service_account": {"env": "GOOGLE_SERVICE_ACCOUNT_JSON"}}
}

サービスアカウントには、読み取りに storage.objects.list と storage.objects.get、アップロードに storage.objects.create、メタデータの変更に storage.objects.update、削除と置換に storage.objects.delete が必要です。roles/storage.objectViewer と roles/storage.objectAdmin でまかなえます。prefix を指定すると、接続をバケットの一部に限定できます。プログラムからその外のオブジェクトを指定することはできません。credentials を省略すると、鍵ファイルの代わりにWorkload Identityや割り当てられたサービスアカウントを使えます。ActQL Cloudでは鍵をCloudのシークレットストアに保存します。接続の項目は同じです。

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

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

プレビュー。 メタデータの更新と削除は full_diff です。Connectorはオブジェクトのメタデータを読み、generationとmetagenerationを記録し、プレビューには現在の値と新しい値が出ます。リクエストは ifGenerationMatch と ifMetagenerationMatch 付きで送るので、その間に変わったオブジェクトは項目が failed（conflict）になります。アップロードは targeted です。プレビューには、書き込まれるオブジェクト名、メディアタイプ、バイト数、コンテンツの識別子（crc32c）と、置き換えるgeneration、または存在しないことという条件が出ます。コピーと結合は、元のオブジェクト、宛先、置き換えるかどうかをプレビューします。

突き合わせ。 Connectorは各項目を、書き込むオブジェクトと、承認されたコンテンツのチェックサムに結び付けます。アップロード、結合、メタデータの更新は、オブジェクトのカスタムメタデータ（x-goog-meta-actql-request）にリクエストのマーカーを持ちます。応答が失われた場合は、宛先のメタデータからそのマーカーを探して突き合わせます。crc32c や項目が一致していてもマーカーがなければ証跡になりません。削除にはマーカーがないので、応答が失われた削除は unknown のままです。再開可能なアップロードはセッションURIを保持しているので、中断したアップロードは何より先にその状態を照会します。承認された識別子と一致しないものは証跡になりません。項目は unknown のままで、何も再送しません。操作はGCSが確認した時点で applied になります。大きなオブジェクトのバケット間コピーはrewriteとして動き、最後のチャンクが確認されるまで accepted です。

上限と費用Copy Anchor Link

• 一覧は1,000オブジェクトずつページングされ、プログラムの件数の上限の範囲で読みます。
• 内容の読み取りはプログラムのバイト数の上限で抑えられます。大きなオブジェクトは途中まで読まれるのではなく拒否されます。読み取りのたびに、オブジェクトのチェックサムで整合性を確認します。
• 8MiBを超えるアップロードは再開可能なセッションを使います。1つのオブジェクトは最大5TiBです。compose は1回に最大32個の元オブジェクトを取ります。
• GCSは操作のクラスごと（一覧と取得はクラスB、作成とrewriteはクラスA）、保存容量、リージョン外への送信に課金します。Connectorのリクエストは通常の課金対象の操作です。

例Copy Anchor Link

昨日アップロードされたものです。

FROM gcs.objects
|> WHERE name LIKE 'inbound/%' AND updated >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 DAY)
|> SELECT name, size, updated
|> ORDER BY updated DESC;

リスクの高い取引先をJSON Linesのオブジェクトとしてエクスポートします。

SELECT a.id, a.name, risk.value.level AS level, risk.value.reason AS reason
FROM salesforce.accounts AS a
JOIN snowflake.usage_30d AS u ON a.id = u.account_id
|> INFER `model-task:churn-risk.v1`
     INPUT (a.description AS notes, u.active_users, u.trend) AS risk
|> WHERE risk.outcome = 'ok' AND risk.value.level = 'high'
|> SERIALIZE FORMAT JSONL ORDER BY id OPTIONS (name = 'high-risk.jsonl')
|> EFFECT STORE gcs.objects SET (name = 'exports/' || name, content = content, overwrite = TRUE);

PDFで届いた請求書を読み、LLMに項目を抜き出させます。

FROM LIST_FILES(gcs, glob => r'invoices/*.pdf') AS f
|> JOIN gcs.contents AS c ON f.resource = c.resource AND f.version = c.version
|> INFER `model-task:invoice-fields.v1` INPUT (document = c.content) AS inv
|> WHERE inv.outcome = 'ok'
|> SELECT f.path, inv.value.supplier, inv.value.total, inv.value.due_date;

処理の終わった受信ファイルをコールドストレージに移します。

FROM gcs.objects
|> WHERE name LIKE 'inbound/processed/%' AND storage_class = 'STANDARD'
|> EFFECT UPDATE gcs.objects ID name SET (storage_class = 'COLDLINE');