- 对于大批量数据,查询es时,需要带条件去查询,否则一下查出所有数据数据量会很大
- es查询需要编写json格式的DSL查询语句,对于复杂查询,DSL编写起来也分很复杂,所以我们这里使用sparksql,通过编写sql语句,spark将sql语句自动转化为DSL语句来查询es。
关于es对spark的支持可参见文档,官网 https://www.elastic.co/guide/en/elasticsearch/hadoop/current/spark.html
废话不多说,上代码
case class people(phone:String)
val sparkConf = new SparkConf().setAppName("application-name").setMaster("local[4]")
val sc = new SparkContext(sparkConf)
val sql = new SQLContext(sc)
import sql.implicits._
//配置
val options = Map(
"pushdown" -> "true",
"es.nodes" -> "192.45.15.15", //ip
"es.port" -> "9200") //端口
val esDF = sql.read.format("org.elasticsearch.spark.sql")
.options(options)
.load("spark_202003/alldata") //索引
// esDF.printSchema() //可打印schema信息
val table = esDF.createOrReplaceTempView("spark_202003") //创建临时表用于查询
val sqlString=“select * from spark_202003 where code=‘1’ ”
val resultDF = sql.sql(sqlString)
var resultDS = resultDF.map(x => people(x(0).toString))) //
这里是转换成了dataset,之后可对dataset 做其他操作
打印下查询的内容
resultDS.collect().toList.toString()