streaming-data

ストリーミングデータ処理

最新のメッセージブローカーとストリームプロセッサーを使用して、本番環境に対応したイベントストリーミングシステムとリアルタイムデータパイプラインを構築します。

このスキルを使用する場面

次のような場合に、このスキルを使用してください。

• イベント駆動型アーキテクチャやマイクロサービス間通信を構築する
• リアルタイム分析、監視、アラートシステムを処理する
• データ統合パイプライン（CDC、ETL/ELT）を実装する
• ログやメトリクスの集約システムを作成する
• IoT プラットフォームや高頻度取引システムを開発する

コアコンセプト

メッセージブローカーとストリームプロセッサーの違い

メッセージブローカー（Kafka、Pulsar、Redpanda）：

• イベントストリームを保存し、配信します
• 耐久性、再生機能、パーティショニングを提供します
• プロデューサーとコンシューマーの連携を処理します

ストリームプロセッサー（Flink、Spark、Kafka Streams）：

• ストリーミングデータを変換し、集約します
• ウィンドウ処理、結合、ステートフルな操作を提供します
• 複合イベント処理（CEP）を実行します

配信保証

最大 1 回：

• メッセージが失われる可能性がありますが、重複はありません
• オーバーヘッドが最小です
• 使用例：損失が許容されるメトリクスやログ

最低 1 回：

• メッセージは失われませんが、重複する可能性があります
• オーバーヘッドは中程度で、冪等なコンシューマーが必要です
• 使用例：ほとんどのアプリケーション（デフォルトの選択肢）

厳密に 1 回：

• メッセージは失われず、重複もしません
• オーバーヘッドが最大で、トランザクション処理が必要です
• 使用例：金融取引、重要な状態更新

クイックスタートガイド

ステップ 1：メッセージブローカーを選択する

詳しい比較については、references/broker-selection.md を参照してください。

簡単な選び方：

• Apache Kafka：成熟したエコシステム、エンタープライズ機能、イベントソーシング
• Redpanda：低レイテンシー、Kafka 互換、運用が簡単（ZooKeeper 不要）
• Apache Pulsar：マルチテナンシー、地理的レプリケーション、階層型ストレージ
• RabbitMQ：従来型のメッセージキュー、RPC パターン

ステップ 2：ストリームプロセッサーを選択する（必要な場合）

詳しい比較については、references/processor-selection.md を参照してください。

簡単な選び方：

• Apache Flink：ミリ秒単位のレイテンシー、リアルタイム分析、CEP
• Apache Spark：バッチとストリームのハイブリッド、ML 統合、分析
• Kafka Streams：マイクロサービスに組み込み可能、独立したクラスターが不要
• ksqlDB：ストリーム処理用の SQL インターフェース

ステップ 3：プロデューサー／コンシューマーパターンを実装する

言語別のガイドを選択してください。

• TypeScript/Node.js：references/typescript-patterns.md（KafkaJS）
• Python：references/python-patterns.md（confluent-kafka-python）
• Go：references/go-patterns.md（kafka-go）
• Java/Scala：references/java-patterns.md（Apache Kafka Java Client）

一般的なパターン

基本的なプロデューサーパターン

エラー処理を行いながら、トピックにイベントを送信します。

1. Create producer with broker addresses
2. Configure delivery guarantees (acks, retries, idempotence)
3. Send messages with key (for partitioning) and value
4. Handle delivery callbacks or errors
5. Flush and close producer on shutdown

基本的なコンシューマーパターン

オフセットを管理しながら、トピックからイベントを処理します。

1. Create consumer with broker addresses and group ID
2. Subscribe to topics
3. Poll for messages
4. Process each message
5. Commit offsets (auto or manual)
6. Handle errors (retry, DLQ, skip)
7. Close consumer gracefully

エラー処理戦略

本番環境のシステムでは、次の機能を実装してください。

• デッドレターキュー（DLQ）：失敗したメッセージを別のトピックに送信する
• 再試行ロジック：バックオフ付きで再試行回数を設定する
• グレースフルシャットダウン：処理を完了し、オフセットをコミットして、接続を閉じる
• 監視：コンシューマーのラグ、エラー率、スループットを追跡する

選択の指針

フレームワーク：メッセージブローカーの選択

START: What are requirements?

