title: Elasticsearch-Spark serialization not working with inner classes
body: <p>Elasticsearch/Spark serialization does not appear to play well with nested types.</p>
  
  <p>For example:</p>
  
  <pre><code>public class Foo implements Serializable {
     private List&lt;Bar&gt; bars = new ArrayList&lt;Bar&gt;();
     // getters and setters
  
     public static class Bar implements Serializable {
     }
  }
  
  List&lt;Foo&gt; foos = new ArrayList&lt;Foo&gt;();
  foos.add( new Foo());
  // Note: Foo object does not contain nested Bar instances
  
  SparkConf sc = new SparkConf(); //
  sc.setMaster("local");
  sc.setAppName("spark.app.name");
  sc.set("spark.serializer", KryoSerializer.class.getName()); 
  JavaSparkContext jsc = new JavaSparkContext(sc);
  JavaRDD javaRDD = jsc.parallelize(ImmutableList.copyOf(foos));
  JavaEsSpark.saveToEs(javaRDD, INDEX_NAME+"/"+TYPE_NAME);  
  </code></pre>
  
  <p>The above code above works, and documents of type <code>Foo</code> will be indexed within Elasticsearch.</p>
  
  <p>The issue arises when the <code>bars</code> list in a <code>Foo</code> object is not empty, for instance:</p>
  
  <pre><code>Foo = new Foo();
  Bar = new Foo.Bar();
  foo.getBars().add(bar);
  </code></pre>
  
  <p>Then, when indexing to Elasticsearch, the following exception is thrown:</p>
  
  <pre><code>org.elasticsearch.hadoop.serialization.EsHadoopSerializationException: 
  Cannot handle type [Bar] within type [class Foo], instance [Bar ...]] 
  within instance [Foo@1cf628a] 
  using writer [org.elasticsearch.spark.serialization.ScalaValueWriter@4e635d]
  at org.elasticsearch.hadoop.serialization.builder.ContentBuilder.value(ContentBuilder.java:63)
  at org.elasticsearch.hadoop.serialization.bulk.TemplatedBulk.doWriteObject(TemplatedBulk.java:71)
  at org.elasticsearch.hadoop.serialization.bulk.TemplatedBulk.write(TemplatedBulk.java:58)
  at org.elasticsearch.hadoop.rest.RestRepository.writeToIndex(RestRepository.java:148)
  at org.elasticsearch.spark.rdd.EsRDDWriter.write(EsRDDWriter.scala:47)
  at org.elasticsearch.spark.rdd.EsSpark$$anonfun$saveToEs$1.apply(EsSpark.scala:68)
  at org.elasticsearch.spark.rdd.EsSpark$$anonfun$saveToEs$1.apply(EsSpark.scala:68)
  at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:61)
  at org.apache.spark.scheduler.Task.run(Task.scala:64)
  at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:203)
  at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
  at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
  at java.lang.Thread.run(Unknown Source)
  </code></pre>
  
  <p>These are the relevant Maven dependencies</p>
  
  <pre><code>&lt;dependency&gt;
     &lt;groupId&gt;com.sksamuel.elastic4s&lt;/groupId&gt;
     &lt;artifactId&gt;elastic4s_2.11&lt;/artifactId&gt;
     &lt;version&gt;1.5.5&lt;/version&gt;
  &lt;/dependency&gt;
  
  &lt;dependency&gt;
     &lt;groupId&gt;org.apache.spark&lt;/groupId&gt;
     &lt;artifactId&gt;spark-core_2.11&lt;/artifactId&gt;
     &lt;version&gt;1.3.1&lt;/version&gt;
  &lt;/dependency&gt;
  
  &lt;dependency&gt;
     &lt;groupId&gt;org.elasticsearch&lt;/groupId&gt;
     &lt;artifactId&gt;elasticsearch-hadoop-cascading&lt;/artifactId&gt;
     &lt;version&gt;2.1.0.Beta4&lt;/version&gt;
  &lt;/dependency&gt;
  
  &lt;dependency&gt;
     &lt;groupId&gt;com.fasterxml.jackson.core&lt;/groupId&gt;
     &lt;artifactId&gt;jackson-databind&lt;/artifactId&gt;
     &lt;version&gt;2.1.3&lt;/version&gt;
  &lt;/dependency&gt;
  
  &lt;dependency&gt;
     &lt;groupId&gt;org.elasticsearch&lt;/groupId&gt;
     &lt;artifactId&gt;elasticsearch-spark_2.10&lt;/artifactId&gt;
     &lt;version&gt;2.1.0.Beta4&lt;/version&gt;
  &lt;/dependency&gt;
  
  &lt;dependency&gt;
     &lt;groupId&gt;org.scala-lang&lt;/groupId&gt;
     &lt;artifactId&gt;scala-xml&lt;/artifactId&gt;
     &lt;version&gt;2.11.0-M4&lt;/version&gt;
  &lt;/dependency&gt;
  </code></pre>
  
  <p>What is the correct way to index when using nested types with ElasticSearch and Spark?</p>
  
  <p>Thanks</p>