Not able to use kryoserializer for writing data into Elastic Search.

[ShubhamSisodia0590]

What kind an issue is this?

• Bug report.

Not able to use kryoserializer for writing data into Elastic Search.

When i just set the "spark.serializer" as KryoSerializer.class.getName() in spark conf it is giving me exception

org.elasticsearch.hadoop.serialization.EsHadoopSerializationException: Cannot handle type [class com.spark.apps.pojo.AnalyticsOneDayData] within type [class com.spark.apps.pojo.AnalyticsDailyAggregatedData],

and when i add

"es.ser.writer.value.class" as KryoSerializer.class.getName() i get below exception

org.elasticsearch.hadoop.EsHadoopIllegalStateException: Cannot instantiate class [org.apache.spark.serializer.KryoSerializer]

as the ObjectUtils does clz.newInstance and there is no default constructor for KryoSerializer.

Issue description

Description

I want to insert a nested structure into Elastic Search. The corresponding java classes are

public class DailyAggregatedData implements Serializable {

	private static final long serialVersionUID = -9150958167164783629L;
	String id;
	String cId;
	OneDayData[] aggregatedData;
}

public class OneDayData implements Serializable{

	private static final long serialVersionUID = 8982330246703144883L;
	String date;
	long xcount;
	long ycount;
}

Steps to reproduce

Code:

                SparkConf sparkConf = new SparkConf();
		sparkConf.set(EsSparkConstants.ES_INDEX_AUTO_CREATE, "true");
		sparkConf.set(EsSparkConstants.ES_RESOURCE, "indexName");
		sparkConf.set(EsSparkConstants.ES_NDOES, "localhost");
		sparkConf.set(EsSparkConstants.ES_PORT, "9200");
		sparkConf.set(EsSparkConstants.ES_NODES_WAN_ONLY, "true");
		sparkConf.set("spark.serializer", KryoSerializer.class.getName());

                 SparkConf sparkConf = buildSparkConf();
		JavaSparkContext javaSparkContext = new JavaSparkContext(sparkConf);

                //build JavaRDD this is working fine.
                JavaRDD<DailyAggregatedData> DailyAggregatedDataRDD = 

                JavaEsSpark.saveToEs(javaRDD,
		 "indexName", propertiesMap);

Test/code snippet

Strack trace:

with just "spark.serializer"

org.elasticsearch.hadoop.serialization.EsHadoopSerializationException: Cannot handle type [class com.spark.apps.pojo.AnalyticsOneDayData] within type [class com.spark.apps.pojo.AnalyticsDailyAggregatedData], instance [com.spark.apps.pojo.AnalyticsOneDayData@41c7150a] within instance [com.spark.apps.pojo.AnalyticsDailyAggregatedData@18fb1a1d] using writer [org.spark.serialization.ScalaValueWriter@660cef49]

when i add "es.ser.writer.value.class"

org.elasticsearch.hadoop.EsHadoopIllegalStateException: Cannot instantiate class [org.apache.spark.serializer.KryoSerializer]
	at org.elasticsearch.hadoop.util.ObjectUtils.instantiate(ObjectUtils.java:43)
	at org.elasticsearch.hadoop.util.ObjectUtils.instantiate(ObjectUtils.java:52)
	at org.elasticsearch.hadoop.util.ObjectUtils.instantiate(ObjectUtils.java:48)
	at org.elasticsearch.hadoop.serialization.bulk.AbstractBulkFactory.<init>(AbstractBulkFactory.java:169)
	at org.elasticsearch.hadoop.serialization.bulk.IndexBulkFactory.<init>(IndexBulkFactory.java:27)
	at org.elasticsearch.hadoop.serialization.bulk.BulkCommands.create(BulkCommands.java:40)
	at org.elasticsearch.hadoop.rest.RestRepository.lazyInitWriting(RestRepository.java:127)
	at org.elasticsearch.hadoop.rest.RestRepository.writeToIndex(RestRepository.java:158)
	at org.elasticsearch.spark.rdd.EsRDDWriter.write(EsRDDWriter.scala:67)
	at org.elasticsearch.spark.rdd.EsSpark$$anonfun$doSaveToEs$1.apply(EsSpark.scala:102)
	at org.elasticsearch.spark.rdd.EsSpark$$anonfun$doSaveToEs$1.apply(EsSpark.scala:102)
	at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:70)
	at org.apache.spark.scheduler.Task.run(Task.scala:86)
	at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:274)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
	at java.lang.Thread.run(Thread.java:745)
