なぜ Kafka を諦めて Pulsar を選んだのでしょうか?

最近、私は Pulsar と Kafka の比較を調べています。ちょっと検索してみると、これら 2 つの最も有名なオープンソースメッセージングシステム間で進行中の「戦争」がわかります。

[[379649]]

画像はPexelsより

Kafka ユーザーとして、私は Kafka のいくつかの問題について本当に困惑していましたが、Pulsar は目を見張るものがあり、とても興奮しました。それで最終的に、私は時間をかけて背景情報を入手し、多くの調査を行うことができました。

この記事では、Pulsar の利点を強調し、Pulsar が Kafka よりも優れている理由を説明します。さあ始めましょう！

Kafka の基礎

Kafka はメッセージングシステムの王様です。これは 2011 年に LinkedIn によって作成され、Confluent のサポートにより広く配布されるようになりました。

Confluent は、スキーマ進化のための Schema Registry、他のデータソースからの簡単なストリーミングのための Kafka Connect など、オープンソースコミュニティに多くの新機能とアドオンをリリースしました。

データベースから Kafka、分散ストリーム処理用の Kafka Streams、最近では Kafka トピックに対する SQL のようなクエリ用の KSQL など。

Kafka は高速で、インストールが簡単で、非常に人気があり、幅広いユースケースに使用できます。 Apache Kafka は開発者の観点からは常に使いやすいものでしたが、運用に関しては混乱を招いていました。

それでは、Kafka の問題点をいくつか見てみましょう。

カフカデモ[2]

Kakfa の多くの問題点は次のとおりです。

ブローカーと保存されるデータの結合されたアーキテクチャ構造のため、Kafka のスケーリングは困難です。ブローカーを分割すると、トピックのパーティションとレプリカを複製する必要があり、時間がかかります。
テナントが完全に分離されたネイティブのマルチテナントはありません。
ストレージは非常に高価になる可能性があり、データを長期間保存できるにもかかわらず、コストの問題によりほとんど使用されません。
レプリカが同期しなくなった場合、メッセージが失われる可能性があります。
スケーリングの問題を回避するには、ブローカー、トピック、パーティション、レプリカの数を事前に計画して計算する必要があります (将来の使用量の増加を確実に計画する)。これは非常に困難です。
メッセージングシステムのみが必要な場合、オフセットの使用は複雑になる可能性があります。
クラスターの再バランス調整は、接続されたプロデューサーとコンシューマーのパフォーマンスに影響を与える可能性があります。
MirrorMaker[3]のGeoレプリケーションメカニズムに問題があります。 Uber のような企業は、これらの問題を克服するために独自のソリューションを生み出しました。

ご覧のとおり、質問のほとんどは運用面に関するものです。 Kafka はインストールが比較的簡単ですが、管理と調整が難しい場合があります。さらに、本来備わっているべき柔軟性と弾力性が欠けています。

パルサーの基礎

Pulsar は Yahoo! によって作成されました。 2013 年に開発され、2016 年に Apache Foundation に寄贈されました。現在、Pulsar は Apache Software Foundation のトップレベルプロジェクトとなっています。

すでに Yahoo!、Verizon、Twitter などの企業で数万件のメッセージを処理するために実稼働で使用されています。運用コストが低く、柔軟性が高いという特徴があります。 Pulsar は、Kafka の課題のほとんどを解決し、スケーリングを容易にすることを目的としています。

Pulsar は非常に柔軟性が高く、Kafka のような分散ログアプリケーションシナリオだけでなく、RabbitMQ のような純粋なメッセージングシステムシナリオにも適用できます。

複数のタイプのサブスクリプション、複数の配信保証、保持ポリシー、スキーマの進化を処理する方法など、さまざまな機能をサポートしています。

パルサーアーキテクチャ図[4]

Pulsar の特徴は次のとおりです。

マルチテナント機能が組み込まれているため、異なるチームが同じクラスターを使用し、それを他のクラスターから分離できるため、多くの管理上の課題を解決できます。分離、認証、承認、および割り当てをサポートします。
多層アーキテクチャ: Pulsar は、すべてのトピックデータを Apache BookKeeper によってサポートされる特殊なデータ層に保存します。

ストレージとメッセージングを分離することで、クラスターのスケーリング、再バランス調整、保守に関する多くの問題が解決されます。また、信頼性も向上し、データ損失が事実上不可能になります。

