Serialization problems using Function implementations with Spark

Question

I'm having troubles understanding Spark Function implementations in Java. The documentation gives three ways to use functions in `map` and `reduce` : - via lambda - via inline classes implementing `Function` and `Function2` - via inner classes implementing `Function` and `Function2` The trouble is that I can't manage to make `2.` and `3.` work. For instance, this code : ``` public int countInline(String path) { String master = "local"; SparkConf conf = new SparkConf().setAppName("charCounterInLine") .setMaster(master); JavaSparkContext sc = new JavaSparkContext(conf); JavaRDD lines = sc.textFile(path); JavaRDD lineLengths = lines .map(new Function() { public Integer call(String s) { return s.length(); } }); return lineLengths.reduce(new Function2() { public Integer call(Integer a, Integer b) { return a + b; } }); // the line causing the error } ``` gives me this mistake : ``` 14/07/09 11:23:20 INFO DAGScheduler: Failed to ru…

Accepted Answer

The relevant part of this stracktrace is:

```
Task not serializable: java.io.NotSerializableException: Hadoop.Spark.basique.CharCounter
```

When you defined your functions as inner classes, their enclosing object is being pulled into the function closure and serialized. If this class is non-serializable or contains a non-serializable field, then you'll run into this error.

You have a few options here:

- Mark the non-serializable fields of the enclosing object as `transient`.

- Define your functions as outer-classes.

- Define your functions as static nested classes.

Serialization problems using Function implementations with Spark

Solution

Problem