Confluentとは？Apache Kafkaとの違いについて

Confluentは、Apache Kafkaをベースに構築されたデータストリーミングプラットフォームです。Kafka単体には含まれていない、エンタープライズ向けツール、マネージドインフラストラクチャ、各種エコシステムとの統合機能を提供します。リアルタイムデータについて調べるときには、Apache KafkaとConfluentという2つの名前を目にすることでしょう。このブログでは、Kafkaが担う役割、ConfluentがKafkaに追加する機能、そして自社のインフラにどちらが適しているのかを判断するためのポイントを解説します。

Apache Kafkaとは？

Confluentについて説明する前に、まずApache Kafkaとは何かを整理しておきましょう。クラウドエンジニアやデータエンジニアにとって、Kafkaはリアルタイムアーキテクチャを支える基盤となりますが、まずは標準機能として何を提供しているかを見ていきます。

Apache Kafkaは、高性能なデータパイプライン、ストリーミング分析、データ統合、ミッションクリティカルなアプリケーション向けに、何千もの企業で利用されているオープンソースの分散型イベントストリーミングプラットフォームです。Kafkaの中核は、分散型かつ水平方向にスケーラブルで、耐障害性を持つコミットログとしての設計です。アプリケーションはイベントストリームを非同期にパブリッシュ（プロデュース）およびサブスクライブ（コンシューム）でき、それらのイベントはクラスター全体にわたって信頼性高く保存されます。

Apache Kafkaで標準提供される機能

オープンソース版の Apache Kafkaをダウンロードすると、次のコンポーネントを利用できます。

• ブローカークラスタ：トピック、パーティション、イベント複製を管理する、データの保存とデリバリーを担う中核エンジンです

• プロデューサーとコンシューマークライアントAPI：アプリケーションからイベントストリームの書き込みや読み取りを行うためのコアライブラリです

• Kafkaストリームライブラリ：クライアント側でストリーム処理アプリケーションを構築するための、Java/Scala向けネイティブライブラリです

• Kafka Connectフレームワーク：コンポーネント化されたコネクタプラグインアーキテクチャです（Kafkaが提供するのはフレームワークのみであり、実際のコネクタは通常、自分で入手、インストール、管理する必要があります）

• CLIツール：トピック管理、コンシューマーグループのオフセット確認、設定変更などを行うための基本的なコマンドラインツールです

• .ACLベースの基本的なセキュリティ：SASL、SSL、およびアクセスコントロールリスト（ACL）による基本的なアクセス制御を標準でサポートしており、トピックへのアクセスを制限できます

Kafkaは非常に強力なエンジンですが、本番環境で運用するには相応の運用負荷が伴います。オープンソース版のKafkaだけでは、インフラストラクチャのプロビジョニング、ゼロダウンタイムのアップグレード、柔軟な拡張、自動リバランス、スキーマ管理、データガバナンス、リージョン間複製などは提供されません。これらの機能は、自身で構築または運用する必要があります。

Confluentとは？

Confluentは、Apache Kafkaの開発者によって設立され、オープンソース版Kafkaが抱える運用面やエコシステム面の課題を解決するために開発されました。Confluentは、Apache Kafkaを基盤とし、その上にエンタープライズ向け機能、管理ツール、フルマネージドのクラウドインフラストラクチャを統合した商用版データストリーミングプラットフォームです。セキュリティ、監視、ガバナンス、統合のためのカスタムツールを、お客様のエンジニアリングチームが何か月もかけて独自に構築する必要はありません。Confluentは、本番環境ですぐに利用できる包括的なエコシステムを提供しています。Confluentは、セルフマネージド型ソフトウェアパッケージ（オンプレミス）である Confluent Platformと、フルマネージドのクラウドサービスであるConfluent Cloudの両方で提供されています 。

ConfluentとKafkaは同じ？

いいえ。 ConfluentはApache Kafkaを基盤としていますが、Kafkaと同じではありません。Kafkaは、オープンソースの分散型イベントストリーミングエンジンです。一方、ConfluentはKafkaに加え、スキーマ管理、コネクタ、ストリーム処理、セキュリティ、ガバナンス、運用管理など、エンタープライズグレードのツールを備えた商用プラットフォームです。Kafkaを高性能な「エンジン」だとすれば、Confluentは「完成車」に例えられます。エンジンだけでなく、高速道路を安全に走行するために必要な車体、ダッシュボード、タイヤ、安全装備まで備わっているというイメージしやすいかも知れません。

