Treasure Data → S3 Tables

Treasure DataからS3バケットへのデータエクスポートに関する簡潔なガイド。Python/Lambdaベースのインフラストラクチャに対応しています。

概要

Treasure DataのS3 Export Integration (V1)を使用すると、クエリ結果をS3バケットに直接書き込むことができます。トリガー方法は3つあります：TD Console、TD CLI、またはWorkflow (Digdag)。インフラがPythonベースのLambda関数を使用している場合、CLI/Workflowアプローチが最もクリーンに統合されます。

公式ドキュメント： Amazon S3 Export Integration V1

前提条件

以下の権限のみを持つAWS IAMユーザー：
- s3:PutObject
- s3:AbortMultipartUpload
TD Toolbeltがインストール済み（CLIアプローチ用）
Access KeyおよびSecret Key（CLIで使用する場合はURL エンコード）

制限事項

クエリ結果エクスポート上限：100GB（超過した場合はクエリを分割してください）
デフォルト形式：CSV (RFC 4180)
サポート形式：CSV、TSV、JSONL
圧縮：gzまたはなし
暗号化されていないリクエストを拒否するS3バケットポリシーを使用している場合は、use_sse=trueを有効にしてください

ステップ1：IAMセットアップ（AWS側）

TDエクスポート用の専用IAMユーザーを作成します。最小限のポリシー：

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "s3:PutObject",
        "s3:AbortMultipartUpload"
      ],
      "Resource": "arn:aws:s3:::your-bucket-name/*"
    }
  ]
}

セキュリティポリシーでIP ホワイトリストが必要な場合は、TDの静的IPをバケットポリシーに追加してください：
TD Static IP Addresses

ステップ2：S3バケット準備

ターゲットバケットを作成するか、既存のバケットを使用します
SSEが必要な場合は、バケットでAES-256サーバーサイド暗号化を有効にします
エクスポート用のフォルダ構造を設定します。例：

s3://your-bucket/
  └── td-exports/
      └── daily/
      └── adhoc/

ステップ3：TD Consoleで認証を作成

Integrations Hub → Catalog → AWS S3で検索
以下の情報でCreate Authenticationを実行します：

パラメータ	値
Endpoint	`s3-ap-northeast-1.amazonaws.com`（バケットのリージョンに合わせてください）
Auth Method	`basic`（インポート専用の場合は`session`を使用）
Access Key ID	IAMユーザーのアクセスキー
Secret Access Key	IAMユーザーのシークレットキー

接続に名前を付ける → Done

エンドポイントリファレンス： AWS S3 Region Endpoints

ステップ4：エクスポート設定

オプションA：TD Console（手動/スケジュール実行）

Data Workbench → Queries → クエリを選択または作成
クエリを実行して結果を検証
Export Resultsをクリック → S3認証を選択
以下を設定します：

フィールド	推奨値
Bucket	`your-bucket-name`
Path	`td-exports/daily/export_${date}.csv.gz`
Format	`csv`または`jsonl`
Compression	`gz`
Include header	Yes
Null string	空文字列
Part Size	10 MB（デフォルト）

（オプション）スケジュールを設定：@daily、@hourly、またはカスタムcron

オプションB：CLI（Lambda統合に最適）

td query \
  --result 's3://ACCESS_KEY:SECRET_KEY@/bucket-name/td-exports/output.csv.gz?compression=gz' \
  -w -d your_database \
  "SELECT * FROM your_table WHERE time > 1234567890"

SSE有効時：

td query \
  --result 's3://ACCESS_KEY:SECRET_KEY@/bucket-name/path/file.csv?use_sse=true&sse_algorithm=AES256' \
  -w -d your_database \
  "SELECT * FROM your_table"

アクセスキーおよびシークレットキーはURL エンコードである必要があります。

オプションC：Workflow（Digdag）

timezone: UTC
_export:
  td:
    database: your_database

