java写法 spark读取es数据

你可以使用以下代码使用Java中的Spark读取Elasticsearch数据：

import org.apache.spark.SparkConf;
import org.apache.spark.SparkContext;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.SQLContext;
import org.elasticsearch.spark.rdd.api.java.JavaEsSpark;

public class ReadFromES {
    public static void main(String[] args) {
        SparkConf sparkConf = new SparkConf()
                .setAppName("ReadFromES")
                .setMaster("local[*]")
                .set("es.nodes", "localhost")
                .set("es.port", "9200")
                .set("es.index.auto.create", "true");

        JavaSparkContext sparkContext = new JavaSparkContext(sparkConf);
        SQLContext sqlContext = new SQLContext(sparkContext);

        JavaRDD<Map<String, Object>> esRDD = JavaEsSpark.esRDD(sparkContext, "index_name/type_name");

        DataFrame dataFrame = sqlContext.read().json(esRDD.map(JSONObject::toJSONString));
        dataFrame.show();
    }
}

这个代码使用了 Elasticsearch Spark插件来读取Elasticsearch数据。在这个例子中，它会读取`index_name`和`type_name`的数据，并将其转换为DataFrame，然后以表格形式输出。你需要将`es.nodes`和`es.port`参数设置为你的Elasticsearch的IP地址和端口号。