Google Pub/Sub にサブスクライブする

組み込みのコネクタを使用して、Google Pub/Sub をサブスクライブします。 このコネクタは、サブスクライバーから受信する行に対して「正確に 1 回」の処理セマンティクスを持っています。

注

Pub/Sub が重複する行を発行したり、行がサブスクライバーに順不同で到着したりする可能性があります。 重複した行や順不同の行に対応するコードを記述する必要があります。

Pub/Sub ストリームを構成する

次のコード例は、Pub/Sub から読み取られた構造化ストリーミングを構成し、秘密キーで認証する方法を示しています。

Python

auth_options = {
    "clientId": client_id,
    "clientEmail": client_email,
    "privateKey": private_key,
    "privateKeyId": private_key_id
}

query = (spark.readStream
  .format("pubsub")
  .option("subscriptionId", "mysub")
  .option("topicId", "mytopic")
  .option("projectId", "myproject")
  .options(auth_options)
  .load()
)

Scala

val authOptions: Map[String, String] =
  Map("clientId" -> clientId,
      "clientEmail" -> clientEmail,
      "privateKey" -> privateKey,
      "privateKeyId" -> privateKeyId)

val query = spark.readStream
  .format("pubsub")
  // Creates a Pub/Sub subscription if one does not already exist with this ID
  .option("subscriptionId", "mysub")
  .option("topicId", "mytopic")
  .option("projectId", "myproject")
  .options(authOptions)
  .load()

SQL

CREATE OR REFRESH STREAMING TABLE pubsub_raw
AS SELECT * FROM STREAM read_pubsub(
  subscriptionId => 'mysub',
  projectId => 'myproject',
  topicId => 'mytopic',
  clientEmail => secret('pubsub-scope', 'clientEmail'),
  clientId => secret('pubsub-scope', 'clientId'),
  privateKeyId => secret('pubsub-scope', 'privateKeyId'),
  privateKey => secret('pubsub-scope', 'privateKey')
);

その他の構成オプションについては、「Pub/Sub ストリーミング読み取りのオプションを構成する」を参照してください。

Pub/Sub へのアクセスを構成する

認証情報には、次のロールが付与されている必要があります。

役割 必須または省略可能 ロールの使用方法
roles/pubsub.viewer または roles/viewer 必須 サブスクリプションが存在するかどうかを確認し、サブスクリプションを取得します。
roles/pubsub.subscriber 必須 サブスクリプションからデータを取得します。
roles/pubsub.editor または roles/editor 省略可能 サブスクリプションが存在しない場合はサブスクリプションの作成を有効にし、 deleteSubscriptionOnStreamStop を使用してストリーム終了時にサブスクリプションを削除できるようにします。

注

プロジェクト レベルではなくリソース レベルで roles/pubsub.viewer と roles/pubsub.subscriber を付与する場合は、トピックとサブスクリプションの両方に両方のロールを適用する必要があります。 省略可能な roles/pubsub.editor または roles/editor ロールを使用しない場合、トピックだけに必要なロールを付与するだけでは不十分です。

Databricks では、キーを使用するときにシークレットを使用することをお勧めします。 接続を承認するには、次のオプションが必要です:

  • clientEmail
  • clientId
  • privateKey
  • privateKeyId

Pub/Sub スキーマについて

ストリームのスキーマは、次の表に示すように、Pub/Sub から取得される行に対応しています。

フィールド タイプ
messageId StringType
payload ArrayType[ByteType]
attributes StringType
publishTimestampInMillis LongType

Pub/Sub ストリーミング読み取りのオプションを構成する

一部の Pub/Sub 構成オプションでは、マイクロバッチ ではなく、フェッチ 概念を使用します。 これは内部実装の詳細であり、オプションは他の Structured Streaming コネクタと同様に機能しますが、行がフェッチされて処理される点が異なります。

オプションの完全な一覧については、 Pub/Sub を参照してください。

Pub/Sub で増分バッチ処理を使用する

Trigger.AvailableNowを使用して、Pub/Sub ソースの使用可能な行を増分バッチとして使用できます。

Azure Databricks では、Trigger.AvailableNow 設定で読み取りを開始したときにタイムスタンプが記録されます。 バッチで処理される行には、以前に取得されたすべてのデータと、記録された開始タイムスタンプより前のタイムスタンプを持つ新たに公開された行が含まれます。 詳細については、「 AvailableNow: 増分バッチ処理」を参照してください。

Pub/Sub ストリーミング メトリックを監視する

構造化ストリーミングの進行状況メトリックは、フェッチされて処理できる行の数、フェッチされて処理できる状態の行のサイズ、およびストリームの開始後に表示された重複の数を報告します。

Pub/Sub メトリックの例を次に示します。

"metrics" : {
  "numDuplicatesSinceStreamStart" : "1",
  "numRecordsReadyToProcess" : "1",
  "sizeOfRecordsReadyToProcess" : "8"
}

制限事項

Pub/Sub では、 spark.speculationを使用した投機的な実行はサポートされていません。

  • Last updated on