さらに、リアルタイムの取り込みに影響を与えることなく、データを読み取るときに BookKeeper に直接接続できます。たとえば、Presto を使用すると、KSQL と同様にトピックに対して SQL クエリを実行できますが、リアルタイムのデータ処理には影響しません。
仮想トピック: n 層アーキテクチャにより、トピックの数に制限はなく、トピックとそのストレージは分離されています。ユーザーは非永続的なトピックを作成することもできます。
N 層ストレージ: Kafka の問題の 1 つは、ストレージが高価になる可能性があることです。したがって、「コールド」データを保存するために使用されることはほとんどなく、メッセージは削除されることが多いです。 Apache Pulsar は、階層型ストレージを利用して古いデータを Amazon S3 やその他のデータストレージシステムに自動的にオフロードし、クライアントに対して透過的なビューを提供できます。 Pulsar クライアントは、すべてのメッセージがログに存在するかのように、タイムノードの先頭から読み取ることができます。
Pulsar 関数: 簡単にデプロイでき、計算プロセスが軽量で、開発者に優しい API を備えており、独自のストリーム処理エンジン (Kafka など) を実行する必要はありません。
セキュリティ: 組み込みプロキシ、マルチテナントセキュリティ、プラグ可能な認証などの機能があります。
高速な再バランス: パーティションは、再バランスが容易なシャードに分割されます。
サーバー側の重複排除と無効なフィールド: クライアント側で行う代わりに、圧縮中に重複データを削除することもできます。
組み込みのスキーマレジストリ: 複数の戦略をサポートし、操作が簡単です。
地理的レプリケーションと組み込み検出: クラスターを複数のリージョンに簡単に複製できます。
統合されたロードバランサーと Prometheus メトリック。
複数の統合: Kafka、RabbitMQ など。
GoLang、Java、Scala、Node、Python など、複数のプログラミング言語をサポートします。
シャーディングとデータのパーティション分割はサーバー側で透過的に実行されるため、クライアントはデータのシャーディングとパーティション分割を理解する必要はありません。

パルサー機能一覧[5]

Pulsarを使い始める

Pulsar を使い始めるのはとても簡単です。使用の前提条件は、JDK をインストールすることです。

① Pulsar をダウンロードして解凍します (注: Apache Pulsar の最新バージョンは 2.7.0 です)。

 $ wget https://archive.apache.org/dist/pulsar/pulsar-2.6.1/apache-pulsar-2.6.1-bin.tar.gz

② コネクタをダウンロードする（オプション）:

 $ wget https://archive.apache.org/dist/pulsar/pulsar-2.6.1/connectors/{connector}-2.6.1.nar

③narファイルをダウンロードしたら、Pulsarディレクトリ内のConnectorsディレクトリにファイルをコピーします。

④パルサーを起動！

 $ bin/pulsar スタンドアロン

Pulsar は、クラスターと対話するために使用できる Pulsar-Client と呼ばれる CLI ツールを提供します。

制作ニュース:

 $ bin/pulsar-client は my-topic --messages "hello-pulsar" を生成します

消費メッセージ:

 $ bin/pulsar-client は my-topic -s "first-subscription"を消費します

Akka ストリームの例

クライアントの例として、Akka 上で Pulsar4s を使用します。

まず、データストリームを消費するためのソースを作成する必要があります。必要なのは、要求に応じてコンシューマーを作成し、メッセージ ID を検索する関数だけです。

 val topic = Topic( "persistent://standalone/mytopic" )
 val consumerFn = () => client.consumer(ConsumerConfig(トピック、サブスクリプション))

次に、ConsumerFn 関数を渡してソースを作成します。

 com.sksamuel.pulsar4s.akka.streams._ をインポートします
val pulsarSource = source(consumerFn、 Some (MessageId.earliest))

Akka ソースの具体化された値は Control のインスタンスであり、メッセージの消費を停止するために使用できる "close" メソッドを提供します。これで、通常どおり Akka Streams を使用してデータを処理できます。

受信機を作成するには:

 val topic = Topic( "persistent://standalone/mytopic" )
 val producerFn = () => client.producer(ProducerConfig(トピック))
 com.sksamuel.pulsar4s.akka.streams._ をインポートします
