******重点中的重点，这是首先要注意的问题：
就是导入的org.elasticsearch.elasticsearch-spark-20_2.11 Jar包的版本一定要和要读取的ES数据库的版本保持一致，
如果比数据库版本低，会直接报错，如果高于数据库的版本，数据的解析会出现问题。
首先配置SparkConf
1 SparkConf conf = new SparkConf() 2 .setAppName("ElasticSearch-spark") 3 .setMaster("local[1]") 4 .set("es.es.index.auto.create", "true") 5 6 .set("es.nodes","127.0.0.1") 7 .set("es.port","9200") 8 .set("es.nodes.wan.only", "true");
第一种读取方式：
1 SparkSession sparkSession = SparkSession.builder().config(conf).getOrCreate(); 2 JavaSparkContext jsc = new JavaSparkContext(sparkSession.sparkContext());//adapter 3 JavaRDD<Map<String, Object>> searchRdd = esRDD(jsc, "index" ).values(); 4 for (Map<String, Object> item : searchRdd.collect()) { 5 item.forEach((key, value)->{ 6 System.out.println("search key:" + key + ", search value:" + value); 7 }); 8 }
第二种读取方式：
1 JavaSparkContext sc = new JavaSparkContext(conf); 2 JavaPairRDD<String, Map<String, Object>> esRDD = JavaEsSpark.esRDD(sc, "index"); 3 System.out.println(esRDD.count()); 4 System.out.println(esRDD.collect().toString()); 5 for(Tuple2 tuple:esRDD.collect()){ 6 System.out.print(tuple._1()+"----------"); 7 System.out.println(tuple._2()); 8 }
第三种读取方式：
1 SparkSession spark = SparkSession.builder().config(conf).getOrCreate(); 2 Dataset<Row> a = spark 3 .read() 4 .format("es") 5 .load("index") 6 ; 7 System.out.println(a.schema()); 8 a.show();
以上三种方式都可以返回ES中的数据，可针对需求自行选择。
记于2019年5月22日20点41分
Spark读取ElasticSearch数据库三种配置方式及其注意事项
原文：https://www.cnblogs.com/langeraa/p/10908418.html