Apache Spark StreamingのtPubSubOutputプロパティ - 7.2

Google PubSub

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

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

Spark Streaming tPubSubOutputコンポーネントは、メッセージングファミリーに属しています。

このコンポーネントは、Talend Real Time Big Data PlatformおよびTalend Data Fabricで利用できます。

基本設定

[Schema] (スキーマ)[Edit schema] (スキーマを編集)

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

このコンポーネントのスキーマは読み取り専用です。公開するメッセージを保管します。

[Define a Goolge Cloud configuration component] (Goolge Cloudの設定コンポーネントを定義)

SparkクラスターにDataprocを使っている場合は、このチェックボックスをオフにします。

それ以外の場合は、このチェックボックスをオンにして、Pub/SubコンポーネントがtGoogleCloudConfigurationコンポーネントの提供するGoogle Cloud設定情報を使えるようにします。

トピック名

メッセージを公開するトピックの名前を入力します。このトピックは既に存在している必要があります。

トピック操作

指定したトピックに対して実行する操作を選択します。
  • None: 使うトピックが既に存在する場合は、このオプションを選択します。

  • Create if not exists: 使うトピックが存在しない場合は、このオプションを選択します。

詳細設定

[Connection pool] (接続プール)

このエリアでは、各Sparkエグゼキューターに、同時に開いたままにする接続の数を制御するための接続プールを設定します。以下の接続プールパラメーターに与えられているデフォルト値は、ほとんどのユースケースでそのまま利用できます。

  • [Max total number of connections] (接続の最大合計数): 同時に開いたままにしておくことができる接続(アイドルまたはアクティブ)の最大数を入力します。

    デフォルトの数は8です。-1を入力すると、同時に開いておける接続の数が無制限となります。

  • [Max waiting time (ms)] (最大待機時間(ミリ秒)): 接続使用の要求に対して接続プールからレスポンスが返されるまでの最大待機時間を入力します。デフォルトでは-1(無制限)となっています。

  • [Min number of idle connections] (アイドル接続の最小数): 接続プール内に維持されるアイドル接続(使用されていない接続)の最小数を入力します。

  • [Max number of idle connections] (アイドル接続の最大数): 接続プール内に維持されるアイドル接続(使用されていない接続)の最大数を入力します。

[Evict connections] (接続の無効化)

接続プール内の接続を破棄する条件を定義するには、このチェックボックスをオンにします。オンにすると、以下のフィールドが表示されます。

  • [Time between two eviction runs] (2つの削除実行の間隔): コンポーネントが接続のステータスを確認し、アイドル状態の接続を破棄するまでの間隔(ミリ秒)を入力します。

  • [Min idle time for a connection to be eligible to eviction] (接続が削除可能になるまでの最小アイドル時間): アイドル接続が破棄されるまでの間隔(ミリ秒)を入力します。

  • [Soft min idle time for a connection to be eligible to eviction] (接続が削除可能になるまでのソフト最小アイドル時間): このパラメーターの機能は[Min idle time for a connection to be eligible to eviction] (接続が削除可能になるまでの最小アイドル時間)と同じですが、[Min number of idle connections] (アイドル接続の最小数)フィールドで定義したアイドル接続の最小数が維持されます。

使用方法

使用ルール

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

このコンポーネントがシリアライズされたデータを送信するには、入力スキーマにserializedValueカラムを定義するためにtWriteJSONFieldなどの書き込みコンポーネントを必要とします。

[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] (スタンドアロンモード): tHDFSConfigurationtS3Configurationなど、クラスターで使われているファイルシステムに対応する設定コンポーネントを使用します。

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

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

PubSubアクセス許可

DataprocクラスターでPub/Subを使う時は、このクラスターにPub/Subサービスにアクセスするための適切な権限があることをご確認ください。

これを行うには、Google Cloud Platformの詳細オプションで同じプロジェクト内の[Allow API access to all Google Cloud services] (すべてのGoogle CloudサービスへのAPIアクセスを許可する)をオンにするか、コマンドラインを使ってスコープを明示的に割り当てることで、Dataprocクラスターを作成します(次の例は低リソースのテストクラスター用です):
gcloud beta dataproc clusters create CLUSTER_ID \
    --zone europe-west1-b \
    --master-machine-type n1-standard-2 \
    --master-boot-disk-size 50 \
    --num-workers 2 \
    --worker-machine-type n1-standard-2 \
    --worker-boot-disk-size 50 \
    --scopes 'https://www.googleapis.com/auth/cloud-platform' \
    --project PROJECT_ID