OCI Streaming

解決する課題

物理的な Kafka クラスタの構築・運用（ブローカー、ZooKeeper、パッチ、スケール）を抱えずに、大量のイベントをリアルタイムに扱えます。

• 大量データをリアルタイムに取り込みたい
• 複数の処理系で同じストリームを並行処理したい（コンシューマグループ）
• 既存の Kafka クライアント／エコシステムをそのまま使いたい
• ストリームを Object Storage / Autonomous Database 等へ配信したい（Service Connector Hub 経由）

主要概念と用語

• Stream（ストリーム）: メッセージを保持する論理的な単位。Kinesis のストリームに相当
• Partition（パーティション）: スループットの単位。1 パーティション = 書き込み 1 MB/s（最大1000メッセージ/秒）、読み取り 2 MB/s。Kinesis のシャードに相当
• Stream Pool（ストリームプール）: 複数ストリームをまとめる管理・分離の単位。Kafka 接続設定・暗号化キー・プライベートエンドポイントをここで設定
• Message Key（メッセージキー）: 同一キーは同一パーティションに振り分けられ、順序が保証される。Kinesis のパーティションキーに相当
• Offset（オフセット）: パーティション内でのメッセージ位置。コンシューマはオフセットを管理して読み直せる
• 保持期間（Retention）: **24時間〜168時間（7日）**で設定。期間内はメッセージを再読み込み可能
• Streaming API / Kafka 互換 API: ネイティブの REST/SDK と、Apache Kafka プロトコル互換の 2 系統で読み書きできる

仕様・制限・クォータ

• ストリームのスループットはパーティション数で決まる（パーティションを増やしてスケールアウト）
• パーティションあたり書き込み 1 MB/s・1000 メッセージ/秒、読み取り 2 MB/s・5 リクエスト/秒
• 1 メッセージの最大サイズは 1 MB
• 保持期間は 24〜168 時間の範囲で設定（作成後の延長は不可、作成時に決定）
• Kafka 互換 API ではコンシューマグループ・__consumer_offsets 相当のオフセットコミットに対応
• リージョン／テナンシごとにパーティション総数などのサービス制限があり、引き上げ申請が可能

内部の仕組み

OCI Streaming は、メッセージキーのハッシュに基づいてレコードをパーティションへ分散し、各パーティション内では追記専用（append-only）ログとして順序を保って保持します。保持期間内であれば、複数のコンシューマが各自のオフセットから独立して読み直せるのが特徴です（キューのように取り出して消えるわけではない）。

• ストリームはストリームプールに属し、プール単位で暗号化キーやエンドポイント（パブリック／プライベート）を管理
• 内部実装は Apache Kafka をベースにしており、Kafka クライアントからブートストラップサーバーとしてストリームプールのエンドポイントを指定すれば、ネイティブの Kafka プロデューサ／コンシューマがそのまま動作
• 他サービスへの配信は Service Connector Hub が担い、ストリームをソースに Object Storage / Functions / Notifications / Monitoring 等へ転送する

設計パターン / ベストプラクティス

• 取り込み＝Streaming、他サービスへの配信＝Service Connector Hub、複雑なリアルタイム分析＝GoldenGate Stream Analytics や OCI 上の Flink/Spark と組み合わせる
• メッセージキー設計で偏り（ホットパーティション）を回避し、スループットを均等化する
• スループットが足りなければパーティション数を増やす（プロデューサ／コンシューマも並列化）
• 既存資産がある場合はKafka 互換 API を使い、アプリ改修を最小化して移行する
• 保持期間は再処理の要件に合わせて設定（最大 7 日）。長期保管は Object Storage へ吐き出す

運用・監視

• OCI Monitoring（oci_streaming 名前空間）でメトリクスを監視: 取り込みバイト/メッセージ数、読み取りバイト数、スロットル、PutMessagesLatency など
• スロットル（書き込み拒否）が出たらパーティション不足／ホットキーを疑い、パーティション増設やキー分散を検討
• コンシューマの遅延はオフセットのラグ（最新オフセットと処理済みオフセットの差）で監視
• 監査は OCI Audit に記録。ストリーム作成・削除・設定変更を追跡

コスト

OCI Streaming は従量課金で、主に「保持されるパーティション数（時間課金）」と「リクエスト数」「取り込み量」で構成されます。サーバーやブローカーの常時起動コストはありません。

セキュリティ

• 保存時暗号化は既定で有効。Oracle 管理キーに加え、OCI Vault のカスタマー管理キー（CMK） をストリームプールに指定可能
• 転送時は TLS で保護。Kafka 互換 API は SASL/PLAIN over SSL（認証トークン）で接続
• アクセス制御は IAM ポリシーでプロデューサ／コンシューマの権限を最小化（use stream-push / use stream-pull 等）
• プライベートエンドポイントをストリームプールに構成すれば、VCN 内からインターネットを経由せずアクセス可能

関連サービス・比較（AWS との対応）

ハンズオン / CLI例

# 1. ストリームプールを作成（プールIDを後続で利用）
oci streaming admin stream-pool create \
  --compartment-id ocid1.compartment.oc1..xxxx \
  --name demo-pool

# 2. ストリームを作成（パーティション数1・保持24時間）
oci streaming admin stream create \
  --compartment-id ocid1.compartment.oc1..xxxx \
  --name events \
  --partitions 1 \
  --retention-in-hours 24 \
  --stream-pool-id ocid1.streampool.oc1..xxxx

# 3. メッセージを投入（value/key は base64 エンコードして渡す）
oci streaming stream message put \
  --stream-id ocid1.stream.oc1..xxxx \
  --messages '[{"key":"'$(echo -n 'user-123' | base64)'","value":"'$(echo -n '{"click":1}' | base64)'"}]'

# 4. カーソルを作成してメッセージを読み取り
CURSOR=$(oci streaming stream cursor create-cursor \
  --stream-id ocid1.stream.oc1..xxxx \
  --partition 0 --type TRIM_HORIZON \
  --query 'data.value' --raw-output)

oci streaming stream message get \
  --stream-id ocid1.stream.oc1..xxxx \
  --cursor "$CURSOR"