spark结合es的代码

好的，下面是一个使用Spark结合Elasticsearch的Scala代码示例。假设您已经安装了Elasticsearch和Spark，并将Elasticsearch Hadoop库添加到Spark的classpath中：

import org.apache.spark.sql.SparkSession

object SparkESExample {
  def main(args: Array[String]) {

    // 创建一个SparkSession
    val spark = SparkSession.builder()
      .appName("Spark Elasticsearch Example")
      .config("es.nodes", "localhost") // 设置Elasticsearch的节点
      .config("es.port", "9200") // 设置Elasticsearch的端口
      .config("es.index.auto.create", "true") // 如果索引不存在，则创建索引
      .master("local[*]")
      .getOrCreate()

    // 读取Elasticsearch中的数据到Spark DataFrame
    val df = spark.read.format("org.elasticsearch.spark.sql")
      .option("es.query", """{"match_all": {}}""") // 设置Elasticsearch的查询
      .load("index/type")

    // 对数据进行处理和分析
    val result = df.groupBy("field").count().orderBy($"count".desc)

    // 将结果保存回Elasticsearch
    result.write.format("org.elasticsearch.spark.sql")
      .option("es.resource", "output_index/output_type") // 设置输出的索引和类型
      .option("es.mapping.id", "field") // 设置文档ID
      .mode("overwrite")
      .save()

    // 关闭SparkSession
    spark.stop()
  }
}

这个示例读取Elasticsearch中的所有文档，对其中的"field"字段进行分组和计数，并将结果保存回Elasticsearch的"output_index"索引和"output_type"类型中。您可以根据自己的需求修改查询、处理和输出的逻辑。