1. Need Kafka API compatibility?
   YES → Kafka or Redpanda
   NO → Continue

2. Is multi-tenancy critical?
   YES → Apache Pulsar
   NO → Continue

3. Operational simplicity priority?
   YES → Redpanda (single binary, no ZooKeeper)
   NO → Continue

4. Mature ecosystem needed?
   YES → Apache Kafka
   NO → Redpanda (better performance)

5. Task queues (not event streams)?
   YES → RabbitMQ or message-queues skill
   NO → Kafka/Redpanda/Pulsar

フレームワーク：ストリームプロセッサーの選択

START: What is latency requirement?

1. Millisecond-level latency needed?
   YES → Apache Flink
   NO → Continue

2. Batch + stream in same pipeline?
   YES → Apache Spark Streaming
   NO → Continue

3. Embedded in microservice?
   YES → Kafka Streams
   NO → Continue

4. SQL interface for analysts?
   YES → ksqlDB
   NO → Flink or Spark

5. Python primary language?
   YES → Spark (PySpark) or Faust
   NO → Flink (Java/Scala)

フレームワーク：言語の選択

TypeScript/Node.js：

• API ゲートウェイ、ウェブサービス、リアルタイムダッシュボード
• KafkaJS ライブラリ（コードスニペット 827 件、高い評価）

Python：

• データサイエンス、ML パイプライン、分析
• confluent-kafka-python（スニペット 192 件、スコア 68.8）

Go：

• 高性能マイクロサービス、インフラストラクチャツール
• kafka-go（42 件のスニペット、Go の慣用的な実装）

Java/Scala：

• エンタープライズアプリケーション、Kafka Streams、Flink、Spark
• Apache Kafka Java Client（683 件のスニペット、スコア 76.9）

高度なパターン

イベントソーシング

状態の変更を不変のイベントとして保存します。次の項目については、references/event-sourcing.md を参照してください。

• イベントストアの設計パターン
• イベントスキーマの進化
• スナップショット戦略
• 時系列クエリと監査証跡

変更データキャプチャ（CDC）

データベースの変更をイベントとしてキャプチャします。以下については、references/cdc-patterns.mdを参照してください。

• Debeziumの統合（MySQL、PostgreSQL、MongoDB）
• リアルタイムデータ同期
• マイクロサービスのデータ統合パターン

Exactly-Once処理

トランザクション保証を実装します。以下については、references/exactly-once.mdを参照してください。

• 冪等プロデューサー
• トランザクショナルコンシューマー
• エンドツーエンドのExactly-Onceパイプライン

エラー処理

本番環境に対応したエラー管理です。以下については、references/error-handling.mdを参照してください。

• デッドレターキューのパターン
• 指数バックオフによる再試行戦略
• バックプレッシャーの処理
• 下流の障害に対するサーキットブレーカー

リファレンスファイル

選定ガイド

• references/broker-selection.md - Kafka、Pulsar、Redpandaの比較
• references/processor-selection.md - Flink、Spark、Kafka Streamsの比較
• references/delivery-guarantees.md - At-least-once、Exactly-Onceのパターン

言語別の実装

• references/typescript-patterns.md - KafkaJSのパターン（プロデューサー、コンシューマー、エラー処理）
• references/python-patterns.md - confluent-kafka-pythonのパターン
• references/go-patterns.md - kafka-goのパターン
• references/java-patterns.md - Apache Kafka Javaクライアントのパターン

高度なトピック

• references/event-sourcing.md - イベントソーシングのアーキテクチャ
• references/cdc-patterns.md - Debeziumによる変更データキャプチャ
• references/exactly-once.md - トランザクショナル処理
• references/error-handling.md - DLQ、再試行、バックプレッシャー
• references/performance-tuning.md - スループットの最適化、パーティショニング戦略

検証スクリプト

トークンを使わずに検証と生成を行うには、これらのスクリプトを実行します。

Kafka設定の検証

python scripts/validate-kafka-config.py --config producer.yaml
python scripts/validate-kafka-config.py --config consumer.yaml

確認項目：ブローカーへの接続、設定の妥当性、シリアライズ形式

Schema Registryテンプレートの生成

python scripts/generate-schema.py --type avro --entity User
python scripts/generate-schema.py --type protobuf --entity Event

作成されるもの：Schema Registry用のAvro／Protobufスキーマ定義