Caused by: java.lang.InstantiationException: org.apache.spark.serializer.KryoSerializer
	at java.lang.Class.newInstance(Class.java:427)
	at org.elasticsearch.hadoop.util.ObjectUtils.instantiate(ObjectUtils.java:41)
	... 16 more
Caused by: java.lang.NoSuchMethodException: org.apache.spark.serializer.KryoSerializer.<init>()
	at java.lang.Class.getConstructor0(Class.java:3082)
	at java.lang.Class.newInstance(Class.java:412)
	... 17 more

Version Info

OS: : Amazon Linux (Running Spark Cluster on AWS)
JVM : JDK8
Hadoop/Spark:
Hadoop distribution:Amazon 2.7.3
Applications:Spark 2.0.2

ES-Hadoop : 5.3
ES : 5.3

Feature description

[jbaiera]

@ShubhamSisodia0590

I think you might be using es.ser.writer.value.class incorrectly here. That is for implementations of the internal ValueWriter class that takes record types from different integrations (Rows from SparkSQL, Tuples from Pig, Writables from MapReduce, etc) and converts them to JSON data for use in Elasticsearch. The KryoSerializer, on the other hand, is used for sending data between Spark Executors (docs). I don't believe configuring the KryoSerializer will work here, since those two uses are very different from each other.

Does this succeed without the KryoSerializer configured in spark.serializer? I see that your classes implement Serializable, but are not case classes nor are Bean compliant with setters and getters. If you convert them to either Scala Case Classes or Java Bean styled objects does this work?

[ShubhamSisodia0590]

It was not working without KryoSerializer as well and was throwing similar exception as with spark.serializer as kryo.

I have getters and setters in my actual code for all the fields but i haven't pasted that here .

Is this because of array of objects within an object ? Is that supported with default serializer ?

[jbaiera]

It looks like a specific instance of AnalyticsOneDayData is not able to be serialized by the value writer code. Java bean classes are only supported as the top level objects for writing out to at the moment.

I dug into why this is and found that it's actually rather difficult to support collections of Java bean classes. We have value writers for Scala and Java objects, and the Scala version delegates to the Java version if it cannot handle the value. The problem is that the Scala object handles the Java bean class logic, and once a java based collection is handed off to the Java based writer, it never calls back to the Scala writer, even if it has a bean based class.

I'm going to open a new issue for this, since nested bean objects should be fine to use. I think I have a solution for this but need to make sure it doesn't break any other integrations' serialization logic.

[jbaiera]

I opened #1021 to track the root cause of this issue.

[ShubhamSisodia0590]

Is there any work around till then ?

Should I create a JSON object and then convert it to JavaRDD using toString on JSON, and then persist the JSON using saveJsonToEs.

I did a POC and saw that was working as it doesn't perform any serialization.

Is that recommended ? Will that have any performance impact ?

[jbaiera]

Using saveJsonToEs is a pretty safe move - It doesn't perform any serialization logic and is generally considered much faster than using the rich serialization since it doesn't have to perform any parsing. Other alternatives would be to use java Map objects instead of java beans, where the key of the map is the field name and the value is the value to store. Most users of the Spark integration are either using pure Scala (which avoids the split Scala/Java logic) or are using Map objects for their containers.

[jbaiera]

This should be fixed with #1021