Spark 1.0 + Hadoop 2.0 + Elasticsearch

gistfile1.scala

$ ADD_JARS=../elasticsearch-hadoop-2.1.0.BUILD-SNAPSHOT/dist/elasticsearch-hadoop-2.1.0.BUILD-SNAPSHOT.jar bin/spark-shell

### COMMAND ### 
val tweet = Map("user" -> "kimchy", "post_date" -> "2009-11-15T14:12:12", "message" -> "trying out Elastic Search")
val tweets = sc.makeRDD(Seq(tweet))

import org.apache.hadoop.io.{MapWritable, Text, NullWritable}
def toWritable(in: Map[String, String]) = {
    val m = new MapWritable
    for ((k, v) <- in)
        m.put(new Text(k), new Text(v))
    m
}


val writables = tweets.map(toWritable)
val output = writables.map { v => (NullWritable.get : Object, v : Object) }

import org.elasticsearch.hadoop.mr.EsOutputFormat
sc.hadoopConfiguration.set("es.resource", "tweets/tweet")
output.saveAsNewAPIHadoopFile(
    "/test",
    classOf[NullWritable],
    classOf[MapWritable],
    classOf[EsOutputFormat]
)