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

初版: 2018/9/28
著者: 伊藤 雅博, 株式会社日立製作所

# はじめに

この投稿では[オープンソースカンファレンス2017.Enterprise](https://www.ospn.jp/osc2017.enterprise/)で発表した「[めざせ！Kafkaマスター　～Apache Kafkaで最高の性能を出すには～](https://www.ospn.jp/osc2017.enterprise/pdf/OSC2017.enterprise_Hitachi_Kafka.pdf)」の検証時に調査した内容を紹介します（全8回の予定）。本投稿の内容は2017年6月にリリースされたKafka 0.11.0 時点のものです。

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

**投稿一覧：**
1. Apache Kafkaの概要とアーキテクチャ (本投稿)
2. [Apache KafkaのProducer/Broker/Consumerのしくみと設定一覧](https://qiita.com/sigmalist/items/3b512e2ab49b07271665)
3. [Apache Kafkaの推奨構成と性能の見積もり方法](https://qiita.com/sigmalist/items/b3d25e914dc30539ece4)
4. [Apache Kafkaの性能検証(1): 検証環境とパラメータチューニングの内容](https://qiita.com/sigmalist/items/730c7c82e02e5a837b9a)
5. [Apache Kafkaの性能検証(2): Producerのチューニング結果](https://qiita.com/sigmalist/items/0cd76d7edb055e89d1e6)
6. [Apache Kafkaの性能検証(3): Brokerのチューニング結果](https://qiita.com/sigmalist/items/0795bde01bcbd78b1c47)
7. [Apache Kafkaの性能検証(4): Producerの再チューニングおよびConsumerのチューニング結果](https://qiita.com/sigmalist/items/b0bc222f449a9f3cc273)
8. [Apache Kafkaの性能検証(5): システム全体のレイテンシについて](https://qiita.com/sigmalist/items/111e96cdae69bb2d89e6)

# Apache Kafkaとは

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

* メッセージキューを介してシステム間の通信を行うことで、システム間の接続経路を簡略化し、システムを疎結合に保つ
* システム間の通信を非同期化することで、データ流量の急激な増加によるシステムの負荷上昇を抑制する

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

# Kafkaのユースケース

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

## データハブの構築

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

![kafka01_01.png](https://qiita-image-store.s3.amazonaws.com/0/285851/36d3cad2-6d57-c982-a250-ae7d08475e66.png)

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

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

![kafka01_02.png](https://qiita-image-store.s3.amazonaws.com/0/285851/215be971-4b22-7eb4-c1ce-5cac888c21a2.png)

# システム構成

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

![kafka01_03.png](https://qiita-image-store.s3.amazonaws.com/0/285851/35ab4fa5-511f-c2a8-59e3-09c88ad7bfdd.png)

Kafkaは複数のBrokerでクラスタを構成し、このクラスタ上にTopicと呼ばれる分散キューを構成します。Kafkaのメッセージはキーバリュー形式であり、Recordと呼びます。1つのTopicは、複数のBrokerに分散配置されたPartitionで構成されており、このPartition単位でRecordの書き込み/読み込みを行うことで、1つのTopicに対する並列書き込み/読み出しを実現します。Broker同士はクラスタコーディネータであるZooKeeperを使用して連携し、Brokerの１つがリーダーとなり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](https://qiita-image-store.s3.amazonaws.com/0/285851/39ecbe0a-78e6-9f6c-ab01-fc4b61f81565.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](https://qiita-image-store.s3.amazonaws.com/0/285851/c46cbadf-94d5-88f2-3d31-1f56a8acc389.png)

# Kafkaのツール

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

**[Kafka Connect](https://kafka.apache.org/0110/documentation/#connect)**

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

**[Kafka Streams](https://kafka.apache.org/0110/documentation/streams/)**

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

**[Miller Maker](https://kafka.apache.org/0110/documentation.html#basic_ops_mirror_maker)**

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

# Kafkaの機能

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

**[セキュリティ](https://kafka.apache.org/0110/documentation/#security)**

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

**[リソース割り当て制限（Quotas）](https://kafka.apache.org/0110/documentation/#design_quotas)**

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

**[Logコンパクション](https://kafka.apache.org/0110/documentation/#compaction)**

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

**[モニタリング](https://kafka.apache.org/0110/documentation/#monitoring)**

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

**[Recordの重複排除、トランザクション、エンドツーエンドのExactly-once保障](https://kafka.apache.org/0110/documentation/#semantics)**

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のしくみと設定一覧](https://qiita.com/sigmalist/items/3b512e2ab49b07271665)