Spark讀取ElasticSearch數據庫三種配置方式及其注意事項

******重點中的重點，這是首先要注意的問題：

就是導入的org.elasticsearch.elasticsearch-spark-20_2.11  Jar包的版本一定要和要讀取的ES數據庫的版本保持一致，

如果比數據庫版本低，會直接報錯，如果高於數據庫的版本，數據的解析會出現問題。

首先配置SparkConf

        SparkConf conf = new SparkConf()
                .setAppName("ElasticSearch-spark")
                .setMaster("local[1]")
                .set("es.es.index.auto.create", "true")

                .set("es.nodes","127.0.0.1")
                .set("es.port","9200")
                .set("es.nodes.wan.only", "true");

 

第一種讀取方式：

        SparkSession sparkSession = SparkSession.builder().config(conf).getOrCreate();
        JavaSparkContext jsc = new JavaSparkContext(sparkSession.sparkContext());//adapter
        JavaRDD<Map<String, Object>> searchRdd = esRDD(jsc, "index" ).values();
        for (Map<String, Object> item : searchRdd.collect()) {
            item.forEach((key, value)->{
                System.out.println("search key:" + key + ", search value:" + value);
            });
        }

 

第二種讀取方式：

        JavaSparkContext sc = new JavaSparkContext(conf);
        JavaPairRDD<String, Map<String, Object>> esRDD = JavaEsSpark.esRDD(sc, "index");
        System.out.println(esRDD.count());
        System.out.println(esRDD.collect().toString());
        for(Tuple2 tuple:esRDD.collect()){
            System.out.print(tuple._1()+"----------");
            System.out.println(tuple._2());
        }

第三種讀取方式：

        SparkSession spark = SparkSession.builder().config(conf).getOrCreate();
        Dataset<Row> a  = spark
                .read()
                .format("es")
                .load("index")
                ;
        System.out.println(a.schema());
        a.show();

以上三種方式都可以返回ES中的數據，可針對需求自行選擇。

記於2019年5月22日20點41分