Kafka

Apache Kafkaの概要とアーキテクチャ

初版: 2018/9/28

著者: 伊藤 雅博, 株式会社日立製作所


はじめに

この投稿ではオープンソースカンファレンス2017.Enterpriseで発表した「めざせ!Kafkaマスター ~Apache Kafkaで最高の性能を出すには~」の検証時に調査した内容を紹介します(全8回の予定)。本投稿の内容は2017年6月にリリースされたKafka 0.11.0 時点のものです。

第1回目となる今回は、Apache Kafkaの概要とアーキテクチャについて紹介します。

投稿一覧:

1. Apache Kafkaの概要とアーキテクチャ (本投稿)

2. Apache KafkaのProducer/Broker/Consumerのしくみと設定一覧

3. Apache Kafkaの推奨構成と性能の見積もり方法

4. Apache Kafkaの性能検証(1): 検証環境とパラメータチューニングの内容

5. Apache Kafkaの性能検証(2): Producerのチューニング結果

6. Apache Kafkaの性能検証(3): Brokerのチューニング結果

7. Apache Kafkaの性能検証(4): Producerの再チューニングおよびConsumerのチューニング結果

8. Apache Kafkaの性能検証(5): システム全体のレイテンシについて


Apache Kafkaとは

Apache Kafka(以降、Kafka)はスケーラビリティに優れた分散メッセージキューです。メッセージキューとは、システム間のデータの受け渡しを仲介し、データを一時的に保持(キューイング)するミドルウェアです。メッセージキューを導入することには、以下のような利点があります。


  • メッセージキューを介してシステム間の通信を行うことで、システム間の接続経路を簡略化し、システムを疎結合に保つ

  • システム間の通信を非同期化することで、データ流量の急激な増加によるシステムの負荷上昇を抑制する

Kafkaは処理性能を重視したメッセージキューであり、複数台のマシンでクラスタを構成して分散処理を行うことで、高いスループットを発揮します。後からクラスタにマシンを追加することで、処理性能とデータ保持容量をスケールアウトすることもできます。また、Kafkaはクラスタ内でデータを複製するため、一部のマシンに障害が発生してもデータを失うことなく処理を継続できます。


Kafkaのユースケース

Kafkaは主に以下の2種類のユースケースで使用されています。


データハブの構築

複雑なシステムをサービス単位で分割して疎結合に保つことで、迅速な機能追加や不良修正が可能なマイクロサービスが多くの企業で導入されています。しかし、多数のサービス間で直接データを受け渡そうとすると接続経路が複雑になります。またデータ流量の急激な増加により、サービスの負荷が急増することもあります。Kafkaはこのようなサービス間で、大量のデータを受け渡すためのハブとして活用されています。

kafka01_01.png


ストリーミングアプリケーションの構築

Webサービスやネットワークに接続されたデバイスは、大量のデータをリアルタイムに生成します。このようなデータをリアルタイムに処理するストリーミングアプリケーションにおいて、データ量の急激な増加に対応するためのキューとしてKafkaが活用されています。

kafka01_02.png


システム構成

Kafkaの全体アーキテクチャを以下に示します。

kafka01_03.png

Kafkaは複数のBrokerでクラスタを構成し、このクラスタ上にTopicと呼ばれる分散キューを構成します。Kafkaのメッセージはキーバリュー形式であり、Recordと呼びます。1つのTopicは、複数のBrokerに分散配置されたPartitionで構成されており、このPartition単位でRecordの書き込み/読み込みを行うことで、1つのTopicに対する並列書き込み/読み出しを実現します。Broker同士はクラスタコーディネータであるZooKeeperを使用して連携し、Brokerの1つがリーダーとなりBrokerクラスタを管理します。