Confluent Cloudとは？オープンソース版Kafkaとの違い

Confluent Cloudは、フルマネージドクラウドの Kafkaサービスであり、お客様による物理的なKafkaブローカーのプロビジョニングや運用、拡張は不要です。オープンソース版 Kafkaでは、お客様のエンジニアが基盤となるインスタンスを管理し、手動でアップグレードを調整し、ブローカーのトラブルシューティングを行う必要があります。一方、Confluent Cloudでは、こうしたインフラストラクチャの管理は完全に抽象化されています。さらに、Schema Registry、120 種類以上の事前構築済みのコネクタ群、ストリーム処理向けのマネージドApache Flink、RBAC や包括的な監査ログなどの高度なエンタープライズ向けセキュリティ機能を、マネージドサービスとしてご利用いただけます。

Confluent Cloudを導入するときには、自社のワークロードに応じて複数の展開モデルと料金体系を選択できます。

• 展開と可用性：Amazon Web Services（AWS）、Google Cloud Platform（GCP）、Microsoft Azureでグローバルに利用可能です。これらのクラウドプロバイダー上で、サーバーレスモデルまたは専用インフラストラクチャを利用して、ネイティブにクラスタを立ち上げることができます。

• 料金体系：軽量なワークロード向けに、BasicおよびStandardクラスタで従量課金制を採用しています。大規模な本番環境向けには、Confluent Capacity Unit（CKU）に基づく専有キャパシティモデルへ移行できます。

• クラスタタイプ：

  • Basic: 開発、プロトタイピング、低スループットのアプリに適しています。サーバーレスで提供され、基本的な機能を利用できます。

  • Standard:標準機能に加え、マルチゾーンの可用性や Schema Registry が必要な本番ワークロード向けです。

  • Dedicated:プライベートネットワーク、予測可能なパフォーマンス、専用インフラストラクチャが必要となる高スループットのエンタープライズワークロード向けです。

  • Enterprise:複雑なアーキテクチャに対応するため、高度なガバナンス機能やデータ共有機能を提供します。

  • Freight: ログ、オブザーバビリティ、バッチパイプライン、AI/ML向けデータ取り込みなど、高スループットでレイテンシの影響を受けにくいワークロード向けに設計されています。これらは、低レイテンシよりも高スループットを重視した、コスト効率に優れたサーバーレスクラスタであり、セルフマネージドの環境と比べて最大 90% 高いスループット効率を実現します。

ConfluentがKafkaに追加する機能

Confluentがオープンソースのエコシステムをどのように拡張しているのかを理解するには、Kafkaブローカーに追加されるアーキテクチャレイヤーを見ると分かりやすいでしょう。

• Schema Registry：Avro、Protobuf、JSONに対応した厳格なデータコントラクトを適用し、プロデューサーがペイロードを任意に変更して下流のアプリケーションへ影響を及ぼすことを防ぎます。オープンソース版Kafkaにはこの機能がないため、気付かないうちにデータが破損するリスクがあります。

• Kafka Connect：SnowflakeやAmazon S3など、120種類以上の事前構築済みマネージドクラウドコネクタを提供し、外部データストアとのシームレスな連携ができます。オープンソース版Kafkaが提供するのはフレームワークのみであり、クラスタやJARファイルはユーザーが手動で管理する必要があります。

• ストリーム処理：Confluentは、フルマネージドのApache FlinkとksqlDBを統合しており、標準SQLを使用してリアルタイムストリーム処理ができます。一方、オープンソース版KafkaではKafkaストリームライブラリを利用し、JavaまたはScalaで独自のマイクロサービスを構築して運用する必要があります。

• ガバナンスとオブザーバビリティ：Confluentには、ストリームカタログ、エンドツーエンドのデータリネージ、品質ルールが組み込まれており、大規模で複雑な環境の運用を支援します。オープンソース版Kafkaには、データマッピングやデータカタログ機能は標準では備わっていません。

