Apache Kafka と Azure Databricks を使用したストリーム処理

[アーティクル]
04/18/2024

この記事では、Azure Databricks で構造化ストリーミングワークロードを実行するときに、Apache Kafka をソースまたはシンクとして使用する方法について説明します。

Kafka の詳細については、Kafka のドキュメントを参照してください。

Kafka からデータを読み取る

Kafka から読み取られたストリーミングの例を次に示します。

df = (spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "<server:ip>")
  .option("subscribe", "<topic>")
  .option("startingOffsets", "latest")
  .load()
)

Azure Databricks では、次の例に示すように、Kafka データソースのバッチ読み取りセマンティクスもサポートされています。

df = (spark
  .read
  .format("kafka")
  .option("kafka.bootstrap.servers", "<server:ip>")
  .option("subscribe", "<topic>")
  .option("startingOffsets", "earliest")
  .option("endingOffsets", "latest")
  .load()
)

増分バッチ読み込みの場合、Databricks では、Trigger.AvailableNow で Kafka を使用することをお勧めします。「増分バッチ処理の構成」を参照してください。

Databricks Runtime 13.3 LTS 以降では、Azure Databricks には Kafka データを読み取るための SQL 関数が用意されています。 SQL を使用したストリーミングは、Delta Live Tables または Databricks SQL のストリーミングテーブルでのみサポートされます。「read_kafka テーブル値関数」を参照してください。

Kafka 構造化ストリーミングリーダーを構成する

Azure Databricks には、Kafka 0.10 以降への接続を構成するためのデータ形式として kafka キーワードが用意されています。

Kafka の最も一般的な構成を次に示します。

サブスクライブするトピックを指定するには、複数の方法があります。次のパラメーターのいずれか 1 つのみを指定する必要があります。

オプション	値	説明
subscribe	トピックのコンマ区切りリスト。	サブスクライブするトピックの一覧。
subscribePattern	Java 正規表現文字列。	トピックのサブスクライブに使用するパターン。
assign	JSON 文字列 `{"topicA":[0,1],"topic":[2,4]}`。	使用する特定の topicPartition。

その他の注目すべき構成:

オプション	値	Default value	説明
kafka.bootstrap.servers	"ホスト:ポート" のコンマ区切りリストです。	empty	[必須] Kafka `bootstrap.servers` の構成。 Kafka からのデータがないことがわかった場合は、まず、ブローカーアドレスの一覧を確認します。ブローカーアドレスの一覧が正しくない場合、エラーは発生しないことがあります。これは、Kafka クライアントによって、ブローカーは最終的に利用可能になると想定され、ネットワークエラーが発生した場合は永久に再試行されるためです。
failOnDataLoss	`true` または `false`。	`true`	[省略可能] データが失われる恐れがある場合にクエリを失敗させるかどうか。クエリは、トピックの削除、処理前のトピックの切り捨てなど、多くのシナリオが原因で、Kafka からのデータの読み取りに永続的に失敗する可能性があります。データが失われた可能性があるかどうかについて控えめな推定を試みます。これにより、誤ってアラームが発生する場合があります。予期したとおりに動作しない場合、またはデータ損失にかかわらずクエリの処理を続行する場合は、このオプションを `false` に設定します。
minPartitions	>= 0 の整数、0 = 無効。	0 (無効)	[省略可能] Kafka から読み取るパーティションの最小数。 `minPartitions` オプションを使用して、Kafka から読み取るパーティションの最小数に任意の数を指定して Spark を構成することができます。通常、Spark では、Kafka の topicPartitions と、Kafka から使用する Spark パーティションが 1 対 1 でマッピングされます。 `minPartitions` オプションを、Kafka の topicPartitions よりも大きな値に設定すると、Spark では、大きな Kafka パーティションが小さな部分に分割されます。負荷のピーク時、データスキュー時、ストリームが遅れているときに、このオプションを設定すると、処理速度を上げることができます。トリガーごとに Kafka コンシューマーを初期化すると、Kafka に接続するときに SSL を使用する場合、パフォーマンスに影響を与える可能性があります。
kafka.group.id	Kafka コンシューマーグループ ID。	設定しない	[省略可能] Kafka から読み取り中に使用するグループ ID。注意して使用する必要があります。既定では、各クエリによって、データを読み取るための一意のグループ ID が生成されます。これにより、各クエリでは、独自のコンシューマーグループが使用され、他のコンシューマーによる干渉を受けないため、サブスクライブされたトピックのすべてのパーティションを確実に読み取ることができます。一部のシナリオ (たとえば、Kafka グループベースの承認) では、データを読み取るために特定の承認済みグループ ID を使用することが必要な場合があります。必要に応じて、グループ ID を設定できます。ただし、予期しない動作が発生する可能性があるため、細心の注意を払って設定してください。 * クエリ (バッチとストリーミングの両方) を同じグループ ID で同時に実行すると、相互に干渉し合い、各クエリでデータの一部しか読み取れなくなる可能性があります。 * これは、クエリを立て続けに起動または再起動した場合も発生する可能性があります。このような問題を最小限に抑えるには、Kafka コンシューマー構成 `session.timeout.ms` が非常に小さくなるように設定します。
startingOffsets	earliest、latest	latest	[省略可能] クエリが開始されたときの開始点。最も古いオフセットの "earliest" か、各 TopicPartition の開始オフセットを指定する JSON 文字列。 JSON では、オフセットとして -2 を使用して earliest を、-1 で latest を示すことができます。注: バッチクエリの場合、latest (暗黙的に、または JSON で -1 を使用して) は許可されません。ストリーミングクエリの場合、これは新しいクエリが開始されるときにのみ適用され、再開では常にクエリが中止された時点の情報が引き継がれます。クエリ中に新しく検出されたパーティションは、earliest で開始されます。