val pulsarSink = sink(producerFn)

完全な例はPulsar4s[6]から引用されています。

オブジェクトの例 {
  com.sksamuel.pulsar4s.{ConsumerConfig、MessageId、ProducerConfig、PulsarClient、Subscription、Topic} をインポートします。
  org.apache.pulsar.client.api をインポートします。スキーマ 
  暗黙の val システム: ActorSystem = ActorSystem()
  暗黙の val マテリアライザー: ActorMaterializer = ActorMaterializer()
  暗黙的な値スキーマ:スキーマ[Array[Byte]] =スキーマ.BYTES
  val クライアント = PulsarClient( "pulsar://localhost:6650" )
  val intotopic = Topic( "persistent://sample/standalone/ns1/in" )
  val outtopic = Topic( "persistent://sample/standalone/ns1/out" )
  val consumerFn = () => client.consumer(ConsumerConfig(topics = Seq(intopic), subscriptionName = Subscription( "mysub" )))
  val producerFn = () => client.producer(ProducerConfig(outtopic))
  val control = source(consumerFn、 Some (MessageId.earliest))
    .map { コンシューマーメッセージ => プロデューサーメッセージ(コンシューマーメッセージ.data) }
    。 ( sink(producerFn)).run() へ
  スレッド.スリープ(10000)
  制御停止()
 }

パルサー関数の例

Pulsar 関数は、1 つ以上のトピックからのメッセージを処理し、変換して、結果を別のトピックに出力します。

パルサー機能[7]

関数を記述するための 2 つのインターフェースから選択できます。

言語ネイティブインターフェース: Pulsar 固有のライブラリや特別な依存関係は必要ありません。コンテキストにアクセスできず、Java と Python のみをサポートします。
Pulsar Function SDK: Java/Python/Go で利用可能で、コンテキストオブジェクトへのアクセスなど、より多くの機能を提供します。

言語ネイティブインターフェイスを使用してメッセージを変換する簡単な関数を記述するだけです。

 defプロセス(入力):
戻る  「{}!」 .format(入力)

Python で記述されたこの単純な関数は、渡された文字列に感嘆符を追加し、結果の文字列をトピックに公開するだけです。

SDK を使用するには依存関係をインポートする必要があります。たとえば、Go では次のように記述できます。

パッケージメイン
輸入 （
 "コンテクスト"  
 「fmt」  
 「github.com/apache/pulsar/pulsar-function-go/pf」  
 ）
 function HandleRequest(ctx context.Context, in []byte) エラー {
 fmt.Println(文字列( in ) + "!" )
ゼロを返す
}
関数main() {
 pf.Start(リクエストの処理)
 }

サーバーレス関数を公開してクラスターにデプロイしたい場合は、Pulsar-Admin CL を使用できます。 Python を使用する場合は、次のように記述できます。

 $ bin/pulsar-admin 関数を作成します\
 --py ~/router.py \  
 --classname router.RoutingFunction \  
 --テナントパブリック\  
 --namespace デフォルト \  
 --name ルート-フルーツ-野菜 \  
 --inputs persistent://public/default/basket-items  
 Pulsar Functionの重要な機能は、ユーザーが関数を公開するときに配信保証を設定できることです。
 $ bin/pulsar-admin 関数を作成します\
 --name 実際に 1 回だけ実行する関数 \  
 --処理保証は EFFECTIVELY_ONCE

オプションは次のとおりです:

パルサーの利点

Kafka と比較した Pulsar の主な利点を見てみましょう。