KafkaはPub/Subメッセージングモデルを採用しています。また、Kafkaは書き込み/読み出し性能を重視しており、Topicに対する書き込み/読み出しには、MQTTなどの標準プロトコルではなく、独自プロトコルを使用します。書き込み側のアプリケーション(Publisher)はProducerと呼ばれる書き込み用ライブラリを通じて、BrokerのTopicにRecordを書き込みます。読み出し側のアプリケーション(Subscriber)はConsumerと呼ばれる読み出し用ライブラリを通じて、Recordを取り出します。Kafka本体に含まれているProducer/ConsumerライブラリはJava製ですが、他にもサードパーティからPythonやC++など様々な言語製のライブラリが提供されており、様々なアプリケーションや機器に組み込むことができます。

Topicに書き込んだデータはBrokerのディスクに保存され、読み出されてもすぐには削除されません。そのため、同じTopicのデータを複数のアプリケーションから読み出すことができます。一定期間が経過したRecordや、Partitionの最大容量を超えた分のRecordは自動的に削除されます。また、Partition 内のRecordはBroker間で複製されるため、一部のBrokerノードに障害が発生してもRecordが失われにくい仕組みになっています。


データ管理

Kafkaの論理/物理データ構造とその管理方法について説明します。


論理データ構造とデータ複製


Topic、Partition、Replica

1つのTopicは複数のPartitionで構成されており、各PartitionはBroker間で複製された複数のReplicaで構成されます。各Partitionには1個のLeader Replicaと0個以上のFollower Replicaが存在し、Leader Replicaに書き込まれたRecordはFollower Replicaに複製されます。

この複製方法はLeader/Follower型と呼ばれ、Replicaの複製数(1個のLeaderと0個以上のFollowerの合計数)をReplication Factorと呼びます。各Leader ReplicaはBrokerに均等に割り当てられ、Partitionへの書き込み/読み出しはLeader Replicaにのみ行います。Brokerノードの障害などでLeader Replicaが使用できなくなった際は、いずれかのFollower ReplicaがLeader Replicaに昇格します。

TopicのPartition数とReplication FactorはTopicの作成時に指定します。Brokerノード3台で1 Topic、3 Partition、Replication Factor = 3の場合の、データ構造の例を以下に示します。

kafka01_04.png


In Sync Replica(ISR)

Brokerやネットワークの問題により、LeaderからFollowerへの複製が遅延または停止して、LeaderとFollowerの内容が同期しなくなることがあります。Kafkaのデフォルトの設定では、複製が追い付いていない状態が10秒間続いたFollowerは同期していないとみなされます。Brokerは常に「同期しているFollowerのリスト」を追跡しており、同期しているReplicaをIn Sync Replica(ISR)と呼びます。

ISRの最小数はTopicごとに設定することが可能で、現在のISR 数はPartitionごとに追跡されます。あるPartitionのISR数が最小数を下回ると、そのPartition(のLeader Replica)はProducerから書き込み出来なくなります。ISR数が最小数まで回復すると、そのPartitionは再びProducerから書き込み出来るようになります。ISRの最小数まで書き込まれたRecordはコミットされたものとみなされ、ConsumerはコミットされたRecordのみを取得できます。


物理データ構造とデータ保存方法


Log、Segment

BrokerはTopicに書き込まれたRecordをファイルに保存することで永続化します。BrokerはReplica ごとにデータディレクトリを作成して、Recordは各ディレクトリ内のSegmentというファイルに書き込みます。このSegmentファイルの集合を、KafkaではLogと呼んでいます。

通常、Brokerノードは複数台のディスクを搭載しており、各ディスクに対してラウンドロビン方式でReplicaのデータディレクトリを割り当てます。そのため、最低でもディスク台数分のReplicaがないとディスクを使い切れません(Replica数 = 全TopicのPartition数 * Replication Factor)。また、Partition間のデータのバランスが悪いと、ディスク間に負荷の不均衡が発生してしまいます。


Segmentファイルのフラッシュ

Segmentファイルはファイルシステムが管理しています。Linuxでは、ファイルシステムに書き込まれたデータは、ディスクに書き出されるまでメモリ上のページキャッシュに保持されます。デフォルト設定では、KafkaはOSに対してディスクへの書き込み指示は行わず、OSがバックグラウンドでページキャッシュ上のデータを定期的にディスクへフラッシュするのに任せます。なお、フラッシュ後もメモリに余裕があれば、データはページキャッシュ上に残り続けるため、高速に読み出すことができます。

