elasticsearch常见CRUD操作与利用spark进行es数据导入实例

最近项目中用到elasticsearch，一款基于Lucene的优秀的搜索引擎，为方便快速入手操作，在此总结了常见的基于curl的crud操作以及利用 spark 将数据导入到es的实例：

#检索文档

curl -XGET

#创建Indix

curl -XPUT &＃8216;mbdt1:9200/customer?pretty&＃8217;

#列出Indix

curl &＃8216;mbdt2:9200/_cat/indices?v&＃8217;

#更新document，doc指定的是文档内容

curl -XPOST &＃8216;localhost:9200/customer/external/1/_update?pretty&＃8217; -d &＃8216;

{

&＃8220;doc&＃8221;: { &＃8220;name&＃8221;: &＃8220;Jane Doe&＃8221; }

}&＃8217;

#执行动态脚本

curl -XPOST &＃8216;localhost:9200/customer/external/1/_update?pretty&＃8217; -d &＃8216;

{

&＃8221; script &＃8221; : &＃8220;ctx._source.age += 5&＃8221;

}&＃8217;

#&＃8221;scripts of type [ inline ], operation [update] and lang [groovy] are disabled&＃8221;

#有可能出现上面的错误， Elasticsearch开启groovy动态语言支持，相关链接：

### 完全开启

#编辑&＃8220;`config/elasticsearch.yml&＃8220;`文件,在最后添加以下代码

script.inline: on

script.indexed: on

script.file: on

### 沙盒中开启

#编辑&＃8220;`config/elasticsearch.yml&＃8220;`文件,在最后添加以下代码

script.inline: sandbox

script.indexed: sandbox

script.file: on

#删除文档

curl -XDELETE &＃8216;localhost:9200/customer/external/2?pretty&＃8217;

##批量操作

#创建

curl -XPOST &＃8216;localhost:9200/customer/external/_bulk?pretty&＃8217; -d &＃8216;

{&＃8220;index&＃8221;:{&＃8220;_id&＃8221;:&＃8221;1&＃8243;}}

{&＃8220;name&＃8221;: &＃8220;John Doe&＃8221; }

{&＃8220;index&＃8221;:{&＃8220;_id&＃8221;:&＃8221;2&＃8243;}}

{&＃8220;name&＃8221;: &＃8220;Jane Doe&＃8221; }

&＃8216;

#用create来创建

curl -XPOST &＃8216;mbdt1:9200/_bulk?pretty&＃8217; -d &＃8216;

{ &＃8220;create&＃8221;: { &＃8220;_index&＃8221;: &＃8220;index1&＃8221;, &＃8220;_type&＃8221;: &＃8220;resource&＃8221;, &＃8220;_id&＃8221;: 13 } }

{ &＃8220;title&＃8221;: &＃8220;&＃8230;..&＃8221; }

&＃8216;

&＃8216;

#多操作

curl -XPOST &＃8216;localhost:9200/customer/external/_bulk?pretty&＃8217; -d &＃8216;

{&＃8220;update&＃8221;:{&＃8220;_id&＃8221;:&＃8221;1&＃8243;}}

{&＃8220;doc&＃8221;: { &＃8220;name&＃8221;: &＃8220;John Doe becomes Jane Doe&＃8221; } }

{&＃8220;delete&＃8221;:{&＃8220;_id&＃8221;:&＃8221;2&＃8243;}}

&＃8216;

#本地文件批量导入

curl -XPOST &＃8216;localhost:9200/bank/account/_bulk?pretty&＃8217; &＃8211;data-binary &＃8220;@accounts.json&＃8221;

curl &＃8216;localhost:9200/_cat/indices?v&＃8217;

#查询，全匹配

curl -XPOST &＃8216;localhost:9200/bank/_search?pretty&＃8217; -d &＃8216;

{

&＃8220;query&＃8221;: { &＃8220;match_all&＃8221;: {} }

}&＃8217;

#排序

curl -XPOST &＃8216;mbdt1:9200/bank/_search?pretty&＃8217; -d &＃8216;

{

&＃8220;query&＃8221;: { &＃8220;match_all&＃8221;: {} },

&＃8220;_source&＃8221;: [&＃8220;account_number&＃8221;, &＃8220;balance&＃8221;],

&＃8220;sort&＃8221;: {&＃8220;balance&＃8221;: {&＃8220;order&＃8221;:&＃8221;desc&＃8221;}},

&＃8220;from&＃8221;: 10,

&＃8220;size&＃8221;:3

}&＃8217;

#字段匹配

curl -XGET mbdt2:9200/z_wo_order/record/_search?pretty -d &＃8216;{

&＃8220;query&＃8221;: {

&＃8220;match&＃8221;: {

&＃8220;cust_name&＃8221;: {

&＃8220;query&＃8221;: &＃8220;黄利欢&＃8221;,

&＃8220;operator&＃8221;: &＃8220;and&＃8221;

}

}

}

}&＃8217;

#组合查询

GET /my_index/my_type/_search

{

&＃8220;query&＃8221;: {

&＃8220;bool&＃8221;: {

&＃8220;must&＃8221;: { &＃8220;match&＃8221;: { &＃8220;title&＃8221;: &＃8220;quick&＃8221; }},

&＃8220;must_not&＃8221;: { &＃8220;match&＃8221;: { &＃8220;title&＃8221;: &＃8220;lazy&＃8221; }},

&＃8220;should&＃8221;: [

{ &＃8220;match&＃8221;: { &＃8220;title&＃8221;: &＃8220;brown&＃8221; }},

{ &＃8220;match&＃8221;: { &＃8220;title&＃8221;: &＃8220;dog&＃8221; }}

]

}

}

}

#控制精度

GET /my_index/my_type/_search

{

&＃8220;query&＃8221;: {

&＃8220;match&＃8221;: {

&＃8220;title&＃8221;: {

&＃8220;query&＃8221;: &＃8220;quick brown dog&＃8221;,

&＃8220;minimum_should_match&＃8221;: &＃8220;75%&＃8221;

}

}

}

}

利用spark导入数据到es：

#调用elasticsearch-hadoop中的spark模块导入数据

命令：

./spark-shell &＃8211;jars /home/ouguangneng/elasticsearch-hadoop-2.2.0/dist/elasticsearch-spark_2.10-2.2.0.jar &＃8211;conf spark.es.nodes=xxx2 &＃8211;conf spark.es.port=9200 &＃8211;master yarn-client &＃8211;num-executors 10 &＃8211;driver-memory 4g &＃8211;executor-memory 3g &＃8211;executor-cores 4

#基于 rdd 导入数据

import org.elasticsearch.spark._

val numbers = Map(&＃8220;one&＃8221; -> 1, &＃8220;two&＃8221; -> 2, &＃8220;three&＃8221; -> 3)

val airports = Map(&＃8220;arrival&＃8221; -> &＃8220;Otopeni&＃8221;, &＃8220;SFO&＃8221; -> &＃8220;San Fran&＃8221;)

sc.makeRDD(Seq(numbers, airports)).saveToEs(&＃8220;test/ext&＃8221;)

#基于RDD导入数据，并指定document id

import org.elasticsearch.spark._

import org.elasticsearch.spark.rdd.EsSpark

case class Trip( oid : String, departure: String, arrival: String)

val upcomingTrip = Trip(&＃8220;1&＃8221;, &＃8220;OTP&＃8221;, &＃8220;SFO&＃8221;)

val lastWeekTrip = Trip(&＃8220;2&＃8221;, &＃8220;MUC&＃8221;, &＃8220;OTP&＃8221;)

val rdd = sc.makeRDD(Seq(upcomingTrip, lastWeekTrip))

EsSpark.saveToEs(rdd, &＃8220;test/ext&＃8221;, Map(&＃8220;es.mapping.id&＃8221; -> &＃8220;oid&＃8221;))

#基于spark sql的dataframe导入数据

import org.elasticsearch.spark.sql._

import org.apache.spark.sql.hive.HiveContext

val hiveCOntext= new HiveContext(sc)

val df = sql(&＃8220;select * from tmp.z_wo_order limit 50&＃8221;)

df.saveToEs(&＃8220;z_wo_order/record&＃8221;, Map(&＃8220;es.mapping.id&＃8221; -> &＃8220;order_id&＃8221;))