• エンタープライズ向けセキュリティ：Confluentでは、きめ細かなロールベースアクセス制御（RBAC）、構造化された監査ログ、VPCピアリングや PrivateLinkなどのプライベートネットワーク機能を利用できます。オープンソース版Kafkaが標準で提供するのは、基本的なACLとSSL/SASLによる暗号化のみです。

• マルチリージョンとディザスタリカバリ：ConfluentはCluster Linkingを利用して、外部ワーカーを使用することなく、リージョン間でトピックをネイティブにミラーリングし、メッセージオフセットも保持できます。一方、オープンソース版KafkaではMirrorMaker 2を使用しますが、別途クラスタを展開し、監視する必要があります。

注：どの選択肢が最適かは、チームの運用体制、予算、そしてアーキテクチャの成熟度によって異なります。

Confluent Kafkaのどちらを選ぶべきか？

試験的な利用や個人用プロジェクトの構築、あるいは1つの開発チーム内で少数のトピックを運用する程度であれば、オープンソース版Kafkaだけで十分な場合もあります。一方、組織として、データコントラクトの自動適用、標準で利用できるシステム統合、高度なセキュリティ監査機能が必要な場合や、分散システムの運用負荷そのものを削減したい場合には、Confluentが適しています。

データストリーミングプラットフォームは必要か？ それともKafkaだけで十分か？

リアルタイムイベントが、単一の機能を支える仕組みから、組織全体を支える基盤へと発展したときには、Kafkaだけでなく、データストリーミングプラットフォームが必要になります。複数の独立したチームが、安全にイベントをパブリッシュ（プロデュース）およびサブスクライブ（コンシューム）し、スキーマによってデータ形式を検証し、レガシーデータベースから動的にデータを取り込み、さらに独自のマイクロサービスを構築することなく、データをリアルタイムに変換する必要がある場合、Kafkaだけでは運用上のボトルネックが生じます。Kafkaは基盤を提供します。一方、データストリーミングプラットフォームは、その基盤をエンタープライズ規模でも現実的かつ効率的に運用できるようにするのです。

はじめに - Confluent CloudでKafkaをPythonから利用する

このチュートリアルでは、軽量なPythonのプロデューサーとコンシューマーをConfluent Cloudクラスタに接続します。それぞれのスクリプトは 20 行未満のシンプルなコードで構成されており、10分もあればクラウドクラスタ上でイベントをやり取りできるようになります。

前提条件

コードを書き始める前に、開発環境が正しく構成されていることを確認してください。次の手順を実行し、必要な環境がすべて準備できていることを確認します。

1. Python 3.8 以降がシステムにインストールされていること次のコマンドで確認します

python3 --version

2. Confluent Cloudアカウント（無料クレジット付きのFreeクラスタを使用できます）

3. 利用可能なConfluent Cloudクラスタ（このチュートリアルでは、Basic クラスタで問題なく動作します）

4. Confluent Cloudコンソールで、このクラスタ専用に生成されたAPIキーとシークレットのペア

5. Confluent公式のPythonクライアントのインストール

pip install confluent-kafka

6. インストールの確認

python3 -c "import confluent_kafka; print(confluent_kafka.version())"

クライアントに関する重要な注 ：confluent-kafka は、Confluent が公式に保守している Pythonクライアントです。高性能なCライブラリlibrdkafkaを基盤として最適化されています。一方、kafka-pythonはコミュニティによって開発されたレガシーライブラリであり、API も大きく異なります。両者を混同しないよう注意してください。

設定

プロデューサーとコンシューマーはどちらも、TLSを使用してConfluent Cloudとの認証を処理するため、共通の基本設定を使用します。

config = {
    "bootstrap.servers": "<BOOTSTRAP_SERVER>",
    "security.protocol": "SASL_SSL",
    "sasl.mechanisms": "PLAIN",
    "sasl.username": "<API_KEY>",
    "sasl.password": "<API_SECRET>",
}

設定： <BOOTSTRAP_SERVER>, <API_KEY>, and <API_SECRET> は、利用しているクラスタの実際の値に置き換えてください。これらの情報は、Confluent Cloud コンソールの「クラスタ設定」 → 「エンドポイントおよび API キー」から確認できます

プロデューサー — 20 行以内でイベントを送信する

producer.py という名前のファイルを作成します。このスクリプトでは、プロデューサーを作成し、非同期の配信確認コールバックを定義したうえで、10 件のサンプルイベントをトピックへ送信します。

