elasticsearch-hadoop 擴展定製 官方包以支持 update upsert doc

官方源碼地址 https://github.com/elastic/elasticsearch-hadoop

相關文檔 https://www.elastic.co/guide/en/elasticsearch/hadoop/current/configuration.html

spark to es

4種操做方式
index
update
upsert
create

只支持四種操做，看文檔描述，目前的需求只能用 upsert 實現，但官方的包對 upsert 支持不完整

upsert 現支持
/*
* {
* "script":{
* "inline": "...",
* "lang": "...",
* "params": ...,
* },
* "upsert": {...}
* }
*/

及
/*
* {
* "doc_as_upsert": true,
* "doc": {...}
* }
*/

並不支持
/*
* {
* "upsert": {},
* "doc": {...}
* }
*/

 

無奈只好自已動手了

代碼量太大，找到相關的部分，理順操做邏輯後，改起來就容易多了，最主要的部分在這裏

source code

把rdd json 化並拼接http 請求

兩種思路

1傳入完整的對像，而後修改 json 解析部分，拼接出請求
2修改對象結構，json 解析部分不變，拼接出請求

1的實現相對複雜，工做量很大，且要求對代碼項目很熟悉，實現的成本很高
2的實現很簡單

修改的地方不多,能夠參照着自已改,需從新編譯

具體看 commit，也提交到了官方，但代碼比較粗暴，極可能通不過，功能優先，官方不採用，能夠用的時候再我的修改。

 

 

以後打包，引用打包後的文件。

另外，程序也要做並要的修改

寫入部分， 對比下就知道要改的地方，很容易

 

case class ES_Upsert(kw_index: String, kw_type: String, id: String, date_idate: String, date_udate: String)

case class ES_Doc(date_udate: String)

case class ES_UpsertDoc(upsert: ES_Upsert, doc: ES_Doc)

.saveToEs(Map[String, String](
"es.resource" -> "{upsert.kw_index}/{upsert.kw_type}",
"es.nodes" -> es,
"es.input.json" -> "false",
"es.nodes.discovery" -> "false",
"es.update.doc" -> "true",
"es.nodes.wan.only" -> "true",
"es.write.operation" -> "upsert",
"es.mapping.exclude" -> "upsert.kw_index,upsert.kw_type,upsert.id",
"es.mapping.id" -> "upsert.id"
))

 

外套一層對象ES_UpsertDoc 字段名稱分別爲upsert，doc熟悉es的就不用解釋吧
"es.update.doc" -> "true"爲 true 才生效。

"es.resource" -> "{upsert.kw_index}/{upsert.kw_type}",
"es.mapping.exclude" -> "upsert.kw_index,upsert.kw_type,upsert.id"
index type field mapping 也要多套一層

項目示例

kafka spark streaming elasticsearch

https://github.com/cclient/elasticsearch-spark-upsert-from-kafka

 

——官方已經拒掉了，主要緣由是這個包要在各類數據平臺上保證可用，按官方的說法是

'whether using Map/Reduce or libraries built upon it such as Hive, Pig or Cascading or new upcoming libraries like Apache Spark'

如今的case只是基於 Spark的，即便在spark上可用，沒有在其餘平臺的測試，不會經過，也沒有精力去挨個試，等用的時候自已改吧