他の省略可能な構成については、「構造化ストリーミング + Kafka 統合ガイド」を参照してください。

Kafka レコードのスキーマ

Kafka レコードのスキーマは次のとおりです。

列	Type
key	binary
value	binary
topic	string
partition	int
offset	long
timestamp	long
timestampType	int

key と value は、ByteArrayDeserializer を使用して常にバイト配列として逆シリアル化されます。キーと値を明示的に逆シリアル化するには、DataFrame 操作 (cast("string") など) を使用します。

Kafka にデータを書き込む

Kafka へのストリーミング書き込みの例を次に示します。

(df
  .writeStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "<server:ip>")
  .option("topic", "<topic>")
  .start()
)

Azure Databricks では、次の例に示すように、Kafka データシンクへのバッチ書き込みセマンティクスもサポートされています。

(df
  .write
  .format("kafka")
  .option("kafka.bootstrap.servers", "<server:ip>")
  .option("topic", "<topic>")
  .save()
)

Kafka 構造化ストリーミングライターを構成する

重要

Databricks Runtime 13.3 LTS 以降には、既定でべき等の書き込みを有効にする新しいバージョンの kafka-clients ライブラリが含まれています。 Kafka シンクでバージョン 2.8.0 以下を使用していて、ACL が構成されているが IDEMPOTENT_WRITE が有効になっていない場合、書き込みはエラーメッセージ org.apache.kafka.common.KafkaException: Cannot execute transactional method because we are in an error state で失敗します。

このエラーを解決するには、Kafka バージョン 2.8.0 以降にアップグレードするか、構造化ストリーミングライターの構成時に .option(“kafka.enable.idempotence”, “false”) を設定します。

DataStreamWriter に提供されるスキーマは、Kafka シンクと対話します。次のフィールドを使用できます。

列名	必須または省略可能	Type
`key`	省略可能	`STRING` または `BINARY`
`value`	必須	`STRING` または `BINARY`
`headers`	省略可能	`ARRAY`
`topic`	省略可能 (`topic` がライターオプションとして設定されている場合は無視されます)	`STRING`
`partition`	省略可能	`INT`

Kafka への書き込み中に設定される一般的なオプションを次に示します。

