この記事はただの集団 Advent Calendar 2019の7日目の記事です。

背景

Elasticsearch(以下、ESと呼びます)はREST APIを提供しており、私自身もちょっとしたことならばシェルスクリプトやpython等の軽量言語で記述することが多いです。
しかし、複雑な処理を扱う場合、例としてSearchを用いる場合はscoreも含めて構成要素が多くなるため、もう少し楽に書きたいというのがあります。

Scalaを使うメリット

仕事ではScalaをメインに開発していますが、私自身がScalaに対して良いと思うのは以下のような点です。

OptionやCollection操作が扱いやすい
型安全・型推論が便利
強力なパターンマッチ

一言でいうと、「パワフルな言語」という位置づけです。

今回紹介するもの

elastic4sはESへの操作をScalaのDSLで記述できるライブラリです。TCPも提供していますが、今回はHTTPについて触れることにします。

サンプルアプリケーション

ESへの検索をWebアプリケーション上で実行するケースを想定して書いていきます。
elastic4sは、執筆時点での最新版である7.3.1を例とします。まず、build.sbtに以下のように記述します。

scalaVersion := "2.13.1"
val elastic4sVersion = "7.3.1"

libraryDependencies ++= Seq(
  "com.sksamuel.elastic4s" %% "elastic4s-core" % elastic4sVersion,
  "com.sksamuel.elastic4s" %% "elastic4s-client-esjava" % elastic4sVersion,
  "com.sksamuel.elastic4s" %% "elastic4s-http-streams" % elastic4sVersion,
  "com.sksamuel.elastic4s" %% "elastic4s-json-play" % elastic4sVersion
)

elastic4s-json-playは必須ではないですが(後述)、playframeworkとplay-jsonを扱うならば選択肢として入れておくと良いと思います。

ローカル環境のES構築

Docker

公式ドキュメントのversion7.3を例にdocker-compose.ymlを記述し、起動すればよいです。

Mapping

データの題材は迷ったのですが、駅を検索できるようなものを想定します。

{
  "mappings": {
    "properties": {
      "stationCode": {
        "type": "keyword"
      },
      "stationName": {
        "type": "keyword"
      },
      "prefectureCode": {
        "type": "keyword"
      },
      "address": {
        "type": "text"
      },
      "point": {
        "type": "geo_point"
      }
    }
  }
}

↑の内容を用いてindexを作成します。

$ curl -XPUT http://localhost:9200/station/ -H 'Content-Type:application/json' -d @mapping.json

Mappingの詳しい内容は、公式ドキュメントを参照あれ。

データ投入

今回はSearchをメインとするため、割愛します。私の手元では駅データJPのものを用いて行いました。

Scalaのサンプルコード

全体はこんな感じになります。

import com.sksamuel.elastic4s.ElasticDsl._
import com.sksamuel.elastic4s.http._
import com.sksamuel.elastic4s.playjson._
import com.sksamuel.elastic4s.requests.searches.SearchBodyBuilderFn
import com.sksamuel.elastic4s.{ElasticClient, ElasticProperties, HitReader}
import javax.inject.Inject
import play.api.Logging
import play.api.libs.json.{Json, Reads}

import scala.concurrent.{ExecutionContext, Future}

case class Station(
  stationCode: String,
  stationName: String,
  point: Point,
  prefectureCode: String,
  address: String
)

case class Point(
  lat: Double,
  lon: Double
)

class Sample @Inject() ()(implicit ec: ExecutionContext) extends Logging {
  val client: ElasticClient = ElasticClient(JavaClient(ElasticProperties("http://localhost:9200")))

  def list(stationName: String, offset: Int, limit: Int): Future[(Long, List[Station])] = {
    implicit val reader: Reads[Point] = Json.reads[Point]
    implicit val hitReader: HitReader[Station] = playJsonHitReader(Json.reads[Station])

    val searchRequest =
      search("station")
        .query(
          termQuery("stationName", stationName)
        )
        .from(offset)
        .size(limit)
    logger.debug(s"search body: ${SearchBodyBuilderFn(searchRequest).string}")

    client.execute(searchRequest).map { response: Response[SearchResponse] =>
      response.body.foreach(body => logger.debug(s"response body: $body"))
      (response.result.totalHits, response.result.to[Station].toList)
    }
  }
}

client生成

いくつか方法があるようですが、今回はシンプルなJavaClientを用いて生成します。

val client: ElasticClient = ElasticClient(JavaClient(ElasticProperties("http://localhost:9200")))

実行

client.execute()で実行します。

client.execute(searchRequest)

SearchRequest

今回はSearchを扱うので、ElasticDslにあるsearchを用います。引数にはインデックス名を指定します。

val searchRequest =
  search("station")
    .query(
      termQuery("stationName", condition.stationName)
    )
    .from(condition.offset)
    .size(condition.limit)

以下、生成したンスタンスに対してメソッドチェーンで追加していくイメージです。上記の例では、Term QueryとFrom・Sizeを指定しています。
Term Query以外のQuery、その他の機能も同じような感覚で使っていけます。

ESからのResponse

SearchResponseは、.resultから取得できます。また、レスポンスのJSON全体は.bodyで取得可能です。

client.execute(searchRequest).map { response: Response[SearchResponse] =>
  response.body.foreach(body => logger.debug(s"response body: $body"))
  (response.result.totalHits, response.result.to[Station].toList)
}

playjsonを使って変換する

結果のリストの部分の取得は.result.hits.hitsの内容を変換することも可能ですが、.result.to[T]を使って変換するほうが楽です。当然のことながら、型クラスに指定するclassのフィールドや型は、ESのmappingのに対応していなければなりません。
この機能を利用するには、HitReader[T]を定義する必要があります。playjsonを使うとこのようになります。

implicit val reader: Reads[Point] = Json.reads[Point]
implicit val hitReader: HitReader[Station] = playJsonHitReader(Json.reads[Station])

デバッグ・テストのやり方

以下のようにすることで、リクエスト時に送られるJSONが確認できます。

logger.debug(s"search body: ${SearchBodyBuilderFn(searchRequest).string}")

全体のbodyに限らず、queryの部分の生成に当たるQueryBuilderFn等も提供されており、これらメソッド群はテストコードにも応用できます。

少し注意することとしては、複数の形式の記述をサポートしている場合、生成されるJSONのパターンは1種類になるということです。
Term Queryの例でいうと、以下のクエリはどちらも同じ意味ですが、後者が生成されます。

{
  "term": {
    "stationName": "東京"
  }
}

{
  "term": {
    "stationName": {
      "value": "東京"
    }
  }
}

この他にも、default値が明示的に指定されるなど少し冗長的なJSONになることがあるようです。
どのクエリが生成されるかは、ソースコードを実際に読んでいみるとわかりますので、興味のある方はどうぞ。

まとめ

elastic4sを用いると、Scalaの恩恵を受けながらElasticsearchを扱うことができます。
REST APIで実現できることは大半のことが可能ですので、ぜひ使ってみてださい。

Scala + PlayFramework + Elasticsearch ことはじめ

背景