search(6)- elastic4s-CRUD

    若是咱們把ES做爲某種數據庫來使用的話,必須熟練掌握ES的CRUD操做。在這以前先更正一下上篇中關於檢查索引是否存在的方法:elastic4s的具體調用以下:java

  //刪除索引
  val rspExists = client.execute(indexExists("company")).await
  if (rspExists.result.exists) client.execute(deleteIndex("company")).await

在下面咱們還會示範如何檢查一條記錄(document)是否存在的方法。數據庫

先示範新建一條記錄。通常來說數據庫表都有個惟一字段,最好用ES裏的id來表明,不然ES會自動產生一個惟一id,那麼隨機讀取get時就會很不方便。若是新插入的記錄id已經在表裏存在,ES會替換新的內容,不會產生異常。能夠在elastic4s裏使用createOnly(true)來強制產生重複id異常:less

import com.sksamuel.elastic4s.{ElasticClient, ElasticProperties} import com.sksamuel.elastic4s.akka._ import akka.actor._ import scala.concurrent.ExecutionContext.Implicits.global
object Lesson05 extends App { import com.sksamuel.elastic4s.ElasticDsl._ private implicit lazy val system: ActorSystem = ActorSystem() val akkaClient = AkkaHttpClient(AkkaHttpClientSettings(List("130.1.1.234:9200"))) val client = ElasticClient(akkaClient) val publisher = indexInto("company").id("c00001") .fields( "code" -> "c00001", "name" -> "人民出版社", "biztype" -> "出版社", "addr" -> Map( "district" -> "北京市東城區", "address" -> "朝陽門內大街166號" ), "regdate" -> "1963-02-18", "contact" -> "65122634@163.com" ).createOnly(true) val pubExists = client.execute(exists("company","c00001")).await
  if (pubExists.isSuccess) { val createPub = client.execute(publisher).await

    if (createPub.isSuccess) { val pub = client.execute(get("company", "c00001").fetchSourceContext(true)).await println(s"${pub.result.sourceAsMap}") } else println(s"${createPub.error.reason}") } else println(s"${pubExists.error.reason}") val dstributor = indexInto("company").id("c00002") .fields( "code" -> "c00002", "name" -> "新華文軒出版傳媒股份有限公司", "biztype" -> "出版發行", "addr" -> Map( "district" -> "四川省成都市錦江區", "address" -> "金石路239號4棟1層1號" ), "regdate" -> "2005-03-09", "contact" -> "52635286@qq.com" ).createOnly(true) val grpExists = client.execute(exists("company","c00002")).await
  if (grpExists.isSuccess) { val createGroup = client.execute(dstributor).await

    if (createGroup.isSuccess) { val dstr = client.execute(get("company", "c00002").fetchSourceContext(true)).await println(s"${dstr.result.sourceAsMap}") } else println(s"${createGroup.error.reason}") } else println(s"${grpExists.error.reason}") val mget = client.execute(multiget( get("company","c00001"), get("company","c00002") )).await
  if(mget.isSuccess) mget.result.items.foreach(i => println(s"${i.sourceAsMap}")) else println(s"${mget.error.reason}") scala.io.StdIn.readLine() system.terminate() client.close() }

上面示範了不一樣類型字段的填寫方式,特別是nested字段如addr。每插入一條新記錄就用get進行一次驗證,輸出顯示:fetch

HashMap(name -> 人民出版社, regdate -> 1963-02-18, contact -> 65122634@163.com, code -> c00001, addr -> Map(district -> 北京市東城區, address -> 朝陽門內大街166號), biztype -> 出版社) HashMap(name -> 新華文軒出版傳媒股份有限公司, regdate -> 2005-03-09, contact -> 52635286@qq.com, code -> c00002, addr -> Map(district -> 四川省成都市錦江區, address -> 金石路239號4棟1層1號), biztype -> 出版發行) HashMap(name -> 人民出版社, regdate -> 1963-02-18, contact -> 65122634@163.com, code -> c00001, addr -> Map(district -> 北京市東城區, address -> 朝陽門內大街166號), biztype -> 出版社) HashMap(name -> 新華文軒出版傳媒股份有限公司, regdate -> 2005-03-09, contact -> 52635286@qq.com, code -> c00002, addr -> Map(district -> 四川省成都市錦江區, address -> 金石路239號4棟1層1號), biztype -> 出版發行)