+export-to-s3:
  td>: queries/export_query.sql
  result_connection: your_connection_name
  result_settings:
    bucket: your-bucket-name
    path: /td-exports/daily/data_${moment(session_time).format("YYYYMMDD")}.csv.gz
    compression: 'gz'
    header: true
    newline: \r\n

ワークフロー例： Treasure Boxes — S3 Export

ステップ5：Python/Lambdaスタックとの統合

インフラがPython Lambda関数を使用している場合、TDエクスポートをパイプラインに統合する方法は以下の通りです：

アプローチ1：TD Python Client → Lambda Trigger

import pytd
import os

# TD クライアント設定
client = pytd.Client(
    apikey=os.environ['TD_API_KEY'],
    endpoint='https://api.treasuredata.com'
)

# エクスポートクエリを実行
client.query(
    'your_database',
    'SELECT * FROM your_table',
    result_url=f"s3://{os.environ['AWS_ACCESS_KEY']}:{os.environ['AWS_SECRET_KEY']}@/your-bucket/exports/data.csv.gz?compression=gz"
)

アプローチ2：LambdaがS3ランディング時にリッスン

TDスケジュール実行エクスポートがs3://bucket/td-exports/に書き込み
S3 Event NotificationがPutObjectでLambdaをトリガー
Lambdaがランディングファイルを処理（変換、RDSへのロード等）

# S3イベントによってトリガーされるLambdaハンドラ
def handler(event, context):
    bucket = event['Records'][0]['s3']['bucket']['name']
    key = event['Records'][0]['s3']['object']['key']

    # TDエクスポートファイルを処理
    # 例：CSV読み込み、変換、別の場所へのロード

アプローチ3：Step Functionsオーケストレーション

TD Scheduled Export → S3 Landing → S3 Event → Lambda (Transform) → Target (RDS/DynamoDB/Athena)

エクスポート形式オプションクイックリファレンス

オプション	値	デフォルト
`format`	`csv`、`tsv`、`jsonl`	`csv`
`compression`	`gz`、なし	なし
`delimiter`	`,`、`\t`、`\|`	`,`
`header`	`true`、`false`	`true`
`null`	空、`\N`、`NULL`、`null`	空
`newline`	`CRLF`、`LF`、`CR`	`CRLF`
`quote`	`"`またはカスタム	`"`

スケジューリング（Cronリファレンス）

Cron	スケジュール
`0 * * * *`	毎時間
`0 0 * * *`	毎日真夜中
`0 0 1 * *`	毎月初日
`0 /6 * *`	6時間ごと
`30 2 * * *`	毎日午前2時30分

有用なリンク

リソース	URL
S3 Export Integration ドキュメント	docs.treasuredata.com/int/amazon-s3-export-integration-v1
TD Toolbelt (CLI)	toolbelt.treasuredata.com
TD Python Client (pytd)	github.com/treasure-data/pytd
ワークフロー例	Treasure Boxes S3 Export
ワークフローシークレット	Secret Management Docs
TD Static IPs（ホワイトリスト用）	IP Addresses Doc
AWS S3 リージョン別エンドポイント	AWS Docs
AWS IAM ベストプラクティス	AWS IAM Docs
S3 サーバーサイド暗号化	AWS SSE Docs

概要​

前提条件​

制限事項​

ステップ1：IAMセットアップ（AWS側）​

ステップ2：S3バケット準備​

ステップ3：TD Consoleで認証を作成​

ステップ4：エクスポート設定​

オプションA：TD Console（手動/スケジュール実行）​

オプションB：CLI（Lambda統合に最適）​

オプションC：Workflow（Digdag）​

ステップ5：Python/Lambdaスタックとの統合​

アプローチ1：TD Python Client → Lambda Trigger​

アプローチ2：LambdaがS3ランディング時にリッスン​

アプローチ3：Step Functionsオーケストレーション​

エクスポート形式オプション クイックリファレンス​

スケジューリング（Cronリファレンス）​

有用なリンク​

概要