Apache Kafkaストリーム処理実践：100万TPSリアルタイムデータパイプラインを構築する5つのコアパターン

Apache Kafkaストリーム処理：リアルタイムデータパイプラインの礎

マイクロサービスのデータサイロ、バッチ処理の高遅延、リアルタイム分析の困難さ——エンタープライズデータインフラが直面するコア課題。Apache Kafkaは分散イベントストリーミングプラットフォームとして、100万TPスループット、ミリ秒レイテンシ、Exactly-Onceセマンティクスでリアルタイムデータパイプラインのデファクトスタンダードとなっています。2026年、Kafkaストリーム処理は金融取引、IoTテレメトリ、ユーザー行動分析などのシナリオで大規模展開されています。

本記事では5つのコアパターンを通じて、Producer/Consumer→Kafka Streams→Connect→Schema Registry→プロダクションチューニングのフルチェーン実践を解説します。

コア概念

概念	説明
Kafka	分散イベントストリーミングプラットフォーム
Topic	メッセージカテゴリ、論理パーティション集合
Partition	Topicの物理シャード、並行度ユニット
Consumer Group	コンシューマーグループ、負荷分散を実現
Kafka Streams	Kafkaネイティブストリーム処理ライブラリ
Kafka Connect	データコネクタフレームワーク
Schema Registry	Avro/Protobuf Schema管理サービス
Exactly-Once	正確に一度セマンティクス、重複処理を回避

問題分析：Kafkaストリーム処理の5つの課題

パーティション戦略の選択：誤ったパーティショニングでデータスキューが発生
Consumer Rebalance：コンシューマーグループのリバランスで消費が一時停止
Exactly-Once実装：エンドツーエンドの正確に一度セマンティクス設定が複雑
Schema進化：Avro Schema変更の互換性管理
監視とアラート：Lag監視とコンシューマー遅延警告

ステップバイステップ：5つのKafkaストリーム処理パターン

パターン1：高スループットProducerとConsumer

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");
props.put(ProducerConfig.ACKS_CONFIG, "all");
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");

KafkaProducer<String, String> producer = new KafkaProducer<>(props);
for (int i = 0; i < 1_000_000; i++) {
    producer.send(new ProducerRecord<>("orders", String.valueOf(i % 64), data));
}
producer.flush();

パターン2：Kafka Streamsトポロジーとウィンドウ集約

StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> orders = builder.stream("orders");

KTable<Windowed<String>, Long> orderCounts = orders
    .groupBy((key, value) -> extractUserId(value))
    .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)))
    .count(Materialized.as("order-counts"));

orderCounts.toStream().to("order-counts-output");

Properties streamsProps = new Properties();
streamsProps.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, "exactly_once_v2");
KafkaStreams streams = new KafkaStreams(builder.build(), streamsProps);
streams.start();

パターン3：Kafka Connectデータパイプライン

{
  "name": "postgres-source",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres",
    "database.dbname": "orders_db",
    "table.include.list": "public.orders,public.customers"
  }
}

パターン4：Schema RegistryとAvroシリアライゼーション

Properties avroProps = new Properties();
avroProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class.getName());
avroProps.put("schema.registry.url", "http://schema-registry:8081");

Schema orderSchema = SchemaBuilder.record("Order")
    .fields().requiredLong("orderId").requiredString("userId").requiredDouble("amount").endRecord();

GenericRecord order = new GenericData.Record(orderSchema);
order.put("orderId", 12345L);
order.put("userId", "user-001");
order.put("amount", 99.99);

パターン5：プロダクション監視とチューニング

kafka-consumer-groups --bootstrap-server kafka:9092 \
  --describe --group order-processor

落とし穴ガイド

落とし穴1：パーティションキーによるデータスキュー

// ❌ 間違い：タイムスタンプをパーティションキーに使用
producer.send(new ProducerRecord<>("orders", String.valueOf(System.currentTimeMillis()), data));

// ✅ 正しい：ビジネスキーのハッシュパーティショニング
producer.send(new ProducerRecord<>("orders", orderId, data));

