Apache Spark StreamingのtElasticSearchLookupInputプロパティ

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

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

[Use an existing configuration] (既存の設定を使用)

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

[Transport addresses] (住所の転送)

コンポーネントを接続する必要のあるElasticsearchノードのアドレスを入力します。

Elasticsearchノードクライアントを使用するtElasticsearchOutputとは異なり、tElasticSearchLookupInputはElasticsearchトランスポートクライアントを使ってElasticrsearchクラスターに接続します。これにより、tElasticSearchLookupInputはクラスターへの複数の接続をすばやく作成できます。

ElasticsearchノードクライアントとElasticsearchトランスポートクライアントについては、https://www.elastic.co/guide/en/elasticsearch/guide/current/_transport_client_versus_node_client.htmlをご覧ください。

[Cluster name] (クラスター名)

使用するElasticsearchクラスターの名前を入力します。

ElasticsearchノードクライアントとElasticsearchトランスポートクライアントについては、https://www.elastic.co/guide/en/elasticsearch/guide/current/_transport_client_versus_node_client.htmlをご覧ください。

Index

ドキュメントの読み取り先とするインデックスの名前を入力します。

インデックスは、Elasticsearchシステム内のストレージにおける最大のユニットです。

Type (タイプ)

読み取るドキュメントが属するタイプの名前を入力します。

たとえば、blogpost_enとblogpost_frをそれぞれ、指定された英語のブログ投稿とフランス語のブログ投稿を表す2つのタイプにできます。

ドキュメントタイプとする指定カラムの値を動的に使うことができます。そうする必要がある場合は、そのカラムの名前を波かっこ({})内に入力します(サンプル: {blog_author})。

クエリー

このコンポーネントによって実行するElasticsearchクエリーを入力します。

クエリーの編集では、Elasticsearchによって要求される構文とJavaによって要求されるエスケープ文字を使い、クエリーを二重引用符で囲む必要があります。

es.query = { "query" : { "term" : { "user" : "costinl" } } }

"{ \"query\" : { \"term\" : {\"user\" : \"costinl\" } } }"

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

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

[Scroll time] (スクロール時間)

入力バッチがElasticsearchから段階的にロードされる期間(ミリ秒)を入力します。

この期間は、クエリーが大量のバッチをもたらす場合にのみ役立ちます。ただし、ストリーミングモードの tMap は各行でデータをリロードするため、クエリーが適切に記述されていれば、巨大なバッチの生成を回避できます。

SSL/TLSの使用

SSLまたはTLS暗号化接続を有効にする場合は、このチェックボックスを選択します。

次に、同じジョブ内のtSetKeyStoreコンポーネントを使用して暗号化情報を指定する必要があります。

設定

アクションをさらにカスタマイズするには、Elasticsearchで受け入れられているパラメーターを追加します。

たとえば[Key] (キー)カラムにes.mapping.idと、[Value] (値)カラムにtrueとそれぞれ入力し、ドキュメントのフィールド/プロパティ名にドキュメントIDが含まれるようにします。入力する情報は二重引用符で囲む必要があります。

使えるパラメーターのリストは、https://www.elastic.co/guide/en/elasticsearch/hadoop/master/configuration.htmlをご覧ください。

[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] (アイドル接続の最小数)フィールドで定義したアイドル接続の最小数が維持されます。

使用ルール

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

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

特に明記していない限り、このドキュメンテーションのシナリオでは、標準ジョブ、つまり従来の Talend Data Integrationジョブだけを扱います。

[Spark Connection] (Spark接続)

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