Amazon S3

Amazon S3 Connectorは、バケットのオブジェクトを一覧し、その内容を読み、オブジェクトを書き込み・置換し、メタデータとストレージクラスを変更し、削除し、コピーし、アーカイブしたオブジェクトを復元します。Google Cloud Storage Connectorと同じ考え方です。バケットは、他のシステムからファイルが届く場所であり、プログラムがエクスポートを置いていく場所です。

読み取り元Copy Anchor Link

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

version は、一覧で見たバイト列を識別します。バージョニングが有効なバケットでは、置き換えられたオブジェクトは別の version_id です。無効なバケットでは、ETagが変わっていれば、一覧で見たファイルとしてではなく競合として読みます。

型Copy Anchor Link

Effectと操作Copy Anchor Link

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

• update — EFFECT UPDATE s3.objects ID key SET (content_type = …, storage_class = …, metadata = …) は、メタデータかストレージクラスを書き換えます。S3にはメタデータだけを更新する手段がないので、Connectorはプレビューで見たETagを条件にして、MetadataDirective: REPLACE でオブジェクトを自身にコピーします。内容は変わりません。
• delete — EFFECT DELETE s3.objects ID key は、オブジェクトを削除します。バージョニングが有効なバケットでは、プレビューした version_id を削除します。

コンテンツ：

• upload — EFFECT STORE s3.objects SET (key = …, content = content [, content_type = …, storage_class = …, overwrite = TRUE]) は、レコードの content（SERIALIZE または contents の読み取り元から来た content_ref）からオブジェクトを作ります。overwrite がなければ If-None-Match: * 付きで、あればプレビューで見たETagに対する If-Match 付きで送ります。

操作：

• copy — EFFECT COMMAND s3.copy TARGET key ID key INPUT (destination = 'other-bucket/path/file.pdf') は、アカウントが書き込めるバケットの中、またはバケット間でオブジェクトをコピーします。
• restore — EFFECT COMMAND s3.restore TARGET key ID key INPUT (days = 7, tier = 'Standard' | 'Bulk' | 'Expedited') は、GlacierやDeep Archiveのオブジェクトの一時的な復元を要求します。

セットアップCopy Anchor Link

"s3": {
  "connector": "s3",
  "bucket": "acme-exports",
  "region": "eu-west-1",
  "prefix": "actql/",
  "credentials": {
    "access_key_id": {"env": "AWS_ACCESS_KEY_ID"},
    "secret_access_key": {"env": "AWS_SECRET_ACCESS_KEY"}
  }
}

credentials を省略すると、キーの代わりにHostに割り当てられたIAMロールを使えます。アイデンティティには、バケットへの s3:ListBucket、読み取りに s3:GetObject（バージョニングが有効なバケットでは s3:GetObjectVersion も）、アップロードと書き換えに s3:PutObject、削除に s3:DeleteObject / s3:DeleteObjectVersion、復元に s3:RestoreObject が必要です。コピーには、元に対する読み取りと宛先に対する s3:PutObject も必要です。prefix を指定すると接続をバケットの一部に限定でき、プログラムからその外のキーを指定することはできません。endpoint の項目でS3互換のサービスを選べます。ActQL CloudではキーをCloudのシークレットストアに保存するか、付与したロールをCloudが引き受けます。接続の項目は同じです。

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

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

プレビュー。 メタデータの更新と削除は full_diff です。Connectorはオブジェクトのメタデータ（HEAD）を読み、ETagと version_id を記録し、プレビューには現在の値と新しい値が出ます。リクエストは If-Match を持つので、その間に変わったオブジェクトは黙って上書きされるのではなく、項目が failed（conflict）になります。アップロードは targeted です。プレビューには、書き込まれるキー、メディアタイプ、バイト数、コンテンツの識別子（Connectorがチェックサムとして送るSHA-256）と、既存のオブジェクトを置き換えるかどうかが出ます。コピーは元と宛先を、復元はオブジェクト、tier、日数をプレビューします。

突き合わせ。 各項目は、キー、期待するチェックサム、マルチパートアップロードならアップロードidに結び付けられます。アップロード、コピー、メタデータの書き換えは、ユーザーメタデータ（x-amz-meta-actql-request）にリクエストのマーカーを持ちます。応答が失われた場合は、宛先の HEAD でそのマーカーを探し、マルチパートアップロードは記録したアップロードidで突き合わせます。チェックサムや項目が一致していてもマーカーがなければ証跡になりません。削除と復元にはS3が保持するマーカーがないので、応答が失われた項目は unknown のままです。進行中のマルチパートアップロードはアップロードidで一覧し、記録済みのパートから完了させるか中止します。やみくもに完了させることはありません。キーは合っていてもチェックサムの違うオブジェクトは証跡になりません。項目は unknown のままで、何も再送しません。アップロード、コピー、更新、削除は確認が取れた時点で applied になります。復元は accepted になり、Glacierが処理しているあいだはそのままで、explain --run と status に完了の見込みが出ます。

上限と費用Copy Anchor Link

• 一覧は1,000キーずつページングされ、プログラムの件数の上限の範囲で読みます。
• 内容の読み取りはプログラムのバイト数の上限で抑えられます。大きなオブジェクトは途中まで読まれるのではなく拒否されます。読み取りではETagかチェックサムを検証します。
• 8MiBを超えるオブジェクトはマルチパートアップロードで送ります。1つのオブジェクトは最大5TiBです。5GiBを超えるオブジェクトのコピーはマルチパートコピーになります。
• S3はリクエストのクラスごと（LISTとGET、PUTとCOPY）、クラス別の保存容量、tier別の復元、リージョン外へのデータ転送に課金します。Connectorのリクエストは通常の課金対象の操作です。

例Copy Anchor Link

直近1時間に届いたファイルです。

FROM s3.objects
|> WHERE key LIKE 'inbound/%' AND last_modified >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 HOUR)
|> SELECT key, size, last_modified
|> ORDER BY last_modified DESC;

進行中の取引を日次でエクスポートします。

FROM hubspot.deals
|> WHERE stage NOT IN ('closedwon', 'closedlost')
|> SELECT id, dealname, amount, stage
|> SERIALIZE FORMAT CSV ORDER BY id
     OPTIONS (name = CONCAT('open-deals-', FORMAT_DATE('%Y%m%d', CURRENT_DATE()), '.csv'),
              header = TRUE, null_marker = '')
|> EFFECT STORE s3.objects SET (key = 'exports/' || name, content = content);

届いたPDFを、そのままアーカイブ用のバケット（archive という2つ目のS3接続）にコピーします。

FROM LIST_FILES(s3, glob => r'inbound/*.pdf') AS f
|> JOIN s3.contents AS c ON f.resource = c.resource AND f.version = c.version
|> EFFECT STORE archive.objects SET (key = 'pdf/' || f.path, content = c.content);

昨年アーカイブした明細を1週間だけ戻します。

FROM s3.objects
|> WHERE key LIKE 'statements/2025/%' AND storage_class = 'DEEP_ARCHIVE'
|> EFFECT COMMAND s3.restore TARGET key ID key INPUT (days = 7, tier = 'Bulk');