スループットのベンチマーク

bash scripts/benchmark-throughput.sh --broker localhost:9092 --topic test

テスト項目：プロデューサー／コンシューマーのスループット、レイテンシのパーセンタイル

コード例

TypeScriptの例（KafkaJS）

examples/typescript/を参照してください。

• basic-producer.ts - エラー処理を備えたシンプルなイベントプロデューサー
• basic-consumer.ts - オフセットを手動でコミットするコンシューマー
• transactional-producer.ts - Exactly-Onceプロデューサーのパターン
• consumer-with-dlq.ts - デッドレターキューの実装

Pythonの例（confluent-kafka-python）

examples/python/を参照してください。

• basic_producer.py - 配信コールバックを備えたプロデューサー
• basic_consumer.py - エラー処理を備えたコンシューマー
• async_producer.py - AsyncIOプロデューサー（aiokafka）
• schema_registry.py - Schema Registryを使用したAvroシリアライズ

Goの例（kafka-go）

examples/go/を参照してください。

• basic_producer.go - Goらしいプロデューサー
• basic_consumer.go - 手動コミットを行うコンシューマー
• high_perf_consumer.go - 並行処理のパターン
• batch_producer.go - メッセージのバッチ送信

Javaの例（Apache Kafka）

examples/java/を参照してください。

• BasicProducer.java - 冪等性を備えたプロデューサー
• BasicConsumer.java - エラー復旧機能を備えたコンシューマー
• TransactionalProducer.java - Exactly-Onceトランザクション
• StreamsAggregation.java - Kafka Streamsによる集約

テクノロジーの比較

メッセージブローカーの比較

機能	Kafka	Pulsar	Redpanda	RabbitMQ
スループット	非常に高い	高い	非常に高い	中程度
レイテンシ	中程度	中程度	低い	低い
イベントの再生	対応	対応	対応	非対応
マルチテナンシー	手動	ネイティブ	手動	手動
運用の複雑さ	中程度	高い	低い	低い
最適な用途	エンタープライズ、ビッグデータ	SaaS、IoT	パフォーマンスが重要な用途	タスクキュー

ストリームプロセッサの比較

機能	Flink	Spark	Kafka Streams	ksqlDB
処理モデル	真のストリーミング	マイクロバッチ	ライブラリ	SQL エンジン
レイテンシ	ミリ秒	秒	ミリ秒	秒
デプロイ	クラスター	クラスター	組み込み	サーバー
最適な用途	リアルタイム分析	バッチ＋ストリーム	マイクロサービス	アナリスト

クライアントライブラリの推奨

言語	ライブラリ	信頼度スコア	スニペット	ユースケース
TypeScript	KafkaJS	高	827	Web サービス、API
Python	confluent-kafka-python	高（68.8）	192	データパイプライン、ML
Go	kafka-go	高	42	高性能サービス
Java	Kafka Java Client	高（76.9）	683	エンタープライズ、Flink/Spark

関連スキル

認証とセキュリティのパターンについては、auth-security スキルを参照してください。 インフラのデプロイ（Kubernetes オペレーター、Terraform）については、infrastructure-as-code スキルを参照してください。 監視メトリクスとトレースについては、observability スキルを参照してください。 API 設計パターンについては、api-design-principles スキルを参照してください。 データアーキテクチャとデータウェアハウジングについては、data-architecture スキルを参照してください。

トラブルシューティング

コンシューマーラグの問題

• パーティション数とコンシューマー数を確認します（並列処理のために一致させます）
• コンシューマーインスタンスを増やすか、処理時間を短縮します
• Kafka のコンシューマーラグメトリクスで監視します

メッセージの消失

• プロデューサーの acks=all 設定を確認します
• ブローカーのレプリケーション係数（>1）を確認します
• コンシューマーが処理後にオフセットをコミットすることを確認します

重複メッセージ

• 冪等性のあるコンシューマーを実装します（メッセージ ID を追跡します）
• exactly-once セマンティクスを使用します（トランザクション）
• at-least-once 配信を前提に設計します

パフォーマンスのボトルネック

• 並列処理のためにパーティション数を増やします
• バッチサイズと linger 時間を調整します
• 圧縮（GZIP、LZ4、Snappy）を有効にします
• 詳細は references/performance-tuning.md を参照してください