search（5）- elastic4s-构建索引

  按照计划，这篇开始尝试用elastic4s来做一系列索引管理和搜索操作示范。前面提过，elastic4s的主要功能之一是通过组合Dsl语句形成json请求。那么我们先试试组合一些Dsl语句，再想办法产生出json请求文本，然后在kibana控制台中验证它的正确性。

首先看看elastic4s提供的一个show函数：

def show(implicit handler: Handler[T, _]): String = Show[ElasticRequest].show(handler.build(t))

又见到了这个Handler[T, _]，作为show的一个隐式参数。前面说过这个Handler[T, _]是个代表构建T类型json请求的typeclass。具体构建函数就是上面的这个build(t)函数。我们先看看CreateIndexRequest类型的show函数示范：

val jsCreate = createIndex("company")
 .shards(1).replicas(1).show
 println(jsCreate)

产生了json如下：

PUT:/company?
StringEntity({"settings":{"index":{"number_of_shards":1,"number_of_replicas":1}}},Some(application/json))

在kibana里是如下表达的：

PUT /company
{
 "settings":{
 "index":{
 "number_of_shards":1,
 "number_of_replicas":1
 }
 }
}

可能是历史原因吧，elastic4s与ES7.6还有很多不兼容的地方，或者说是elastic4s还有许多没来得及更新的地方。具体有问题的语句或参数都可以通过把json body放在kibana里进行验证，如果elastic4s还有地方没有完成覆盖ES7.6功能的话，我们可以把一个正确的ES7.6 json脚本直接通过source传人到操作类型中去：

val js =
 """
 |{
 | "settings":{
 | "index":{
 | "number_of_shards":1,
 | "number_of_replicas":1
 | }
 | }
 |}
 |""".stripMargin
 val createFromJs = (createIndex("company").source(js)).show
 println(createFromJs)

下面的例子是一套完整的索引创建过程：先删除同名称索引、创建索引、构建mapping：

import com.sksamuel.elastic4s.ElasticClient
import com.sksamuel.elastic4s.akka._
import akka.actor._
import com.sksamuel.elastic4s.requests.mappings.{KeywordField, MappingDefinition, NestedField, SearchAsYouTypeField}
import scala.concurrent.ExecutionContext.Implicits.global
object Lesson03 extends App {
 import com.sksamuel.elastic4s.ElasticDsl._
 //akka客户端
 private implicit lazy val system: ActorSystem = ActorSystem()
 val akkaClient = AkkaHttpClient(AkkaHttpClientSettings(List("localhost:9200")))
 val client = ElasticClient(akkaClient)
 //删除索引
 val idxDelete = client.execute(deleteIndex("company")).await
 //构建索引
 val idxCreate = client.execute(createIndex("company")
 .shards(1).replicas(1)).await
 //创建表结构
 if(idxCreate.isSuccess) {
 val compMapping = client.execute(
 putMapping("company").fields(
 KeywordField("code"),
 SearchAsYouTypeField("name")
 .fielddata(true)
 .fields(KeywordField("keyword")),
 textField("biztype"),
 NestedField("addr").fields(
 textField("district"),
 textField("address"),
 KeywordField("zipcode")
 ),
 dateField("regdate")
 .ignoreMalformed(true)
 .format("strict_date_optional_time||epoch_millis"),
 textField("contact")
 )).await
 if(compMapping.isSuccess)
 println(s"mapping successfully created.")
 else
 println(s"mapping creation error: ${compMapping.error.reason}")
 } else {
 println(s"index creation error: ${idxCreate.error.reason}")
 }
 system.terminate()
 client.close()
}

以上代码有几个地方值得提一下：

1、这上面使用了一个基于akka-stream的客户端。优点是响应式标准兼容，用队列queue来缓冲密集请求

2、在删除索引前为甚么不先检查一下同名索引是否存在？elastic4s ExistApi还是ES7以前版本，不能用

3、client.execute(...)返回Future, 为什么不用for-yield？试过了，一是deleteIndex,createIndex返回结果与实际删除、构建操作可能有些延迟，createIndex会返回索引已经存在错误, mapping会出现索引不存在错误。