上面提到過,若是咱們想把ES看成普通的數據庫來使用的話,仍是要備齊了CRUD功能。具體操做按照關係數據庫方式圍繞着惟一鍵id進行。好比,咱們能夠用id來檢查記錄是否已經存在:spa

  val pubExists = client.execute(exists("company","c00001")).await
  if (pubExists.isSuccess) {...}

上面咱們示範了針對索引的create,read操做。下面討論一下update:update 可分單筆或批次兩類,分別爲:updateById, updateByQuery,很明顯:updateByQuery是以query做爲目標篩選條件的成批update操做。與上面的create操做同樣,咱們仍是須要考慮惟一鍵id,這個能夠在updateById操做裏處理:當目標id存在時,用update請求裏的字段值更新對應的字段。如目標id不存在的話就把update請求裏的字段值看成新記錄內容插入:scala

import com.sksamuel.elastic4s.http.JavaClient
import com.sksamuel.elastic4s.requests.common.RefreshPolicy
import com.sksamuel.elastic4s.{ElasticClient, ElasticProperties}
import scala.concurrent.ExecutionContext.Implicits.global
object Lesson06 extends App {
  import com.sksamuel.elastic4s.ElasticDsl._

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

  val doc1 = updateById("company","t00001")
      .docAsUpsert(
        Map(
          "code" -> "t00001",
          "name" -> "test company1"
        )
      )

  val doc2 = updateById("company","t00002")
    .docAsUpsert(
      Map(
        "code" -> "t00002",
        "name" -> "test company2"
      )
    )

  val doc3 = updateById("company","t00003")
    .docAsUpsert(
      Map(
        "code" -> "t00003",
        "name" -> "test company3"
      )
    )

  val updateAll = for {
    _ <- client.execute(doc1)
    _ <- client.execute(doc2)
    _ <- client.execute(doc3)
  } yield()
  updateAll.await

  val getResults = client.execute(multiget(
     get("company","t00001").fetchSourceInclude("code","name"),
     get("company","t00002").fetchSourceInclude("code","name"),
     get("company","t00003").fetchSourceInclude("code","name")
   )
  ).await

  getResults.result.items.foreach(i => println(i.sourceAsMap))

  client.close()
}

成批更新比較麻煩,由於一般每條記錄的更新均可能涉及到當前記錄的字段值,或做爲判斷條件,或爲更新值,咱們須要使用並處理當前記錄中某些字段。這就須要在數據層面運行某些計算方法,能夠用腳本語言來實現這樣的功能,以下:code

 import com.sksamuel.elastic4s.requests.script.Script val script = "ctx._source.fullname = ctx._source.code+' '+ctx._source.name" val updateByQ = updateIn("company") .query(matchQuery("name","test")) .script(Script(script,Some("painless"))) val qupResult = client.execute(updateByQ).await val getResults = client.execute(multiget( get("company","t00001"), get("company","t00002"), get("company","t00003") ) ).await getResults.result.items.foreach(i => println(i.sourceAsMap))

與update同樣,delete也分單個或成批刪除模式。delete by Id 示例以下:blog

  (for { _ <- client.execute(delete("t00001").from("company")) _ <- client.execute(deleteByQuery("company", "t00002")) } yield()).await

delete by Query 用法以下:索引

 import com.sksamuel.elastic4s.Index._ client.execute( deleteByQuery(toIndex("company"), termQuery("code","t00003")) ).await
相關文章
相關標籤/搜索