Apache Spark アプリケーションを Azure Event Hubs と接続する

[アーティクル]
03/14/2023

このチュートリアルでは、リアルタイムストリーミングのために Spark アプリケーションを Event Hubs に接続する方法について説明します。この統合により、お使いのプロトコルクライアントを変更せずにストリーミングを行ったり、独自の Kafka または Zookeeper クラスターを実行したりすることができます。このチュートリアルには、Apache Spark v2.4 以上および Apache Kafka v2.0 以上が必要となります。

Note

このサンプルは GitHub で入手できます。

このチュートリアルでは、以下の内容を学習します。

Event Hubs 名前空間を作成します
サンプルプロジェクトを複製する
Spark を実行する
Kafka 用 Event Hubs から読み取る
Kafka 用 Event Hubs に書き込む

前提条件

このチュートリアルを開始する前に、以下のものを用意してください。

Azure のサブスクリプション。アカウントがない場合は、無料アカウントを作成してください。
Apache Spark v2.4
Apache Kafka v2.0
Git

注意

Spark v2.4 以降、Spark-Kafka アダプターは、Kafka v2.0 をサポートするように更新されています。それより前にリリースされた Spark のアダプターは、Kafka v0.10 以降をサポートしますが、厳密には Kafka v0.10 の API に依存します。 Kafka 用 Event Hubs は Kafka v0.10 をサポートしていないため、Kafka エコシステム用 Event Hubs では、バージョン v2.4 未満の Spark の Spark-Kafka アダプターがサポートされません。

Event Hubs 名前空間を作成します

Event Hubs サービスとの間で送受信を行うには、イベントハブの名前空間が必要です。名前空間とイベントハブを作成する手順については、イベントハブの作成に関するページを参照してください。 Event Hubs の接続文字列と完全修飾ドメイン名 (FQDN) を、後で使用するために取得します。手順については、「Get an Event Hubs connection string (Event Hubs の接続文字列を取得する)」を参照してください。

サンプルプロジェクトを複製する

Azure Event Hubs リポジトリを複製し、tutorials/spark サブフォルダーに移動します。

git clone https://github.com/Azure/azure-event-hubs-for-kafka.git
cd azure-event-hubs-for-kafka/tutorials/spark

Kafka 用 Event Hubs から読み取る

構成にわずかな変更を加えれば、Kafka 用 Event Hubs からの読み取りを開始することができます。 BOOTSTRAP_SERVERS と EH_SASL を実際の名前空間の情報で更新すれば、Kafka を使った場合と同じように、Event Hubs を使ったストリーミングを開始できます。完全なサンプルコードについては、GitHub にある sparkConsumer.scala ファイルを参照してください。

//Read from your Event Hub!
val df = spark.readStream
    .format("kafka")
    .option("subscribe", TOPIC)
    .option("kafka.bootstrap.servers", BOOTSTRAP_SERVERS)
    .option("kafka.sasl.mechanism", "PLAIN")
    .option("kafka.security.protocol", "SASL_SSL")
    .option("kafka.sasl.jaas.config", EH_SASL)
    .option("kafka.request.timeout.ms", "60000")
    .option("kafka.session.timeout.ms", "30000")
    .option("kafka.group.id", GROUP_ID)
    .option("failOnDataLoss", "true")
    .load()

//Use dataframe like normal (in this example, write to console)
val df_write = df.writeStream
    .outputMode("append")
    .format("console")
    .start()

次のようなエラーが発生した場合は、spark.readStream 呼び出しに .option("spark.streaming.kafka.allowNonConsecutiveOffsets", "true") を追加して、もう一度お試しください。

IllegalArgumentException: requirement failed: Got wrong record for <spark job name> even after seeking to offset 4216 got offset 4217 instead. If this is a compacted topic, consider enabling spark.streaming.kafka.allowNonConsecutiveOffsets

Kafka 用 Event Hubs に書き込む

Event Hubs に対する書き込みも、Kafka に対する書き込みと同様の方法で行えます。 BOOTSTRAP_SERVERS と EH_SASL は、実際の Event Hubs 名前空間の情報に変更します。この構成を忘れずに更新してください。完全なサンプルコードについては、GitHub にある sparkProducer.scala ファイルを参照してください。

df = /**Dataframe**/

//Write to your Event Hub!
df.writeStream
    .format("kafka")
    .option("topic", TOPIC)
    .option("kafka.bootstrap.servers", BOOTSTRAP_SERVERS)
    .option("kafka.sasl.mechanism", "PLAIN")
    .option("kafka.security.protocol", "SASL_SSL")
    .option("kafka.sasl.jaas.config", EH_SASL)
    .option("checkpointLocation", "./checkpoint")
    .start()

次のステップ

Event Hubs と Kafka 用 Event Hubs の詳細については、次の記事を参照してください。