Spark StreamingとKafkaの統合を分析する2つの方法

Spark Streaming は、マイクロバッチ処理に基づくストリーミング コンピューティング エンジンであり、通常は Spark Core または Spark Core と Spark Sql を使用してデータを処理します。エンタープライズ リアルタイム処理アーキテクチャでは、通常、Spark Streaming と Kafka は、ビッグ データ処理アーキテクチャ全体のコア リンクの 1 つとして統合されます。

異なる Spark および Kafka バージョンでは、データを統合して処理する方法として、レシーバー ベースのアプローチと直接アプローチの 2 つの方法があります。さまざまな統合バージョンのサポートについては、次の図を参照してください。

レシーバーベースのアプローチ

レシーバーベースのアプローチは、Kafka コンシューマー高レベル API を使用して実装されます。

すべての受信者にとって、Kafka を介して受信されたデータは Spark エグゼキューターに保存されます。基礎となるレイヤーは BlockManager に書き込まれ、デフォルトでは 200 ミリ秒ごとにブロックが生成されます (構成パラメータ spark.streaming.blockInterval によって決定されます)。その後、BlockRdd は Spark ストリーミングによって送信されたジョブによって構築され、最終的に Spark コア タスクとして実行されます。

受信方式に関しては、いくつか注意すべき点があります。

• レシーバーは、エグゼキューター上で常駐スレッドとして実行するようにスケジュールされ、1つのCPUを占有します。
• 受信者の数は、KafkaUtils.createStream の呼び出し回数によって決まり、一度に 1 つの受信者が呼び出されます。
• Kafka のトピック パーティションは、Spark Streaming で生成された RDD パーティションに関連付けることはできません。
• KafkaUtils.createStream() で指定されたトピック パーティションの数を増やすと、単一のレシーバーによって消費されるトピックあたりのスレッド数のみが増加します。データを処理する並列スパークの数は増加しません。 [topicMap[topic,num_threads]マップの値は、トピックあたりのコンシューマースレッドの数に対応します。]
• デフォルトでは、受信機は 200 ミリ秒ごとにブロックを生成します。データ量に応じてブロック生成周期を調整することをお勧めします。
• 受信側が受信したデータは BlockManager に格納されます。各エグゼキュータには BlockManager インスタンスがあります。データの局所性により、レシーバーを持つエグゼキューターはより多くのタスクを実行するようにスケジュールされ、一部のエグゼキューターがアイドル状態になります。

パラメータ spark.locality.wait を使用してデータの局所性を調整することをお勧めします。このパラメータが適切に設定されていない場合、たとえば、10 に設定されていてタスクが 2 秒で処理される場合、データが存在するエグゼキュータにスケジュールされるタスクがますます多くなり、タスクの実行が遅くなったり、失敗したりします (これはデータ スキューとは区別する必要があります)。

複数の Kafka 入力を持つ DStreams は、異なるグループとトピックを使用して作成でき、複数のレシーバーを使用してデータを受信および処理できます。

2 種類の受信機: 信頼できる受信機:

• 信頼できる受信者は、データを受信し、レプリケーション メカニズムを通じて Spark に保存するときに、信頼できるデータ ソースに ACK 確認を正確に送信します。信頼できない受信機:
• 信頼できない受信者は、データが受信されたことの確認をデータ ソースに送信しません。これは、ack をサポートしていないデータ ソースに適しています。もちろん受信機のカスタマイズも可能です。
• 受信機はデータの信頼性を処理します。デフォルトでは、受信者はデータを失う可能性があります。
• spark.streaming.receiver.writeAheadLog.enable を true に設定して先行書き込みログ メカニズムを有効にし、HDFS などの信頼性の高い分散ファイル システムにデータを書き込むことで、データが失われないようにすることができますが、パフォーマンスは多少低下します。

消費者が消費する最大速度を制限するには、次の 3 つのパラメータが関係します。

• spark.streaming.backpressure.enabled: デフォルト値は false です。 true に設定すると、バックプレッシャー メカニズムが有効になります。
• spark.streaming.backpressure.initialRate: デフォルトでは、初期消費率は設定されていません。各受信機が最初に起動したときに受信したデータの最大値。
• spark.streaming.receiver.maxRate: デフォルト値は設定されていません。各受信者がデータを受信する最大速度 (1 秒あたりのレコード数)。各ストリームは、1 秒あたり最大でこの数のレコードを消費します。この設定を 0 または負の数に設定すると、最大レートは制限されません。

ジョブが生成されると、現在のジョブの有効範囲内のすべてのブロックが BlockRDD に結合され、1 つのブロックが 1 つのパーティションに対応します。

Kafka082 バージョンのコンシューマー高レベル API には、グループ化の概念があります。コンシューマー グループ内のスレッド数 (コンシューマー数) を Kafka パーティションの数と一致させることをお勧めします。パーティションの数を超えると、一部のコンシューマがアイドル状態になります。

直接アプローチ

直接的なアプローチは、Spark Streaming がレシーバーを使用せずに Kafka と統合する方法です。一般的には、エンタープライズの生産環境でより多く使用されます。受信機と比較すると、次の特徴があります。

1. 受信機を使用しない

複数のKafkaストリームを作成して集約する必要がない

不要なCPU使用率を削減

データを受信して​​ BlockManager に書き込むプロセス全体が削減され、実行時に blockId、ネットワーク転送、ディスク読み取りなどを通じてデータを取得するため、効率が向上します。

WALが不要になり、ディスクIO操作がさらに削減されます。

2. 直接方式で生成された RDD は KafkaRDD であり、そのパーティション数は Kafka パーティション数と一致します。消費には同じ数の RDD パーティションが使用されるため、並列度を制御しやすくなります。

注意: この対応は、シャッフルまたは再パーティション操作後に生成された RDD では無効になります。

3. オフセットを手動で管理して、正確に1回のセマンティクスを実装できます。

4. データの局所性の問題。 KafkaRDD のコンピューティング関数では、SimpleConsumer を使用して、指定されたトピック、パーティション、オフセットに従って Kafka データを読み取ります。

ただし、バージョン 010 以降では、Kafka と Spark が同じクラスター内にある場合、データの局所性に問題が発生します。

5. 消費者の消費の最大速度を制限する

spark.streaming.kafka.maxRatePerPartition: 各 kafka パーティションからデータを読み取ることができる最大速度 (1 秒あたりのレコード数)。これは各パーティションのレート制限です。システム スループットを評価するには、Kafka パーティションの数を事前に知っておく必要があります。