オプション	値	Default value	説明
`kafka.boostrap.servers`	`<host:port>` のコンマ区切りリスト。	なし	[必須] Kafka `bootstrap.servers` の構成。
`topic`	`STRING`	設定しない	[省略可能] すべての行のトピックが書き込まれるように設定します。このオプションは、データに存在するすべてのトピック列をオーバーライドします。
`includeHeaders`	`BOOLEAN`	`false`	[省略可能] 行に Kafka ヘッダーを含めるかどうか。

他の省略可能な構成については、「構造化ストリーミング + Kafka 統合ガイド」を参照してください。

Kafka メトリックを取得する

avgOffsetsBehindLatest、maxOffsetsBehindLatest、minOffsetsBehindLatest メトリックを使用して、サブスクライブされたすべてのトピックの中で、ストリーミングクエリが利用可能な最新のオフセットより後であるオフセットの平均、最小、最大数を取得できます。「対話形式によるメトリックの読み取り」を参照してください。

注意

Databricks Runtime 9.1 以降で使用できます。

estimatedTotalBytesBehindLatest の値を調べることで、クエリプロセスによって消費されていない推定合計バイト数をサブスクライブされたトピックから取得します。この推定値は、過去 300 秒間に処理されたバッチ数に基づきます。推定値の基になる期間は、オプション bytesEstimateWindowLength を別の値に設定することによって変更できます。たとえば、10 分に設定するには、次のようにします。

df = (spark.readStream
  .format("kafka")
  .option("bytesEstimateWindowLength", "10m") # m for minutes, you can also use "600s" for 600 seconds
)

ノートブックでストリームを実行している場合、これらのメトリックは、ストリーミングクエリの進行状況ダッシュボードの [生データ] タブに表示されます。

{
  "sources" : [ {
    "description" : "KafkaV2[Subscribe[topic]]",
    "metrics" : {
      "avgOffsetsBehindLatest" : "4.0",
      "maxOffsetsBehindLatest" : "4",
      "minOffsetsBehindLatest" : "4",
      "estimatedTotalBytesBehindLatest" : "80.0"
    },
  } ]
}

SSL を使用して Azure Databricks を Kafka に接続する

Kafka への SSL 接続を有効にするには、Confluent のドキュメント「SSL による暗号化と認証」の手順に従います。オプションとして、そこで説明されている構成を、プレフィックス kafka. を付けて指定できます。たとえば、プロパティ kafka.ssl.truststore.location で、信頼ストアの場所を指定します。

Databricks では、次のことが推奨されています。

証明書をクラウドオブジェクトストレージに格納します。証明書へのアクセスは、Kafka にアクセスできるクラスターのみに制限できます。「Unity Catalog を使用したデータガバナンス」を参照してください。
証明書のパスワードをシークレットとしてシークレットスコープに格納します。

次の例では、オブジェクトストレージの場所と Databricks シークレットを使用して SSL 接続を有効にします。

df = (spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", ...)
  .option("kafka.security.protocol", "SASL_SSL")
  .option("kafka.ssl.truststore.location", <truststore-location>)
  .option("kafka.ssl.keystore.location", <keystore-location>)
  .option("kafka.ssl.keystore.password", dbutils.secrets.get(scope=<certificate-scope-name>,key=<keystore-password-key-name>))
  .option("kafka.ssl.truststore.password", dbutils.secrets.get(scope=<certificate-scope-name>,key=<truststore-password-key-name>))
)

HDInsight 上の Kafka を Azure Databricks に接続する

HDInsight Kafka クラスターを作成します。

手順については、「Azure Virtual Network 経由で HDInsight 上の Apache Kafka に接続する」を参照してください。
正しいアドレスをアドバタイズするように Kafka ブローカーを構成します。

「IP をアドバタイズするように Kafka を構成する」の手順に従います。 Azure Virtual Machines で Kafka を自分で管理する場合、ブローカーの advertised.listeners 構成がホストの内部 IP に設定されていることを確認してください。
Azure Databricks クラスターを作成します。
Kafka クラスターを Azure Databricks クラスターにピアリングします。

「仮想ネットワークをピアリングする」の手順に従います。

Microsoft Entra ID (旧称 Azure Active Directory) でのサービスプリンシパル認証と Azure Event Hubs

