Load data with Apache Spark

For large-scale ingestion jobs, using Apache Spark with the Neo4j Connector can be the right solution.

Use the suggestions below to improve ingestion performance.

Perform preliminary data work in Spark

When loading large amounts of data into Neo4j, use Spark as much as possible for data-heavy work (ETL, quality fixes and checks, dropping missing records, changing data types).

For example, a typical use case is to create a graph from a Spark DataFrame structured like the following:

name,surname,customer_id,product,product_id,quantity,order
John,Doe,C1,Product 1,P1,200,ABC100
Jane,Doe,C2,Product 2,P2,100,ABC200

In this case, you will get the best performance by creating three separate datasets (representing source nodes, target nodes, and relationships respectively) and writing each of them to Neo4j separately:

Python example
customers = spark.sql("SELECT name, surname, customer_id FROM purchases")
products = spark.sql("SELECT product, product_id FROM purchases")
bought = spark.sql("SELECT product_id, customer_id, quantity, order FROM purchases")

# Write each DataFrame to Neo4j

Tune Spark and Neo4j together

  • Turn on (default) or off any Spark optimizations (filters, aggregations, push-down, partitioning) depending on your use case. In general, when Spark optimizations are turned on, the amount of data transferred from Neo4j to Spark is decreased.

  • Size your Neo4j instance appropriately before using aggressive parallelism or large batch sizes in Spark. The instance must have enough free memory for the heap and the page cache. See Operations manual → Performance → Memory configuration for a few examples of capacity planning.

  • New in 6.0 Tune the queries themselves (specify the Cypher® version as CYPHER 25 or CYPHER 5, enable parallel runtimes for reads, tune any existing Cypher preamble parameters) See Cypher query tuning for more information.

Use the correct indexes and constraints

Since the Spark connector may generate many MERGE queries, make sure to create indexes or constraints on any graph property that you use as part of node.keys, relationship.source.node.keys, relationship.target.node.keys or other similar key options.

Due to the concurrency of Spark jobs, when using the Overwrite mode you should use the property uniqueness constraint to guarantee node uniqueness.

Performance tips for reading data

If a schema is not specified, the Spark Connector uses sampling. Since this is a potentially expensive operation, consider defining a schema explicitly instead.

Reading with the script option

The main purpose of the script option is to ensure that constraints related to its associated read query are actually in place. It is strongly recommended to not inject data with the script option for reads, as doing so results in inefficient queries.

The recommended process is to create a dedicated DataFrame containing the new data, then join it with another DataFrame performing the actual reads against Neo4j. Spark supports multiple pushdown optimizations including filter pushdown, which means the resulting read query will be efficient.

import org.apache.spark.sql.functions.col

val filterDf = spark.read.format("org.neo4j.spark.DataSource")
  .option("labels", "Filter")
  .load()

val commentDf = spark.read.format("org.neo4j.spark.DataSource")
  .option("labels", "Comment")
  .load()

// Pushdown optimizations filter data at the source (Neo4j) to minimize data transfer
val resultDf = commentDf.join(filterDf, col("filterId") === filterDf("id"))
// Suppose there are many nodes that match,
// resulting in a large list of values
val script = "MATCH (f:Filter) RETURN collect(f.id) AS ids"

// Sending a large list to the server would add a lot of memory pressure
val query = """
  MATCH (n:Comment)
  WHERE n.filterId IN scriptResult[0].ids
  RETURN n.id AS id
"""
spark.read.format("org.neo4j.spark.DataSource")
  .option("script", script)
  .option("query", query)
  .load()
  .show()

Performance tips for writing data

Since writing is typically an expensive operation, make sure you write only the DataFrame columns you need.

For example, if the columns from the data source are name, surname, age, and livesIn, but you only need name and surname, run the following code (the example is in Scala, but it applies to Python as well):

df.select(df("name"), df("surname"))
  .write
  .format("org.neo4j.spark.DataSource")
  .mode(SaveMode.Append)
  .option("labels", ":Person")
  .save()

To increase write performance:

  • Increase the batch size with the batch.size option. The larger the batch, the fewer transactions are executed to write all of your data, and the less transactional overhead is incurred.

  • Ensure that your Neo4j instance has ample free heap and properly sized page cache. Small heaps make you unable to commit large batches, which in turn slows overall import.