es 7.4.2+spark2.0 访问es

pom.xml

 <dependency>
            <groupId>org.elasticsearch</groupId>
            <artifactId>elasticsearch-spark-20_2.11</artifactId>
            <version>7.4.2</version>
        </dependency>

 

代码

private static Map<String, String> getEsConf() {
        Map<String, String> esOptions = new HashMap<String, String>();
        esOptions.put("es.nodes", "ip");
        esOptions.put("es.port", "9200");
        esOptions.put("pushdown", "true");
        esOptions.put ("es.nodes.wan.only", "true");
        esOptions.put("es.mapping.date.rich", "false");

         esOptions.put("es.index.auto.create", "true");
        return esOptions;
    }

    private static Dataset registerEsTable(SparkSession spark, String indexAndType) {

        Dataset ds = spark.read().format("es").options(getEsConf()).load(indexAndType);
        return ds;
    }

main:

SparkSession spark = SparkSession.builder()
                .master("local[*]")
                .config("spark.sql.warehouse.dir", "/data/sparkwh")
                .config("spark.executor.memory","5g")

                .appName("xxd").getOrCreate();
        Dataset ds = registerEsTable(spark, "xxindex");