Azure Databricks は、Event Hubs サービスを使用した Spark ジョブの認証をサポートしています。この認証は、Microsoft Entra ID (旧称 Azure Active Directory) を使用して OAuth 経由で行われます。

AAD 認証のダイアグラム

Azure Databricks では、以下のコンピューティング環境で、クライアント ID とシークレットを使用した Microsoft Entra ID 認証をサポートしています。

シングルユーザーアクセスモードで構成されたコンピューティング上の Databricks Runtime 12.2 LTS 以降。
共有アクセスモードで構成されたコンピューティング上の Databricks Runtime 14.3 LTS 以降。
Unity Catalog を使用せずに構成された Delta Live Tables パイプライン。

Azure Databricks では、コンピューティング環境内または Unity Catalog を使用して構成された Delta Live Tables パイプラインで、証明書を使用した Microsoft Entra ID 認証はサポートされていません。

この認証は、共有クラスターや Unity Catalog の Delta Live Tables では機能しません。

構造化ストリーミング Kafka コネクタの構成

Microsoft Entra ID で認証を実行するには、次の値が必要です。

テナント ID。これは、Microsoft Entra ID の [サービス] タブにあります。
clientID (アプリケーション ID とも呼ばれます)。
クライアントシークレット。これを取得したら、シークレットとして Databricks ワークスペースに追加する必要があります。このシークレットを追加するには、「シークレットの管理」を参照してください。
EventHubs トピック。トピックの一覧は、特定の Event Hubs 名前空間ページの [エンティティ] セクションの [Event Hubs] セクションにあります。複数のトピックを操作するために、Event Hubs レベルで IAM ロールを設定できます。
EventHubs サーバー。これは、特定の Event Hubs 名前空間の概要ページにあります。

さらに、Entra ID を使用するには、OAuth SASL メカニズム (SASL は汎用プロトコルであり、OAuth は SASL "メカニズム" の一種です) を使用するように Kafka に指示する必要があります。

kafka.security.protocol は、SASL_SSL である必要があります。
kafka.sasl.mechanism は、OAUTHBEARER である必要があります。
kafka.sasl.login.callback.handler.class は、シェーディングされた Kafka クラスのログインコールバックハンドラーについて値 kafkashaded を持つ Java クラスの完全修飾名にする必要があります。正確なクラスについては、次の例を参照してください。

例

次に、実行例を見てみましょう。

Python

# This is the only section you need to modify for auth purposes!
# ------------------------------
tenant_id = "..."
client_id = "..."
client_secret = dbutils.secrets.get("your-scope", "your-secret-name")

event_hubs_server = "..."
event_hubs_topic = "..."
# -------------------------------

sasl_config = f'kafkashaded.org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required clientId="{client_id}" clientSecret="{client_secret}" scope="https://{event_hubs_server}/.default" ssl.protocol="SSL";'

kafka_options = {
# Port 9093 is the EventHubs Kafka port
"kafka.bootstrap.servers": f"{event_hubs_server}:9093",
"kafka.sasl.jaas.config": sasl_config,
"kafka.sasl.oauthbearer.token.endpoint.url": f"https://login.microsoft.com/{tenant_id}/oauth2/v2.0/token",
"subscribe": event_hubs_topic,

# You should not need to modify these
"kafka.security.protocol": "SASL_SSL",
"kafka.sasl.mechanism": "OAUTHBEARER",
"kafka.sasl.login.callback.handler.class": "kafkashaded.org.apache.kafka.common.security.oauthbearer.secured.OAuthBearerLoginCallbackHandler"
}

df = spark.readStream.format("kafka").options(**kafka_options)

display(df)

Scala

// This is the only section you need to modify for auth purposes!
// -------------------------------
val tenantId = "..."
val clientId = "..."
val clientSecret = dbutils.secrets.get("your-scope", "your-secret-name")

val eventHubsServer = "..."
val eventHubsTopic = "..."
// -------------------------------

val saslConfig = s"""kafkashaded.org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required clientId="$clientId" clientSecret="$clientSecret" scope="https://$eventHubsServer/.default" ssl.protocol="SSL";"""

