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

MongoDB

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

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

Spark Streaming tMongoDBLookupInputコンポーネントはデータベースファミリーのコンポーネントです。

このフレームワーク内のコンポーネントは、Talend Real Time Big Data PlatformおよびTalend Data Fabricで使用できます。

基本設定

[Property type] (プロパティタイプ)

[Built-In] (組み込み)または[Repository] (リポジトリー)のいずれか。

[Built-In] (組み込み): プロパティデータは一元的に保存されません。

[Repository] (リポジトリー): プロパティを保存するリポジトリーファイルを選択します。

[MongoDB configuration] (MongoDB設定)

定義済みの接続の詳細を再利用するには、このチェックボックスをオンにして、[Component List] (コンポーネントリスト)ドロップダウンリストで目的の接続コンポーネントを選択します。

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

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

スキーマを変更するには[Edit schema] (スキーマを編集)をクリックします。現在のスキーマがリポジトリータイプの場合は、3つのオプションを使用できます。

  • [View schema] (スキーマの表示): スキーマのみを表示する場合は、このオプションを選択します。

  • [Change to built-in property] (組み込みのプロパティに変更): ローカルで変更を行うためにスキーマを組み込みに変更する場合は、このオプションを選択します。

  • [Update repository connection] (リポジトリー接続を更新): リポジトリーに保存されているスキーマに変更を加え、変更後にそのコンテンツをすべてのジョブにプロパゲートするかどうかを決める場合は、このオプションを選択します。変更を現在のジョブにのみ反映する場合は、変更後、[No] (いいえ)を選択し、[Repository Content] (リポジトリーのコンテンツ)ウィンドウで再びこのスキーマのメタデータを選択します。

データベースのカラムがJSONドキュメントで、ドキュメント全体を読み取る必要がある場合は、引用符を使用せずに[DB column] (DBカラム)内でアスタリスク(*)を付けます。

[Collection] (コレクション)

使用するコレクションの名前を入力します。

MongoDBコレクションはRDBMSテーブルと同等であり、そこにはドキュメントが含まれています。

[Set read preference] (読み取り環境設定の設定)

このチェックボックスをオンにして、[Read preference] (読み取り環境設定)ドロップダウンリストが表示されたら、読み取りオペレーションをダイレクトする必要のあるメンバーを選択します。

このチェックボックスをオフのままにすると、ジョブはデフォルトの読み取り環境設定を使用します。すなわち、レプリカセット内のプライマリーメンバーが使用されます。

詳細は、MongoDBのドキュメンテーションで複製とその読み取り環境設定に関する説明を参照してください。

[Query] (クエリー)

クエリーステートメントを指定して、[Collection] (コレクション)フィールドで指定したコレクションからドキュメントを選択します。

たとえば、以下のようにします
"{'customer_id':" + row1.customer_id +"}"

このコードでrow1は、tMongoDBLookupInputのリンクのラベルではなく、tMapに入るメイン行を表します。

クエリーの結果には、tMapで使用する結合キーに一致するレコードのみ含まれていなければなりません。言い換えれば、一致するレコードのみがルックアップフローに読み込まれるSQLステートメントを作成するため、メインフローのスキーマをtMapで使用する必要があります。

この方式では、重複レコードはメモリに読み込まれず、後続のコンポーネントに出力されます。

Mapping (マッピング)

このコンポーネント用に定義されたスキーマの各カラムは、読み取るドキュメントのフィールドを表します。このテーブルで、これらのフィールドの親ノードがあればそれを指定する必要があります。

たとえば、次のようなドキュメントがあるとします。
{
               _id: ObjectId("5099803df3f4948bd2f98391"),
               person: { first: "Joe", last: "Walker" }
            }
フィールドfirstおよびlastには親ノードpersonがありますが、_idフィールドには親ノードがありません。したがって、完了すると、この[Mapping]テーブルは次のようになります。
Column     Parent node path
_id
first       "person"
last        "person"

[Limit] (制限事項)

取得するレコードの最大数を入力します。

詳細設定

[No query timeout] (クエリータイムアウトなし)

アイドルカーソルの非アクティブ状態が10分間続いた時にMongoDBサーバーがカーソルを停止することを防ぐには、このチェックボックスをオンにします。この状況では、このカーソルの結果が枯渇するか、またはcursor.close()メソッドを使用して手動で閉じるまで、アイドルカーソルが開いたままとなります。

MongoDBのカーソルはクエリーの結果セットを指すポインターです。デフォルトでは、すなわち、このチェックボックスがオフの場合、MongoDBサーバーは、メモリの過剰使用を避けるために、非アクティブ状態が特定の期間続くと、アイドルカーソルを自動的に停止します。MongoDBカーソルの詳細は、https://docs.mongodb.org/manual/core/cursors/を参照してください。

使用方法

使用ルール

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

このコンポーネントは、MongoDBデータベースに接続するために同じジョブ内にあるtMongoDBConfigurationコンポーネントを使用する必要があります。tMongoDBConfigurationを使用するには、このコンポーネントと共にtMongoDBConfigurationコンポーネントをドロップし、このコンポーネントの[Basic settings] (基本設定)を指定する必要があります。

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

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

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