No se puede indexar JSON desde HDFS utilizando SchemaRDD.saveToES () en Elasticsearch-hadoop - json, scala, elasticsearch, apache-spark, apache-spark-sql

Este es mi primer intento real de chispa / escala así que sé gentil.

Tengo un archivo llamado prueba.json en HDFS que estoy tratando de leer e indexar usando Spark. Puedo leer el archivo a través de SQLContext.jsonFile () pero cuando trato de usar SchemaRDD.saveToEs () recibo un fragmento de JSON no válido recibido error. Estoy pensando que la función saveToES () no está formateando realmente la salida en json y en su lugar solo está enviando el campo de valor del RDD.

¿Qué estoy haciendo mal?

Spark 1.2.0

Elasticsearch-hadoop 2.1.0.BUILD-20150217

prueba.json:

{"key":"value"}

chispa:

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>

entrada:

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)

Respuestas

2 para la respuesta № 1

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

cambiado:

import org.elasticsearch.spark._

a:

import org.elasticsearch.spark.sql._

---