Apache Spark StreamingのtMongoDBInputプロパティ - 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ジョブフレームワークで実行するtMongoDBInputの設定で使用されます。

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

この種類のジョブでは、ルックアップデータのサイズがジョブの実行に割り当てられたメモリ容量に収まっている場合、ルックアップデータを提供するためにtMongDBInputが使用されます。MongoDBからのデータ読み取りで一度実行され、データをメモリに保管してメインフローからのマイクロバッチでデータに簡単にアクセスできるようにします。ルックアップデータが大きすぎてメモリに保管できない場合は、代わりに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テーブルと同等であり、そこにはドキュメントが含まれています。

使用するコレクションがシャードでない場合は、mongo.input.split_sizeプロパティを[Advanced Hadoop MongoDB properties] (詳細Hadoop MongoDBプロパティ)テーブルに追加することをお勧めします。このパラメーターは、コレクションがどのようにパーティション分割され、Sparkエグゼキューターによって読み取られるかを決定します。入力コレクションのパーティション数は、次の式を使用して計算できます。
Number of partitions = Collection size in MB / mongo.input.split_size
このプロパティがない場合、Sparはパーティションのサイズにデフォルト値の8 MBを使用します。
たとえば次のような質問です:
mongo.input.split_size   1
このサンプルでは、Sparkは非シャードコレクションを並列で読み取るために、各Sparkエグゼキューターに1 MBを転送します。コレクションのサイズが10 MBの場合は、10エグゼキューターが使用されます。

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

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

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

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

[Query] (クエリー)

クエリーステートメントを指定して、[Collection] (コレクション)フィールドで指定したコレクションからドキュメントを選択します。たとえば、"{'id':'4'}"と入力して、id4のレコードをコレクションから取得します。

このコンポーネントで提供されている二重引用符で囲まれたデフォルトクエリー{}は、すべてのファイルを選択することを意味します。{'filename':{'$regex':'REGEX_PATTERN'}}を使用することで正規表現を適用し、使用するファイル名を定義することもできます。

MongoDBクライアントソフトウェアで求められるクエリーステートメントとは異なり、ここのクエリーはfind()内のコンテンツを参照します。ここのクエリー{'filename':{'$regex':'REGEX_PATTERN'}}などは、MongoDBクライアントクエリーのdb.blog.find({filename:{$regex:REGEX_PATTERN}})に相当します。

Mapping (マッピング)

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

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

[Limit] (制限事項)

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

詳細設定

[Advanced Hadoop MongoDB properties] (高度なHadoop MongoDBのプロパティ)

プロパティを追加して、tMongoDBInputにデータの読み取り時に実行させる追加のオペレーションを定義します。

使用可能なプロパティは、MongoDB Connector for Hadoopにリスト表示され、説明されています。

使用するコレクションがシャードでない場合は、mongo.input.split_sizeプロパティを[Advanced Hadoop MongoDB properties] (詳細Hadoop MongoDBプロパティ)テーブルに追加することをお勧めします。このパラメーターは、コレクションがどのようにパーティション分割され、Sparkエグゼキューターによって読み取られるかを決定します。入力コレクションのパーティション数は、次の式を使用して計算できます。
Number of partitions = Collection size in MB / mongo.input.split_size
このプロパティがない場合、Sparはパーティションのサイズにデフォルト値の8 MBを使用します。
たとえば次のような質問です:
mongo.input.split_size   1
このサンプルでは、Sparkは非シャードコレクションを並列で読み取るために、各Sparkエグゼキューターに1 MBを転送します。コレクションのサイズが10 MBの場合は、10エグゼキューターが使用されます。

使用方法

使用ルール

このコンポーネントは、ルックアップデータのサイズがジョブの実行に割り当てられたメモリ容量に収まっている場合、ルックアップデータを提供するために使用されます。MongoDBからのデータ読み取りで一度実行され、データをメモリに保管してメインフローからのマイクロバッチでデータに簡単にアクセスできるようにします。ルックアップデータが大きすぎてメモリに保管できない場合は、代わりにtMongoDBLookupInputを使用することをお勧めします。ルックアップ結合キーに一致するデータのみが読み込むまれます。

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

このコンポーネントは、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)に直接書き込まれます。

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