【各オプションの意味】 gsutil mb -p $PROJECT_ID -c STANDARD -l asia-northeast1 gs://$BUCKET │ │ │ │ │ │ │ │ │ │ │ └── バケット名 │ │ │ │ └── ロケーション（東京） │ │ │ └── ストレージクラス │ │ └── プロジェクトID │ └── mb = make bucket（バケット作成） └── Google Cloud Storage CLI 【ストレージクラス】 STANDARD: 通常（頻繁にアクセス） NEARLINE: 月1回程度のアクセス（30日以上保存） COLDLINE: 年数回のアクセス（90日以上保存） ARCHIVE: ほぼアクセスしない（365日以上保存）

【Pub/Subの仕組み】 Publisher（送り主） │ │ メッセージを送信 ▼ ┌─────────────────────────────────────┐ │ Topic: app-logs │ │ （メッセージの集約点） │ └──────────────┬──────────────────────┘ │ ┌──────────┼──────────┐ ▼ ▼ ▼ ┌────────┐ ┌────────┐ ┌────────┐ │Sub 1 │ │Sub 2 │ │Sub 3 │ │→BigQuery│→GCS │ │→Dataflow│ └────────┘ └────────┘ └────────┘ 💡 1つのトピックに複数のサブスクリプション → 同じデータを複数の処理先に送れる！ 【ack-deadline=60 の意味】 ・60秒以内に処理完了を通知（ack）しないと ・メッセージが再送される（処理失敗とみなす）

【各行の意味を理解しよう】 from google.cloud import pubsub_v1 ├── GCPのPub/Subライブラリをインポート └── pip install google-cloud-pubsub で事前インストール ┌─────────────────────────────────────────────────────────┐ │ publisher = pubsub_v1.PublisherClient() │ ├─────────────────────────────────────────────────────────┤ │ Pub/Subにメッセージを送信するクライアントを作成 │ │ ・PublisherClient = 送信者用クライアント │ │ ・SubscriberClient = 受信者用クライアント（別のクラス）│ │ │ │ 認証: │ │ ・環境変数 GOOGLE_APPLICATION_CREDENTIALS に │ │ サービスアカウントキーのパスを設定 │ │ ・GCE/GKE上では自動で認証される │ └─────────────────────────────────────────────────────────┘ ┌─────────────────────────────────────────────────────────┐ │ topic_path = publisher.topic_path(project_id, topic_id) │ ├─────────────────────────────────────────────────────────┤ │ トピックのフルパスを生成 │ │ │ │ 入力: │ │ ・project_id = “my-project” │ │ ・topic_id = “app-logs” │ │ │ │ 出力: │ │ ・”projects/my-project/topics/app-logs” │ │ │ │ なぜ必要？ │ │ ・Pub/Sub APIはフルパスで指定する必要がある │ │ ・手動で文字列を組み立てるより安全 │ └─────────────────────────────────────────────────────────┘ ┌─────────────────────────────────────────────────────────┐ │ message_bytes = json.dumps(log_data).encode(‘utf-8’) │ ├─────────────────────────────────────────────────────────┤ │ JSONをバイト列に変換 │ │ │ │ ステップ1: json.dumps(log_data) │ │ ・Pythonの辞書 → JSON文字列に変換 │ │ ・{‘key’: ‘value’} → ‘{“key”: “value”}’ │ │ │ │ ステップ2: .encode(‘utf-8’) │ │ ・文字列 → バイト列に変換 │ │ ・Pub/Subはバイト列しか受け付けない │ │ ・UTF-8は日本語も扱えるエンコーディング │ └─────────────────────────────────────────────────────────┘ ┌─────────────────────────────────────────────────────────┐ │ future = publisher.publish(topic_path, data=…) │ ├─────────────────────────────────────────────────────────┤ │ メッセージを非同期で送信 │ │ │ │ ・publish() = 送信処理を開始（すぐ戻る） │ │ ・future = 送信結果を後で取得するためのオブジェクト │ │ ・非同期 = 送信完了を待たずに次の処理に進める │ │ │ │ なぜ非同期？ │ │ ・大量のメッセージを高速に送信できる │ │ ・1件ずつ送信完了を待つと遅くなる │ └─────────────────────────────────────────────────────────┘ ┌─────────────────────────────────────────────────────────┐ │ future.result() │ ├─────────────────────────────────────────────────────────┤ │ 送信完了を待って結果を取得 │ │ │ │ ・送信が完了するまでブロック（待機） │ │ ・成功時: メッセージIDが返る（例: “1234567890”） │ │ ・失敗時: 例外が発生 │ │ │ │ ⚠️ 大量送信時は result() を毎回呼ばない方が高速 │ │ → バッチで送信して最後にまとめて確認する │ └─────────────────────────────────────────────────────────┘

【Apache Beamパイプラインの構造】 p | ‘ステップ名’ >> 処理内容 1️⃣ ‘Read from Pub/Sub’ >> ReadFromPubSub(…) ├── Pub/Subからメッセージを読み込む └── subscriptionで受け取り口を指定 2️⃣ ‘Parse JSON’ >> beam.ParDo(ParseLogFn()) ├── 各メッセージに対してParseLogFn()を実行 ├── JSON文字列を辞書に変換 └── 必要なフィールドだけ抽出 3️⃣ ‘Write to BigQuery’ >> WriteToBigQuery(…) ├── BigQueryテーブルに書き込む ├── CREATE_IF_NEEDED: テーブルがなければ作成 └── WRITE_APPEND: 既存データに追加 【streaming = True の意味】 ・パイプラインが終了せず、ずっと動き続ける ・新しいメッセージが来たら即座に処理

【パーティション = 本棚の棚分け】 ・日付ごとにデータを物理的に分割 ・WHERE timestamp = ‘2025-01-15’ で その日のデータだけスキャン 【クラスタリング = 棚の中の整理】 ・同じ棚の中で、さらに並び替え ・event_type, platformで並べておく ・WHERE event_type = ‘article_read’ で さらに効率的に検索 【効果】 パーティションなし: 全データスキャン → $50/クエリ パーティションあり: 1日分だけ → $0.50/クエリ（1/100！） クラスタリング追加: さらに絞り込み → $0.10/クエリ 💡 両方設定するのがベストプラクティス！