我在使用elasticsearchSpark连接器时遇到了一些问题，如下所述：https://www.elastic.co/guide/en/elasticsearch/hadoop/master/spark.html. 我甚至无法在他们的页面上找到使用ElasticSearch7.4.0的普通示例的例子，而我是通过下载和启动ElasticSearch7.4.0的

<downloadDir>/bin/elasticsearch

这是我为跑步所做的。我通过命令启动了spark：

spark-shell --packages "org.elasticsearch:elasticsearch-hadoop:7.4.0"

然后，我输入了上述文档页面上给出的代码行：

import org.apache.spark.SparkContext        // attempt 1
import org.apache.spark.SparkContext._
import org.elasticsearch.spark._

val numbers = Map("one" -> 1, "two" -> 2, "three" -> 3)
val airports = Map("arrival" -> "Otopeni", "SFO" -> "San Fran")

spark.sparkContext.makeRDD( Seq(numbers, airports)).saveToEs("spark/docs")

我收到一些奇怪的错误，表明es连接到的不是默认的主节点[127.0.0.1:9200]，然后即使使用该节点也会失败：

[Stage 0:>                                                        (0 + 12) / 12]20/10/13 19:39:21 ERROR NetworkClient: Node [172.20.0.3:9200] failed (org.apache.commons.httpclient.ConnectTimeoutException: The host did not accept the connection within timeout of 60000 ms); selected next node [127.0.0.1:9200]
20/10/13 19:39:21 ERROR NetworkClient: Node [172.20.0.3:9200] failed (org.apache.commons.httpclient.ConnectTimeoutException: The host did not accept the connection within timeout of 60000 ms); selected next node [127.0.0.1:9200]

注意如果我输入http://127.0.0.1：9200/在我的浏览器url栏中，我得到一个json文档，指示集群已启动localhost:9200. 所以，我被难住了！非常感谢您的指导。

更新

我尝试了mikalai建议的答案（必须通过rdd调用savetoes，而不是dataframe，因为它由于某种原因无法编译）。不幸的是，我得到了同样的结果。

import org.apache.spark.rdd.RDD   // attempt 2
import org.apache.spark.sql.SparkSession
import org.elasticsearch.spark._

object classes {
  case class AlbumIndex(group: String, year: Int,  title: String)

}
object App extends App {
  import classes._
  val spark = SparkSession .builder() .appName("writetoes") .master("local[*]") .config("spark.es.nodes","localhost").config("spark.es.port","9200").getOrCreate()
  val indexDocuments: Seq[AlbumIndex] = Seq(
      AlbumIndex("Led Zeppelin",1969,"Led Zeppelin"),
      AlbumIndex("Boston",1976,"Boston"),
      AlbumIndex("Fleetwood Mac", 1979,"Tusk")
  )
  val rdd: RDD[AlbumIndex] = spark.sparkContext.makeRDD( indexDocuments)
  rdd.saveToEs("demoindex/albumindex")
}