落とし穴2：Consumerの自動コミットOffset

// ❌ 間違い：自動コミット、メッセージ損失の可能性
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");

// ✅ 正しい：処理完了後の手動コミット
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
consumer.commitSync();

落とし穴3：Rebalanceの未処理

// ✅ 正しい：Rebalanceリスナーを登録
consumer.subscribe(List.of("orders"), new ConsumerRebalanceListener() {
    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        consumer.commitSync();
    }
});

落とし穴4：非互換Schema変更

// ✅ 正しい：デフォルト値付きの新フィールド追加で互換性維持
Schema newSchema = SchemaBuilder.record("Order")
    .fields().requiredLong("orderId").optionalString("newField").endRecord();

落とし穴5：Exactly-Onceの未設定

// ✅ 正しい：Exactly-Once v2を有効化
streamsProps.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, "exactly_once_v2");

エラートラブルシューティング

#	エラー	原因	解決方法
1	`NotLeaderOrFollowerException`	パーティションLeader切り替え	Producer自動リトライ、Broker健全性を確認
2	`CommitFailedException`	RebalanceによるOffsetコミット失敗	max.poll.interval.msを削減
3	`SerializationException`	Schema不一致	Schema Registry互換性を確認
4	`TimeoutException`	リクエストタイムアウト	request.timeout.msを増加
5	`RecordTooLargeException`	メッセージがmax.message.bytes超過	Brokerのmessage.max.bytesを増加
6	`GroupAuthorizationException`	コンシューマーグループ権限不足	ACL設定を確認
7	`TopicExistsException`	Topicが既に存在	--if-not-existsを使用
8	`WakeupException`	Consumerが起動されてシャットダウン	正常シャットダウン手順
9	`RebalanceInProgressException`	リバランス進行中	リバランス完了を待機
10	`SchemaRegistryException`	Schema Registryに到達不能	URLとネットワークを確認

高度な最適化

階層型ストレージTiered Storage：コールドデータをS3に自動移行、Brokerストレージコストを削減
KRaftモード：ZooKeeper依存を排除、運用を簡素化
冪等Producer+トランザクション：エンドツーエンドExactly-Onceセマンティクス
Consumer Lag自動スケーリング：Lagに基づきConsumerインスタンス数を動的調整
MirrorMaker 2クラスタ間レプリケーション：マルチリージョン災害復旧と移行

比較分析

次元	Kafka	Pulsar	RabbitMQ	Redis Streams
スループット	⭐⭐⭐⭐⭐	⭐⭐⭐⭐⭐	⭐⭐⭐	⭐⭐⭐⭐
レイテンシ	⭐⭐⭐⭐	⭐⭐⭐⭐	⭐⭐⭐⭐⭐	⭐⭐⭐⭐⭐
Exactly-Once	⭐⭐⭐⭐⭐	⭐⭐⭐⭐	⭐⭐	⭐⭐
ストリーム処理	⭐⭐⭐⭐⭐	⭐⭐⭐⭐	⭐⭐	⭐⭐
エコシステム	⭐⭐⭐⭐⭐	⭐⭐⭐	⭐⭐⭐⭐	⭐⭐⭐
運用複雑さ	⭐⭐⭐	⭐⭐⭐	⭐⭐⭐⭐⭐	⭐⭐⭐⭐⭐

まとめ：Apache Kafkaストリーム処理は100万TPSスループットとExactly-Onceセマンティクスにより、リアルタイムデータパイプラインのデファクトスタンダードです。Kafkaは高スループットのリアルタイムデータ処理が必要なシナリオに適しており、特に金融取引、IoTテレメトリ、ユーザー行動分析に最適です。2026年、KRaftモードの成熟と階層型ストレージの普及により、Kafkaの運用コストは継続的に低下しています。

オンラインツール推薦

JSONフォーマッター：/ja/json/format
Hash計算：/ja/encode/hash
cURL to Code：/ja/dev/curl-to-code