Spark SQL 集成ElasticSearch的案例实战

Spark SQL 集成ElasticSearch的案例实战

ElasticSearch 概念回顾

Spark与ElasticSearch的集成

Java版本

public class SparkESJavaOps {
public static void main(String[] args) {
SparkConf conf = new SparkConf().setAppName(SparkESJavaOps.class.getSimpleName()).setMaster("local");
//在spark中自动创建es中的索引
conf.set("es.index.auto.create","true");
//设置在spark中连接es的url和端口
conf.set("es.nodes","master");
conf.set("es.port","9200");
JavaSparkContext sc = new JavaSparkContext(conf);
JavaRDD<String>  jsonRDD = sc.textFile("E:/test/scala/sqldf.json");
//向es中插入数据
JavaEsSpark.saveJsonToEs(jsonRDD,"spark3/person");
//从es中查询数据
/***
* es索引库的三级目录
* index/type/id
* es
* index -->db
* type -->table
* document -->row
* field --> column
* esJsonRDD 返回的数据JavaPairRDD <String,String> -->t._1就是es中每一个document的id,唯一标识
* t._2就是document数据
* */
JavaEsSpark.esJsonRDD(sc,"spark3/person").foreach(new VoidFunction<Tuple2<String, String>>(){
@Override
public void call(Tuple2<String, String> tuple) throws Exception {
System.out.println(tuple._1()+" ===> " +tuple._2());
}
});

}
}

Scala版本

/**
* Spark 与 ES 的集成
*/
object SparkESOps extends App{
val conf  = new SparkConf().setAppName("SparkESOps").setMaster("local")
//在spark中自动创建es中的索引
conf.set("es.index.auto.creaete","true")
//设置在Spark 中连接 es的url和端口
conf.set("es.nodes","master")
conf.set("es.port","9200")
val sc = new SparkContext(conf)

//向es中写入数据
val linesRDD = sc.textFile("E:/test/scala/sqldf.json")
//这里面参数resources指定的就是es中的索引库和类型格式为：index/type
linesRDD.saveJsonToEs("spark2/person")
//查询数据
val personRDD = sc.esJsonRDD("spark2/person")
personRDD.foreach(println)

sc.stop()

}