mcpField
K
kafka

kafka

github.com/alphaonedevデータ・データベース50 回インストール
リポジトリ

kafka について

kafka は、データエンジニアリングやアプリケーション開発向けに Apache Kafka を紹介するスキルです。トピックベースのメッセージング、コンシューマーグループ、トランザクション配信、Kafka Streams によるストリーム処理を扱います。

主な機能

  1. 01分散トピックとパーティション
  2. 02トランザクションによる厳密に1回の処理
  3. 03メッセージ保持ポリシー
  4. 04負荷分散用コンシューマーグループ
  5. 05Kafka Streams による状態処理

活用シーン

  1. 01リアルタイムデータパイプラインの構築
  2. 02ストリームによるログ集約
  3. 03イベントを使ったマイクロサービス連携
  4. 04Kafka Streams によるイベントストリーム変換

インストール

コマンドをコピーしてターミナルで実行します。

$npx skills add https://github.com/alphaonedev/openclaw-graph --skill kafka

インストール後、Claude Code で/kafkaを入力して読み込まれたか確認します。

詳細

kafka

目的

Apache Kafka は、リアルタイムのデータパイプラインやストリーミングアプリを構築するための分散イベントストリーミングプラットフォームです。高スループットでフォールトトレラントなメッセージングを実現します。

使用する場面

ログ集約、イベント駆動アーキテクチャ、マイクロサービス間の通信など、リアルタイムのデータ取り込みと処理が必要なシナリオでは Kafka を使用します。RabbitMQ のような軽量なツールで十分な単純なキューイングには使用しないでください。

主な機能

  • スケーラビリティと耐久性を実現するため、トピック、パーティション、レプリカを使用した分散ストリーミングをサポートします。
  • トランザクション API による厳密に1回のセマンティクスを提供し、データの損失や重複を防ぎます。
  • ブローカー設定で log.retention.hours=168 を指定してメッセージを7日間保持するなど、設定可能な保持ポリシーで大量のデータを処理します。
  • 複数のコンシューマーがグループ ID を共有してトピックの消費を分担する、負荷分散用のコンシューマーグループを提供します。
  • Kafka Streams API によるストリーム処理と、KTable オブジェクトを使ったイベントの集計などのステートフル変換を統合します。

使用パターン

メッセージを生成するには、まずトピックを作成してからプロデューサークライアントを使用します。メッセージを消費するには、トピックをサブスクライブし、ループ内でメッセージを処理します。再処理を避けるため、オフセットは必ず手動または自動コミットで処理してください。バッチ処理では、Kafka Connect を使ってデータベースなどのソースからデータを取り込みます。パターン：プロデューラー設定で enable.idempotence=true を指定し、冪等性プロデューサーを使用して少なくとも1回の配信を実現します。

主なコマンド／API

手早く操作するには、Kafka CLI を使用します。

  • トピックを作成：kafka-topics.sh --create --topic mytopic --bootstrap-server localhost:9092 --partitions 3 --replication-factor 2
  • メッセージを生成：kafka-console-producer.sh --topic mytopic --bootstrap-server localhost:9092（メッセージを入力し、Ctrl+D を押して送信します）
  • メッセージを消費：kafka-console-consumer.sh --topic mytopic --from-beginning --bootstrap-server localhost:9092 --group mygroup Java での API の使用方法：
  • プロデューサーの例： 
    Properties props = new Properties(); props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.send(new ProducerRecord<>("mytopic", "key", "value"));
  • コンシューマーの例： 
    Properties props = new Properties(); props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "mygroup"); props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("mytopic"));
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));

環境変数を使って SASL で認証します。$KAFKA_SASL_JAAS_CONFIG に "org.apache.kafka.common.security.plain.PlainLoginModule required username='$SERVICE_API_KEY';" を設定します。

統合に関する注意事項

Kafka Connect を使って、データベース用の JDBC コネクターなどを通じて、Kafka を他のシステムのソース／シンクと統合します。認証には、$KAFKA_CLIENT_API_KEY などの環境変数に保存したキーを使用して、SSL または SASL を利用します。Spark と連携する場合は、spark.kafka.bootstrap.servers を指定して Spark Streaming を設定し、spark-sql-kafka-0-10_2.12 などの依存関係を追加します。マイクロサービスでは、プロデューサーがトピックにイベントを送信し、コンシューマーが Webhook 経由で反応するバックボーンとして Kafka を使用します。必ず正確なバージョンを指定してください。たとえば、Kafka 3.4.0 と、エンドポイント http://localhost:8081/subjects の Confluent Schema Registry を使用します。

エラー処理

接続障害などの一般的なエラーには、ブローカーの稼働状況を確認し、指数バックオフで再試行して対処します。たとえば、コード内で producer.send() を try-catch で囲み、最大3回再試行します。オフセットが範囲外の場合は、コンシューマー設定で auto.offset.reset=earliest を使用し、最初から処理を開始します。認証エラー（例：401 Unauthorized）が発生した場合は、$SERVICE_API_KEY などの環境変数を確認し、SASL のメカニズムが一致していることを確認します。詳細を含めてエラーを記録します。例：Java の場合は catch (KafkaException e) { log.error("Kafka error: {}", e.getMessage()); }。レプリカを監視し、min.insync.replicas=2 を設定して確認応答を必須にすることで、ブローカーのクラッシュに対処します。

グラフの関係

  • 所属クラスター：data-engineering
  • 関連タグ：event-streaming、data-pipelines、kafka
  • 関連候補：データ処理パイプライン向けに、spark や hadoop など、data-engineering クラスター内のスキルと統合できます。

コメント

Markdown 対応