kafka-engineer

Kafka エンジニア

目的

スケーラブルなイベント駆動型アーキテクチャとリアルタイムデータパイプラインを専門とする、Apache Kafka とイベントストリーミングの専門知識を提供します。厳密に1回だけの処理、Kafka Connect、Schema Registry の管理に対応した、フォールトトレラントなストリーミングプラットフォームを構築します。

使用する場面

• イベント駆動型マイクロサービスアーキテクチャの設計
• Kafka Connect パイプライン（CDC、S3 Sink）のセットアップ
• ストリーム処理アプリの作成（Kafka Streams／ksqlDB）
• コンシューマーラグ、再調整の頻発、ブローカーのパフォーマンス問題のデバッグ
• Schema Registry を使ったスキーマ（Avro／Protobuf）の設計
• ACL と mTLS セキュリティの設定

---

---

2. 判断フレームワーク

アーキテクチャの選定

What is the use case?
│
├─ **Data Integration (ETL)**
│  ├─ DB to DB/Data Lake? → **Kafka Connect** (Zero code)
│  └─ Complex transformations? → **Kafka Streams**
│
├─ **Real-Time Analytics**
│  ├─ SQL-like queries? → **ksqlDB** (Quick aggregation)
│  └─ Complex stateful logic? → **Kafka Streams / Flink**
│
└─ **Microservices Comm**
   ├─ Event Notification? → **Standard Producer/Consumer**
   └─ Event Sourcing? → **State Stores (RocksDB)**

設定の調整（「重要な3項目」）

1. スループット： batch.size、linger.ms、compression.type=lz4。
2. レイテンシ： linger.ms=0、acks=1。
3. 耐久性： acks=all、min.insync.replicas=2、replication.factor=3。

危険信号 → sre-engineer にエスカレーションしてください：

• 「Unclean leader election」が有効（データ損失のリスク）
• 新しいクラスターで Zookeeper に依存（KRaft モードを使用）
• ブローカーのディスク使用率が80％超
• コンシューマーラグが常に増加（容量の不一致）

---

---

3. 主要なワークフロー

ワークフロー1：Kafka Connect（CDC）

目標： PostgreSQL から S3 に変更をストリーミングします。

手順：

1. ソース設定（postgres-source.json）

   {
     "name": "postgres-source",
     "config": {
       "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
       "database.hostname": "db-host",
       "database.dbname": "mydb",
       "database.user": "kafka",
       "plugin.name": "pgoutput"
     }
   }
   

2. シンク設定（s3-sink.json）

   {
     "name": "s3-sink",
     "config": {
       "connector.class": "io.confluent.connect.s3.S3SinkConnector",
       "s3.bucket.name": "my-datalake",
       "format.class": "io.confluent.connect.s3.format.parquet.ParquetFormat",
       "flush.size": "1000"
     }
   }
   

3. デプロイ

   • curl -X POST -d @postgres-source.json http://connect:8083/connectors

---

---

ワークフロー3：Schema Registry との統合

目標： スキーマの互換性を強制します。

手順：

1. スキーマの定義（user.avsc）

   {
     "type": "record",
     "name": "User",
     "fields": [
       {"name": "id", "type": "int"},
       {"name": "name", "type": "string"}
     ]
   }
   

2. プロデューサー（Java）

   • KafkaAvroSerializer を使用します。
   • レジストリ URL：http://schema-registry:8081。

---

---

5. アンチパターンと注意点

❌ アンチパターン1：大きなメッセージ

例：

• Kafka メッセージで10MBの画像ペイロードを送信する。

問題点：

• Kafka は小さなメッセージ（1MB未満）向けに最適化されています。大きなメッセージはブローカーのスレッドを塞ぎます。

適切な方法：

• 画像を S3 に保存します。
• Kafka メッセージには 参照 URL を送信します。

❌ アンチパターン2：パーティションが多すぎる

例：

• 小規模なクラスターに10,000個のパーティションを作成する。

問題点：

• リーダー選出が遅くなる（Zookeeper のオーバーヘッド）。
• ファイルハンドルの使用量が増加する。

適切な方法：

• ブローカーあたりのパーティション数を約4,000個に制限します。トピック数を減らすか、より大きなクラスターを使用します。

❌ アンチパターン3：ブロッキングコンシューマー

例：

• コンシューマーが各メッセージに対して時間のかかる HTTP 呼び出し（30秒）を実行する。

問題点：

• タイムアウトによりコンシューマーがグループから離脱し、再調整が頻発します。

適切な方法：

• 非同期処理： 処理をスレッドプールに移します。
• 一時停止／再開： バッファーが満杯の場合は consumer.pause() を使います。

---

---

7. 品質チェックリスト

設定：

• レプリケーション： 本番環境では係数を3にします。
• Min.ISR： 2（データ損失を防止）。
• 保持期間： 正しく設定します（時間またはサイズ）。

可観測性：

• ラグ： コンシューマーラグを監視します（Burrow／Prometheus）。
• レプリカ不足： レプリケーション不足のパーティション（1個以上）に対してアラートを設定します。
• JMX： メトリクスをエクスポートします。

例

例 1：リアルタイム不正検知パイプライン

シナリオ： ある金融サービス企業では、Kafka のストリーミングを使用したリアルタイムの不正検知が必要です。

アーキテクチャの実装：

