Neo4j/Apache Spark task is not serializable

user_13314495 0
Viewed 33

My goal is to load some documents that are loaded as RDD in Apache Spark into a Neo4j Graph. The issue seems to be with the creation of such nodes because Eclipse reports the task not being serialiazable

Exception in thread "main" org.apache.spark.SparkException: Task not serializable

My code is the following:

public class Spark {

    public static void createRdd() {
        
        SparkSession spark = SparkSession.builder()
                .master("local")
                .appName("MongoSparkConnector")
                .config("spark.mongodb.input.uri", "mongodb://127.0.0.1/TheFoodPlanner.join")           
                .getOrCreate();
            
            JavaSparkContext jsc = new JavaSparkContext(spark.sparkContext());
            
            JavaMongoRDD<Document> rddRecipes = MongoSpark.load(jsc);
                    
            Driver driver = GraphDatabase.driver("bolt://localhost:7687", AuthTokens.basic("neo4j", "estrella100"));
            try (Session session = driver.session()) {
                rddRecipes.foreach(f -> {
                    String id = f.getString("id");
                    session.run("CREATE (n: Recipe {id:'" + id + "'})");            
                });             
                session.close();
            }
            
            
            driver.close();
            jsc.close();
        
    }
}

and the error appears in the line session.run("CREATE (n: Recipe {id:'" + id + "'})");

What would be the correct way to load the nodes in paralel?

apache-spark neo4j
1 Answers

The problem is that you're mixing up the Spark-based connector of data from Mongo, with "traditional" Java connector for Neo4j. And it's usual problem in that case that such drivers aren't optimized for use with Spark without additional code that will take care for running code on multiple executors - one common problem here is that the Driver opens connection only on driver node, but not on the executors.

For your case, you have 2 possibilities:

  1. Use existing Spark connector for Neo4j - you'll need to convert data from Mongo into format consumable by Neo4j connector - it could be more optimized that "manual" solution, plus easier to use

  2. Continue to use foreach, but you need to change the approach - initialization of the driver should happen on each executor - this easiest would be done with something like this (pseudo-code, not tested) - this will make that each partition is processed by executor that holding that partition - in this case each executor has its own connection to Neo4j:

rddRecipes.foreachPartition { partitionOfRecords =>
    Driver driver = GraphDatabase.driver("bolt://localhost:7687", 
           AuthTokens.basic("neo4j", "estrella100"));
    try (Session session = driver.session()) {
        partitionOfRecords.foreach(f -> {
          String id = f.getString("id");
          session.run("CREATE (n: Recipe {id:'" + id + "'})");            
          });             
        session.close();
    }
}
Alex_Ott_186270
Related