idea maven 依赖
<dependency>
<groupId>org.elasticsearch</groupId>
<artifactId>elasticsearch-hadoop</artifactId>
<version>7.11.1</version>
</dependency>
<dependency>
<groupId>org.elasticsearch.client</groupId>
<artifactId>elasticsearch-rest-high-level-client</artifactId>
<version>7.1.1</version>
</dependency>
import org.apache.hbase.thirdparty.io.netty.util.concurrent.Future
import org.apache.http.HttpResponse
import org.apache.spark.sql.SparkSession
import org.elasticsearch.spark.rdd.EsSpark
import java.io.{BufferedWriter, FileWriter}
object ReadEs {
def main(args: Array[String]): Unit = {
val spark = SparkSession
.builder()
//.master("local[1]") //本地跑打开
.appName("read es")
.config("pushdown", "true")
.config("es.nodes", "host1,host2,host3")
.config("es.port", "9200")
.config("es.index.auto.create", "true")
.config("es.index.read.missing.as.empty", "yes")
.config("es.mapping.date.rich", "false")
.getOrCreate()
val query =
s"""
{
"query": {
"match_all": {}
}
}
|""".stripMargin
val value = EsSpark.esRDD(spark.sparkContext, "index*", query)
value.map(it => {
val x= it._2("x").toString
val t= it._2("t").toString.toInt
(x, t)
})
.reduceByKey(_+_)
.foreachPartition(mp=>{
//写本地文件
val file = new BufferedWriter(new FileWriter("data/1.txt", true))
mp.foreach(it => {
val t= it._2
val x= it._1
file.append(s"$x ${t}\n")
})
file.flush()
file.close()
})
spark.stop()
}
}