from confluent_kafka import Producer
config = {
    "bootstrap.servers": "<BOOTSTRAP_SERVER>",
    "security.protocol": "SASL_SSL",
    "sasl.mechanisms": "PLAIN",
    "sasl.username": "<API_KEY>",
    "sasl.password": "<API_SECRET>",
}
producer = Producer(config)
def delivery_report(err, msg):
    if err:
        print(f"Delivery failed: {err}")
    else:
        print(
            f"Delivered to {msg.topic()} [{msg.partition()}] @ {msg.offset()}"
        )
for i in range(10):
    producer.produce(
        "my-topic", key=str(i), value=f"event-{i}", callback=delivery_report
    )
    producer.poll(0)
producer.flush()

コードの解説：

• producer.produce(...): メッセージを内部の高性能キューに格納し、まとめてバックグラウンドでブローカーへ送信します。

• producer.poll(0): 定期的なハートビートとして機能するノンブロッキング呼び出しです。イベントを確認し、メッセージがクラスタで受信確認されると、delivery_report コールバックをすぐに実行します。

• producer.flush(): ブロッキング呼び出しです。ローカルバッファ内で待機しているすべてのメッセージが正常に送信され、受信確認が返されることを保証してから、スクリプトを終了します。

コンシューマー - 20行以内でイベントを読み取る

次は、consumer.pyという名前のファイルを作成して、トピックからこれらのイベントを取得します。

from confluent_kafka import Consumer
config = {
    "bootstrap.servers": "<BOOTSTRAP_SERVER>",
    "security.protocol": "SASL_SSL",
    "sasl.mechanisms": "PLAIN",
    "sasl.username": "<API_KEY>",
    "sasl.password": "<API_SECRET>",
    "group.id": "my-group",
    "auto.offset.reset": "earliest",
}
consumer = Consumer(config)
consumer.subscribe(["my-topic"])
try:
    while True:
        msg = consumer.poll(1.0)
        if msg is None:
            continue
        if msg.error():
            print(f"Error: {msg.error()}")
        else:
            print(f"{msg.key().decode()}: {msg.value().decode()}")
finally:
    consumer.close()

コードの解説：

• group.id: コンシューマーインスタンスを、指定したコンシューマーグループに参加させます。これにより、Kafka はコミット済みのコンシューマーオフセットを追跡し、パーティションの負荷を自動的に分散できます。

• auto.offset.reset: earliest: このコンシューマーグループに対して以前保存されたオフセットが存在しない場合、トピックパーティションのログの先頭から読み取りを開始するようコンシューマーに指示します。

• consumer.close(): シャットダウン時にコンシューマーがコンシューマーグループから正常に離れるようにします。また、保留中のメッセージオフセットを安全にコミットしながら、パーティションの再割り当てを即座に実行します。

よくあるエラーと解決方法

プロデューサーとコンシューマーを作成した後の次のステップ

基本的なPythonデータパイプラインが動作するようになったら、次は、スタンドアロンのブローカーにはない、包括的なストリーミングプラットフォームのエンタープライズ向け機能を試してみましょう。

Schema Registryを追加してデータコントラクトを実現する

現在のプロデューサーは単純な文字列を使用しています。実際の本番環境では、パイプラインの安全性を維持するために、構造化データの検証（Avro、Protobuf、JSON Schema）が必要になります。下流サービスを不正なペイロードから保護するために、Confluentの Schema RegistryをPythonから実装する方法を習得してください。

Kafka Connectで外部システムと接続する

独自の統合コードを作成することなく、稼働中のデータベースから直接データを取り込んだり、トピックイベントを分析用クラウドデータウェアハウスへ自動的にストリーミングしたりすることができます。Confluent Cloud 内で、フルマネージドのDebezium PostgreSQL CDC ソースコネクタをプロビジョニングする方法を確認してください。

Apache Flinkでストリームを処理する

シンプルな SQL クエリを使用して、リアルタイムで移動するメッセージストリームをその場でクリーンアップ、変換、結合、集約できます。Confluent Cloudでは、フルマネージドのスケーラブルなApache Flinkランタイムを提供しており、Webコンソールから最初の Flink SQLデータ変換スクリプトを直接作成できます。