その他の機能: Pulsar Function、マルチテナント、スキーマレジストリ、n 層ストレージ、複数の消費モードと永続モードなど。
柔軟性の向上: 3 つのサブスクリプションタイプ (排他、共有、フェイルオーバー) により、ユーザーは 1 つのサブスクリプションで複数のトピックを管理できます。
永続オプション: 非永続 (高速)、永続、圧縮 (メッセージごとに最後のキーのみ)、ユーザーは配信保証を選択できます。 Pulsar は、サーバー側の重複排除、複数の保持ポリシー、無効な単語の TTL を備えています。
拡張要件を事前に定義する必要はありません。
Pulsar はキューとストリームの両方のメッセージ消費モデルをサポートしているため、RabbitMQ と Kafka の両方を置き換えることができます。
ストレージはブローカーから分離されているため、スケーラビリティが向上し、再バランスがより高速かつ信頼性が高くなります。
操作と保守が簡単: アーキテクチャ分離と n 層ストレージ。
Presto との SQL 統合により、ブローカーに影響を与えることなくストアを直接クエリできます。
n 層自動ストレージオプションを使用すると、より低コストでストレージを実現できます。
より高速：ベンチマーク[8]では、さまざまな状況でより優れたパフォーマンスが示されています。 Pulsar はレイテンシが低く、スケーリング機能が優れています。
Pulsar Function はサーバーレスコンピューティングをサポートし、デプロイメント管理は必要ありません。
スキーマレジストリと統合します。
統合されたロードバランサーと Prometheus メトリック。
ジオレプリケーションはより適切に機能し、セットアップも簡単になります。 Pulsar には検出機能が組み込まれています。
作成できるトピックの数に制限はありません。
Kafka と互換性があり、簡単に統合できます。

パルサーの欠点

パルサーは完璧ではありません。 Pulsar にもいくつか問題があります:

サポート、ドキュメント、例が比較的不足しています。
n 層アーキテクチャでは、もう 1 つのコンポーネントである BookKeeper が必要になります。
Kafka よりもプラグインとクライアントの数が少ないです。
クラウドではサポートが少ないため、Confluent ではマネージドクラウドを提供しています。

しかし、上記の状況は急速に改善しており、Pulsar は徐々に多くの企業や組織で利用されるようになっています。

Apache Pulsar の商用サポート会社 StreamNative も StreamNative Cloud を立ち上げました。 Apache Pulsar は急速に成長しており、私たちは皆、刺激的な変化を目にすることができます。

Confluent は Pulsar と Kafka を比較するブログを公開していますが、これらの質問には偏りがある可能性があることに注意してください。

パルサーの使用シナリオ

Pulsar は幅広いシナリオで使用できます。

パブリッシュ/サブスクライブキューメッセージング。
分散ログ記録。
永続的なイベントストレージのためのイベントソーシング。
マイクロサービス。
SQL 分析。
サーバーレス関数。

Pulsar を検討すべきなのはいつですか?

RabbitMQ のようなキューと Kafka のようなストリームハンドラーの両方が必要です。
使いやすい地理的レプリケーションが必要です。
マルチテナントを実装し、各チームのアクセス権限を確保します。
メッセージを長期間保持する必要があり、別のストレージにオフロードしたくない。
高いパフォーマンスが求められており、ベンチマークでは Pulsar がより低いレイテンシと高いスループットを提供することが示されています。

クラウドの場合は、必ずクラウドベースのソリューションを検討してください。クラウドプロバイダーには、特定のシナリオをカバーするさまざまなサービスがあります。

たとえば、キューメッセージングの場合、クラウドプロバイダーは Google pub/sub などの多くのサービスを提供しています。分散ログには、Confluent Cloud または AWS Kinesis があります。 StreamNative は Pulsar をベースにしたクラウドサービスも提供しています。

クラウドプロバイダーは非常に優れたセキュリティも提供します。 Pulsar の利点は、1 つのプラットフォームで多くの機能を提供していることです。

チームによっては、これをマイクロサービスのメッセージングシステムとして使用する場合もあれば、データ処理用の分散ログとして使用する場合もあります。

結論は

私は Kafka の大ファンですが、Pulsar にとても興味を持っているのには理由があります。それは、競争がイノベーションを推進するからです。

Kafka は成熟した、回復力のある、実戦でテストされた製品であり、世界中で大成功を収めており、ほとんどの企業にとって Kafka なしでの事業は考えられないほどです。

しかし、私は Kafka が自身の成功の犠牲者になるだろうと考えています。Kafka の急成長により、多くの大企業をサポートする必要性から機能開発が遅くなり、ZooKeeper への依存の排除などの重要な機能に時間がかかりすぎて、Pulsar のようなツールが繁栄する余地が生まれています。

パルサーはまだ若いですが、勢いは抜群です。 Pulsar を組織に組み込む前に、分析、ベンチマーク、調査、POC を実施する必要があります。

まずは小規模に開始し、Kafka を Pulsar に移行する前に概念実証を実施し、完全な移行を決定する前に影響を評価します。

参考リンク: