spark怎么结合es使用

1、启动es后，spark shell导入es-hadoop jar包：cp elasticsearch-hadoop-2.1.2/dist/elasticsearch-spark* spark-1.6.0-bin-hadoop2.6/lib/

cd spark-1.6.0-bin-hadoop2.6/bin

./spark-shell --jars ../lib/elasticsearch-spark-1.2_2.10-2.1.2.jar

2、互交结果如下:import org.apache.spark.SparkConf

import org.elasticsearch.spark._

val conf = new SparkConf()

conf.set("es.index.auto.create", "true")

conf.set("es.nodes", "127.0.0.1")

val numbers = Map("one" -> 1, "two" -> 2, "three" -> 3)

val airports = Map("OTP" -> "Otopeni", "SFO" -> "San Fran")

sc.makeRDD(Seq(numbers, airports)).saveToEs("spark/docs")

3、然后查看ES中的数据:http://127.0.0.1:9200/spark/docs/_search?q=*

4、结果就是这样啦:{"took":71,"timed_out":false,"_shards":{"total":5,"successful":5,"failed":0},"hits":{"total":2,"max_score":1.0,"hits":[{"_index":"spark","_type":"docs","_id":"AVfhVqPBv9dlWdV2DcbH","_score":1.0,"_source":{"OTP":"Otopeni","SFO":"San Fran"}},{"_index":"spark","_type":"docs","_id":"AVfhVqPOv9dlWdV2DcbI","_score":1.0,"_source":{"one":1,"two":2,"three":3}}]}}