EMRを使って少し大きめなボリュームのデータ処理をしているのですが、その中のあるデータの中身をWebフォームからニアリアルタイムでフィルタリングしたいと言う要望があり、その基盤としてElasticsearchを採用する事にしました。管理のドメインを使用ロードバランサー越しにしか通信できないプロビジョンしたElasticsearchのバージョンは執筆時点で最新版の6.7Elasticsearchについてデータの投入方法以前に、ユースケースとして事前に想定ボリューム数で検証した結果、Elasticsearchであれば(ボリューム的に当然ですが)問題なく検索が可能と言う事が分かったので事前に決定していました。ESで管理されています。Sparkについて投入方法については先述の要件が満たせれば何でも良かったのですが、色々試した結果、そもそもEMRでのデータ処理自体にも使っていた事もあって、Sparkが良さそうと言う事になり、これに決定しました。Hadoopと言うインテグレーションライブラリを使いました。ここからダウンロードが可能です。6.7.2を使いました。※実際のアプリケーションではbuild.sbtとかで依存管理をしますが今回はテストなので直接👆のjarを使います。https://mvnrepository.com/artifact/org.elasticsearch/elasticsearch-hadoop/6.7.2試してみる事前に、先程👆でダウンロードしたJARをS3の適当なバケットにアップロードしておきます。elasticsearch-hadoop-6.7.2.jarその後、EMRで適当にSparkのクラスタを作成します。選択しています:EMRクラスタ作成画面クラスタがReadyになったら、SSHでログインします:早速spark-shellを起動します。接頭辞を付ける事で、インテグレーションAPIの設定を指定する事ができます。言うkeyで設定が行われます。ESとの疎通では必須となる設定です。ようなドメインのロードバランサーを経由する必要があります。ポートについてもデフォルトの9200ではなく80が使われてるので、指定しておく必要があります。明示しておかないとエラーが発生します。設定はSparkアプリケーションを起動後も個別に設定を指定する事ができます(後述)が、こう言う基本的な設定は起動時に設定しても良いでしょう。_docに保存していますこれで、2行のレコードが追加されました。}}データを読み込んでみる反対に、Elasticsearchに入っているデータからDataFrameを作ってみたいと思います。Alice))無事取得する事ができました。実行時にいくつか設定を指定したと思いますが、設定は各オペレーションを実行時にもオーバーライド指定する事が可能です。マッピングし、データを更新可能にしたいとします。実行時に指定した設定とマージされます。どのフィールドを書き込み対象から外すかをそれぞれ指定できます。}}想定した通りデータが入っています。指定しているので、既存データの更新を行う事ができます。読み込み時にも同様の事が可能です。含めないようにしましたが、このままだとDataFrame側でidが参照できず不便な事があります。言うオブジェクトが入れ子になっています。APIを呼び出す際のサイズで、コンポーズしたJSONエントリがいずれかのサイズに到達したらAPIがコールされると言う内容になっており、当然ながら大きいサイズを指定する事でElasticsearchへの接続数が減ります。設定値についてはドキュメントサイズの傾向や、クラスタの設定、ネットワークの構成、Spark上でのタスク数に依存する為正解はありませんが、エラーが頻発しない範囲で、可能な限り大きい数値を使って調整しています。APIを呼び出さないといけなくなり、現実的ではないのでこれも可能な限り大きい値を設定しておく必要があります。※パフォーマンスについては公式ドキュメントのこのページが詳しいです。まとめElasticsearchが提供するインテグレーションを使うと、Sparkとの連携が楽に出ると言う事が分かりました。現在はまだ本格的な運用フェーズに入っていない為細かいチューニング等はまだしていないので、また別の機会でそのあたりをいつかご紹介できればと思います。優秀な技術者と一緒に、好きな場所で働きませんか株式会社もばらぶでは、優秀で意欲に溢れる方を常に求めています。働く場所は自由、働く時間も柔軟に選択可能です。現在、以下の職種を募集中です。ご興味のある方は、リンク先をご参照下さい。採用情報プロジェクトマネージャー(リモート)ソフトウェアエンジニア(リモート)デザイナー(リモート)Subscriptionメールアドレスを記入して購読すれば、更新をメールで受信できます。ThemePacific。