Brokerの設定により、一定期間または一定量のデータが書き込まれた際に、データを強制的にフラッシュすることも可能です。ただし、KafkaはBrokerノードが故障した際は別のBrokerノードのReplicaからデータを回復するため、ディスクへのデータ同期を必要とはしていません(他のノードのページキャッシュ上に複製されていればよいという考え方)。そのため、デフォルトのフラッシュ設定を使用することを推奨します。

kafka01_05.png


Kafkaのツール

Kafkaに付属する主なツールについて、その概要を簡単に説明します。詳細についてはリンク先の公式ドキュメントを参照してください。

Kafka Connect

Kafka Connectを使用することで、DBなどの外部システムとKafka間でデータを書き込み/読み出しするコネクタを定義できます。Offsetの永続化など、Kafkaと外部システムを接続する際に必要な処理を簡素化してくれます。

Kafka Streams

Kafka Streamsは、Kafkaに格納されたデータを処理および分析するためのクライアントライブラリです。ウィンドウ処理などが可能であり、Spark StreamingやFlinkなどの並列分散処理フレームワークに相当します。

Miller Maker

Miller MakerはKafkaクラスタ間のミラーリングを行います。複数のデータセンタにクラスタを用意することで、データセンタ間でのバックアップが可能となります。Miller Makerの実体は一方のクラスタからConsumeしたRecordを他方のクラスタにProduceするクライアントアプリケーションであり、ミラーリングは非同期に行われます。そのため、フォールトトレランス機能としての使用は意図されていません。


Kafkaの機能

Kafkaの主な機能について、その概要を簡単に説明します。詳細についてはリンク先の公式ドキュメントを参照してください。

セキュリティ

KafkaはSSL/SASLを用いた接続認証、通信データの暗号化、クライアントによる読み取り/書き込み操作の認証などを提供しています 。

リソース割り当て制限(Quotas)

Quotasによりクライアントが使用するBrokerリソース(ネットワーク帯域幅およびネットワーク・ディスクI/OスレッドのCPU要求レート)を制御できます 。

Logコンパクション

Logコンパクションを使用することで、各キーの最新Recordのみを保持できます(一定期間が過ぎたRecordを削除するのではなく、同じキーの最新Record以外を削除する)。この保存ポリシーはTopicごとに設定可能です。

モニタリング

KafkaはJMX (Java Management Extensions) により性能情報などのメトリックを公開しています。JDK付属のjconsoleや、JMXに対応した監視ソフトウェアを使用することで、これらのメトリックを収集できます。

Recordの重複排除、トランザクション、エンドツーエンドのExactly-once保障

Kafka 0.11では以下の機能が追加されました。


  • Producerは送信したRecordがコミットされたことを示す応答を、Brokerから受信できなかった場合にRecordを再送するため、Logに同じRecordが書き込まれる可能性があります。Kafka 0.11では、Producerによって送信されるシーケンス番号を使用することで、この重複を排除する機能が追加されました。

  • Kafka 0.11では、Producer/Consumerは複数のTopicにまたがるトランザクション機能が追加されました。トランザクションを使用することで、あるTopicから読み出したRecordを変換処理して別のTopicに書き込む、といった一連の処理をアトミックに実行できます。

  • Kafkaはデフォルト設定において、At least once(データが重複する可能性があるが、欠損はしない)のRecord配信を保障しています。Kafka 0.11では、Recordの重複排除やトランザクションなどの設定を行い、Kafka StreamsとKafka Connectを組み合わせることで、エンドツーエンドのExactly-onceのRecord配信(ProducerのRecord送信から、変換処理、外部DBへの出力までをRecordの欠損や重複なしに実行)が可能になりました。


おわりに

本投稿ではApache Kafkaの概要とアーキテクチャについて紹介しました。次回の投稿ではKafkaのコンポーネントであるBroker、Producer、Consumerについて、詳細な処理の流れとパラメータ設定を紹介します。

第2回:Apache KafkaのProducer/Broker/Consumerのしくみと設定一覧