如果我們把ES作為某種資料庫來使用的話,必須熟練掌握ES的CRUD操作。在這之前先更正一下上篇中關於檢查索引是否存在的方法:elastic4s的具體調用如下: //刪除索引 val rspExists = client.execute(indexExists("company")).await if ...
如果我們把ES作為某種資料庫來使用的話,必須熟練掌握ES的CRUD操作。在這之前先更正一下上篇中關於檢查索引是否存在的方法:elastic4s的具體調用如下:
//刪除索引
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異常:
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" -> "[email protected]"
).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" -> "[email protected]"
).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進行一次驗證,輸出顯示:
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來檢查記錄是否已經存在:
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請求里的欄位值當作新記錄內容插入:
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() }
成批更新比較麻煩,因為通常每條記錄的更新都可能涉及到當前記錄的欄位值,或作為判斷條件,或為更新值,我們需要使用並處理當前記錄中某些欄位。這就需要在數據層面運行某些計算方法,可以用腳本語言來實現這樣的功能,如下:
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 示例如下:
(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