Apache Spark StreamingのtMQTTInputプロパティ - 7.3

MQTT

Version
7.3
Language
日本語 (日本)
Product
Talend Data Fabric
Talend Real-Time Big Data Platform
Module
Talend Studio
Content
ジョブデザインと開発 > サードパーティーシステム > メッセージングコンポーネント > MQTT
データガバナンス > サードパーティーシステム > メッセージングコンポーネント > MQTT
データクオリティとプレパレーション > サードパーティーシステム > メッセージングコンポーネント > MQTT

このプロパティはSpark Streamingジョブフレームワークで実行されているtMQTTInputを設定するために使います。

Spark Streaming tMQTTInputコンポーネントはメッセージングファミリーのコンポーネントです。

このコンポーネントのストリーミングバージョンは、Talend Real Time Big Data PlatformおよびTalend Data Fabricで使用できます。

基本設定

[Broker URL] (ブローカーURL)

公開されたメッセージをサブスクライバー(tMQTTInputコンポーネント)にルーティングするために使われるMQTTブローカーのロケーションを入力します。

トピック

tMQTTInputにサブスクライブさせるトピックを入力します。

QoS

使うメッセージに割り当てるQoS (Quality of Service)の数値レベルを引用符で囲まずに入力します。

この品質レベルは、MQTTをメッセージ配信要求にどの程度応答させるかを示します。
  • 0: メッセージが配信されないか、1回だけ配信されることを意味します。

  • 1: メッセージが少なくとも1回配信されることを意味します。

  • 2: メッセージが1回だけ配信されることを意味します。

QoSのさまざまなレベルの説明は、http://www-01.ibm.com/support/knowledgecenter/SSFKSJ_8.0.0/com.ibm.mq.dev.doc/q029090_.htmを参照してください。

[Include topic column] (トピックカラムを含める)

このチェックボックスをオンにして、topicカラムをスキーマに追加し、トピックの名前とメッセージを次のコンポーネントに送信します。

Schema (スキーマ)およびEdit schema (スキーマを編集)

スキーマとは行の説明のことです。処理して次のコンポーネントに渡すフィールド(カラム)数を定義します。Sparkジョブを作成する場合、フィールドの命名時は予約語のlineを避けます。

このコンポーネントのスキーマは読み取り専用です。[Edit schema] (スキーマを編集)をクリックすると、スキーマを表示できます。

この読み取り専用のpayloadカラムは、処理するMQTTメッセージの本文を保持するために使われます。

入力メッセージの本文では、非常に異なるデータ形式が使えます。たとえば、形式がJSONの場合、tMQTTInputに続いてtExtractJSONFieldを使って、この本文から処理するデータを抽出する必要があります。

詳細設定

[Encoding] (エンコーディング)

リストからエンコーディングを選択するか、[CUSTOM] (カスタム)を選択して、手動で定義します。

このエンコーディングは、tMQTTInputが入力メッセージ配列をデコードするために使います。

使用方法

使用ルール

このコンポーネントは開始コンポーネントとして使用され、出力リンクを必要とします。

ランタイムにtMQTTInputコンポーネントはトピックをリッスンし続け、このトピックに新しいメッセージがバッファーリングされると読み取ります。

このコンポーネントは、所属するSpark Streamingコンポーネントのパレットと共に、Spark Streamingジョブを作成している場合にだけ表示されます。

特に明記していない限り、このドキュメントのシナリオでは、[Standard] (標準)ジョブ、つまり従来の Talend データ統合ジョブだけを扱います。

[Spark Connection] (Spark接続)

[Run] (実行)ビューの[Spark Configuration] (Spark設定)タブで、ジョブ全体でのSparkクラスターへの接続を定義します。また、ジョブでは、依存jarファイルを実行することを想定しているため、Sparkがこれらのjarファイルにアクセスできるように、これらのファイルの転送先にするファイルシステム内のディレクトリーを指定する必要があります。
  • Yarnモード(YarnクライアントまたはYarnクラスター):
    • Google Dataprocを使用している場合、[Spark configuration] (Spark設定)タブの[Google Storage staging bucket] (Google Storageステージングバケット)フィールドにバケットを指定します。

    • HDInsightを使用している場合、[Spark configuration] (Spark設定)タブの[Windows Azure Storage configuration] (Windows Azure Storage設定)エリアでジョブのデプロイメントに使用するブロブを指定します。

    • Altusを使用する場合は、[Spark configuration] (Spark設定)タブでジョブのデプロイにS3バケットまたはAzure Data Lake Storageを指定します。
    • Quboleを使用する場合は、ジョブにtS3Configurationを追加し、QuboleでS3システム内に実際のビジネスデータを書き込みます。tS3Configurationを使用しないと、このビジネスデータはQubole HDFSシステムに書き込まれ、クラスターをシャットダウンすると破棄されます。
    • オンプレミスのディストリビューションを使用する場合は、クラスターで使用されているファイルシステムに対応する設定コンポーネントを使用します。一般的に、このシステムはHDFSになるため、tHDFSConfigurationを使用します。

  • [Standalone mode] (スタンドアロンモード): tHDFSConfigurationまたはtS3Configurationなど、クラスターで使用されているファイルシステムに対応する設定コンポーネントを使用します。

    ジョブ内に設定コンポーネントがない状態でDatabricksを使用している場合、ビジネスデータはDBFS (Databricks Filesystem)に直接書き込まれます。

この接続は、ジョブごとに有効になります。

制限事項

ライセンスの互換性の問題のため、このコンポーネントの使用に必要な1つ以上のJARが提供されていません。この特定のコンポーネントに不足しているJARをインストールするには、Component (コンポーネント)タブビューの[Install] (インストール)ボタンをクリックします。Studioの Integration パースペクティブの[Modules] (モジュール)タブでも、不足しているすべてのJARを簡単に見つけて追加できます。詳細は、外部モジュールのインストールを参照してください。外部モジュールをインストールする方法の詳細は、Talend Help Center (https://help.talend.com)を参照してください。