scala - 给定 Elasticsearch 无效模式

Question

我在下面的代码:

def main(args: Array[String]) {
val sparkConf = new SparkConf().setAppName("Spark-Hbase").setMaster("local[2]")
      .set("es.index.auto.create", "true")
      .set("es.resource", "test")
      .set("es.nodes", "127.0.0.1")
      .set("es.output.json", "true")

/* More code */

DStream.map {
      _._2
}.foreachRDD { (rdd: RDD[String]) =>
    EsSpark.saveJsonToEs(rdd, "127.0.0.1")
}

我不断收到 elastic search es.nodes属性的错误消息:

Caused by: org.elasticsearch.hadoop.EsHadoopIllegalArgumentException: invalid pattern given 127.0.0.1/
    at org.elasticsearch.hadoop.util.Assert.isTrue(Assert.java:50)
    at org.elasticsearch.hadoop.serialization.field.AbstractIndexExtractor.compile(AbstractIndexExtractor.java:51)
    at org.elasticsearch.hadoop.rest.RestService.createWriter(RestService.java:398)
    at org.elasticsearch.spark.rdd.EsRDDWriter.write(EsRDDWriter.scala:40)
    at org.elasticsearch.spark.rdd.EsSpark$$anonfun$saveToEs$1.apply(EsSpark.scala:67)
    at org.elasticsearch.spark.rdd.EsSpark$$anonfun$saveToEs$1.apply(EsSpark.scala:67)
    at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66)
    at org.apache.spark.scheduler.Task.run(Task.scala:89)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
    ... 3 more

我通过放置 127.0.0.1做错什么了吗？我尝试通过执行 127.0.0.1:9200来放置端口，但仍然无法正常工作。有人有指针吗？谢谢。

Answer 1

没关系，我想通了。此方法saveJsonToEs:

EsSpark.saveJsonToEs(rdd, "127.0.0.1")

不使用 flex 搜索集群的IP地址，而是将 flex 搜索中的索引保存到 String rdd。

所以应该是这样的:

EsSpark.saveJsonToEs(rdd, "test/sampleApp") /* Where `test` is the `index` and `sampleApp` is the `type` */ 

第二个参数必须为 Requires the format <index>/<type>类型。索引和类型可能对您的特定应用程序有意义。请注意，先前 flex 搜索中不必存在该索引。您可以在 Spark Conf对象上使用该属性: set("es.index.auto.create", "true")以自动创建它。如我上面的属性所示。

供引用: https://www.elastic.co/guide/en/elasticsearch/hadoop/current/spark.html

在Command + f上方的链接中，搜索 Writing existing JSON to Elasticsearch。