apache-spark - elasticsearch-hadoop spark连接器无法使用开箱即用的ES服务器设置和默认库设置进行连接/写入

Question

使用用于Spark的Elasticsearch连接器时遇到一些问题，如下所述:https://www.elastic.co/guide/en/elasticsearch/hadoop/master/spark.html。我什至无法在其页面上使用与我使用的Elasticsearch 7.4.0的普通 Vanilla 实例一起工作的示例
通过下载并开始

<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]

请注意，如果我在浏览器URL栏中键入 http://127.0.0.1:9200/，则会返回一个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")
}

Answer 1

所以，问题是我在另一个监听相同端口的窗口中有另一个Elasticsearch实例。总是以奇怪的方式来处理事情。所以..这个适配器根本没有问题。问题是我。