val kafkaOptions = Map(
// Port 9093 is the EventHubs Kafka port
"kafka.bootstrap.servers" -> s"$eventHubsServer:9093",
"kafka.sasl.jaas.config" -> saslConfig,
"kafka.sasl.oauthbearer.token.endpoint.url" -> s"https://login.microsoft.com/$tenantId/oauth2/v2.0/token",
"subscribe" -> eventHubsTopic,

// You should not need to modify these
"kafka.security.protocol" -> "SASL_SSL",
"kafka.sasl.mechanism" -> "OAUTHBEARER",
"kafka.sasl.login.callback.handler.class" -> "kafkashaded.org.apache.kafka.common.security.oauthbearer.secured.OAuthBearerLoginCallbackHandler"
)

val scalaDF = spark.readStream
  .format("kafka")
  .options(kafkaOptions)
  .load()

display(scalaDF)

潜在的なエラーの処理

ストリーミングオプションがサポートされません。

Unity Catalog を使用して構成された Delta Live Tables パイプラインでこの認証メカニズムを使用しようとすると、次のエラーが表示されることがあります。

このエラーを解決するには、サポートされているコンピューティング構成を使用します。「Microsoft Entra ID (旧称 Azure Active Directory) と Azure Event Hubs を使用したサービスプリンシパル認証」を参照してください。
新しい KafkaAdminClient の作成に失敗しました。

これは、次のいずれかの認証オプションが正しくない場合に Kafka がスローする内部エラーです。
- クライアント ID (アプリケーション ID とも呼ばれます)
- テナント ID
- Event Hubs サーバー
エラーを解決するには、これらのオプションの値が正しいことを確認します。

また、この例で既定で提供されている (変更しないように求められた) kafka.security.protocol などの構成オプションを変更すると、このエラーが表示される場合があります。
返されるレコードがありません

データフレームを表示または処理しようとしているが結果が得られない場合は、UI に次の情報が表示されます。

このメッセージは、認証は成功したが、Event Hubs がデータを返さなかったことを意味します。次のような理由が考えられます (ただし、決して網羅的ではありません)。
- 正しくない EventHubs トピックを指定しました。
- startingOffsets の既定の Kafka 構成オプションが latest であり、現在、トピック経由でデータをまだ受信していません。 Kafka の最も古いオフセットからデータの読み取りを開始するように、startingOffsetstoearliest を設定できます。

次の方法で共有

Apache Kafka と Azure Databricks を使用したストリーム処理

Kafka からデータを読み取る

Kafka 構造化ストリーミングリーダーを構成する

Kafka レコードのスキーマ

Kafka にデータを書き込む

Kafka 構造化ストリーミングライターを構成する

Kafka メトリックを取得する

SSL を使用して Azure Databricks を Kafka に接続する

HDInsight 上の Kafka を Azure Databricks に接続する

Microsoft Entra ID (旧称 Azure Active Directory) でのサービスプリンシパル認証と Azure Event Hubs

構造化ストリーミング Kafka コネクタの構成

例

Python

Scala

潜在的なエラーの処理

フィードバック

フィードバック

その他のリソース

次の方法で共有

Apache Kafka と Azure Databricks を使用したストリーム処理

Kafka からデータを読み取る

Kafka 構造化ストリーミング リーダーを構成する

Kafka レコードのスキーマ

Kafka にデータを書き込む

Kafka 構造化ストリーミング ライターを構成する

Kafka メトリックを取得する

SSL を使用して Azure Databricks を Kafka に接続する

HDInsight 上の Kafka を Azure Databricks に接続する

Microsoft Entra ID (旧称 Azure Active Directory) でのサービス プリンシパル認証と Azure Event Hubs

構造化ストリーミング Kafka コネクタの構成

例

Python

Scala

潜在的なエラーの処理

フィードバック

フィードバック

その他のリソース

Kafka 構造化ストリーミングリーダーを構成する

Kafka 構造化ストリーミングライターを構成する

Microsoft Entra ID (旧称 Azure Active Directory) でのサービスプリンシパル認証と Azure Event Hubs