Nicht in der Lage, JSON aus HDFS mit SchemaRDD.saveToES () in Elasticsearch-hadoop zu indexieren - json, scala, elasticsearch, apache-spark, apache-spark-sql

Dies ist mein erster echter Versuch in Funken / Scala, also sei vorsichtig.

Ich habe eine Datei namens test.json auf HDFS, das ich mit Spark zu lesen und zu indizieren versuche. Ich bin in der Lage, die Datei über SQLContext.jsonFile () zu lesen, aber wenn ich versuche, SchemaRDD.saveToEs () zu verwenden, erhalte ich einen ungültigen JSON-Fragment-Fehler. Ich denke, die saveToES () - Funktion formatiert die Ausgabe nicht wirklich in json und sendet stattdessen nur das Wertefeld der RDD.

Was mache ich falsch?

Spark 1.2.0

Elasticsearch-hadoop 2.1.0.BUILD-20150217

test.json:

{"key":"value"}

Funkenschale:

import org.apache.spark.SparkContext._
import org.elasticsearch.spark._

val sqlContext = new org.apache.spark.sql.SQLContext(sc)
import sqlContext._

val input = sqlContext.jsonFile("hdfs://nameservice1/user/mshirley/test.json")
input.saveToEs("mshirley_spark_test/test")

Error:

<snip>
org.elasticsearch.hadoop.rest.EsHadoopInvalidRequest: Found unrecoverable error [Bad Request(400) - Invalid JSON fragment received[["value"]][MapperParsingException[failed to parse]; nested: ElasticsearchParseException[Failed to derive xcontent from (offset=13, length=9): [123, 34, 105, 110, 100, 101, 120, 34, 58, 123, 125, 125, 10, 91, 34, 118, 97, 108, 117, 101, 34, 93, 10]]; ]]; Bailing out..
<snip>

Eingang:

res2: org.apache.spark.sql.SchemaRDD =
SchemaRDD[6] at RDD at SchemaRDD.scala:108
== Query Plan ==
== Physical Plan ==
PhysicalRDD [key#0], MappedRDD[5] at map at JsonRDD.scala:47

input.printSchema ():

root
|-- key: string (nullable = true)

Antworten:

2 für die Antwort № 1

https://github.com/elastic/elasticsearch-hadoop/issues/382

geändert:

import org.elasticsearch.spark._

zu:

import org.elasticsearch.spark.sql._

---