1. イベント取り込み：PostgreSQL のトランザクションデータベースから Kafka Connect CDC を使用
2. ストリーム処理：リアルタイムのパターン検知を行う Kafka Streams アプリケーション
3. アラートシステム：アラートトピックへのプロデューサーが通知をトリガー
4. ストレージ：履歴分析とコンプライアンス対応のための S3 シンク

パイプライン設定：

コンポーネント	設定	目的
トピック	3（transactions、alerts、enriched）	データの整理
パーティション	12（ブローカー 3 台 × 4）	並列処理
レプリケーション	3	高可用性
圧縮	LZ4	スループットの最適化

主なロジック：

• 取引頻度のパターンを検知します（1 分間に 5 件以上の取引）
• 地理的な異常を特定します（不可能な移動）
• 高リスクの加盟店カテゴリにフラグを付けます

結果：

• 不正の 99.7% を 100ms 未満で検知
• 誤検知率が 5% から 0.3% に低下
• 指摘事項ゼロでコンプライアンス監査に合格

例 2：E コマース注文処理システム

シナリオ： 高い信頼性を実現するため、Kafka を使用して堅牢な注文処理システムを構築します。

システム設計：

1. 注文イベント：注文ライフサイクルイベント用のトピック
2. 在庫サービス：注文を消費して在庫を更新
3. 決済サービス：決済を処理して結果を公開
4. 通知サービス：メール／SMS で確認通知を送信

レジリエンスパターン：

• 処理に失敗した場合のデッドレターキュー
• 厳密に 1 回だけのセマンティクスを実現する冪等プロデューサー
• 手動でオフセットを管理するコンシューマーグループ
• 指数バックオフによる再試行

設定：

# Producer Configuration
acks: all
retries: 3
enable.idempotence: true

# Consumer Configuration
auto.offset.reset: earliest
enable.auto.commit: false
max.poll.records: 500

結果：

• メッセージ配信信頼性 99.99%
• 6 か月間、注文の重複はゼロ
• 最大処理能力：毎秒 10,000 件の注文

例 3：IoT テレメトリープラットフォーム

シナリオ： Kafka を使用して、数百万件の IoT デバイスのテレメトリーメッセージを処理します。

プラットフォームアーキテクチャ：

1. デバイスゲートウェイ：MQTT から Kafka へのプロキシ
2. データエンリッチメント：ストリーム処理でデバイスのメタデータを追加
3. 時系列ストレージ：device_id/date でパーティション分割した S3 シンク
4. リアルタイムアラート：異常を検知するしきい値ベースのアラート

スケーラビリティ設定：

• 並列処理用に 50 パーティション
• コスト最適化のため圧縮を有効化
• 保持期間：ホットデータは 7 日間、S3 のコールドデータは 1 年間
• データコントラクト用の Schema Registry

パフォーマンス指標：

指標	値
スループット	毎秒 500,000 メッセージ
レイテンシー（P99）	50ms
コンシューマーラグ	1 秒未満
ストレージ効率	圧縮により 60% 削減

ベストプラクティス

トピック設計

• 命名規則：明確で階層的なトピック名を使用します（domain.entity.event）
• パーティション戦略：将来の拡張を見込んで計画します（予想スループットの 3 倍）
• 保持ポリシー：保持期間をビジネス要件に合わせます
• クリーンアップポリシー：時間ベースの場合は delete、状態データの場合は compact を使用します
• スキーマ管理：Schema Registry を通じてスキーマを適用します

プロデューサーの最適化

• バッチ処理：スループット向上のため、batch.size と linger.ms を増やします
• 圧縮：速度とサイズのバランスを取るため、LZ4 を使用します
• Acks 設定：信頼性を重視する場合は all、レイテンシーを重視する場合は 1 を使用します
• 再試行戦略：バックオフを伴う再試行を実装します
• 冪等性：重要な処理経路で厳密に 1 回だけのセマンティクスを実現するため、有効にします

コンシューマーのベストプラクティス

• オフセット管理：重要な処理には手動コミットを使用します
• バッチ処理：効率化のため、max.poll.records を増やします
• リバランス処理：グレースフルシャットダウンを実装します
• エラー処理：処理不能なメッセージにはデッドレターキューを使用します
• モニタリング：コンシューマーラグと処理時間を追跡します

セキュリティ設定

• 暗号化：すべてのクライアントとブローカー間の通信に TLS を使用します
• 認証：本番環境では SASL/SCRAM または mTLS を使用します
• 認可：最小権限の原則に基づく ACL を使用します
• クォータ：不正利用を防ぐため、クライアントクォータを設定します
• 監査ログ：すべてのアクセスと設定変更を記録します

パフォーマンスのチューニング

• ブローカー設定：ワークロードの種類（スループット対レイテンシ）に合わせて最適化します
• JVM のチューニング：ヒープサイズとガベージコレクターを選択します
• OS のチューニング：ファイルディスクリプターの上限やネットワーク設定を調整します
• 監視：スループット、レイテンシ、エラーのメトリクスを収集します
• キャパシティプランニング：定期的に見直し、スケーリングの必要性を評価します

セキュリティ：

• 暗号化： クライアントとブローカー間、およびブローカー間で TLS を有効にします。
• 認証： SASL/SCRAM または mTLS を有効にします。
• ACL： 最小権限の原則を適用します（トピックの読み取り／書き込み）。