Spark optimizations

Filters

The Neo4j Connector for Apache Spark implements the SupportPushdownFilters interface, that allows you to push the Spark filters down to the Neo4j layer. In this way the data that Spark receives have been already filtered by Neo4j, decreasing the amount of data transferred from Neo4j to Spark.

You can manually disable the PushdownFilters support using the pushdown.filters.enabled option and set it to false (default is true).

If you use the filter function more than once, like in this example:

import org.apache.spark.sql.{SaveMode, SparkSession}

val spark = SparkSession.builder().getOrCreate()

val df = (spark.read.format("org.neo4j.spark.DataSource")
  .option("url", "neo4j://localhost:7687")
  .option("authentication.basic.username", "neo4j")
  .option("authentication.basic.password", "letmein!")
  .option("labels", ":Person")
  .load())

df.where("name = 'John Doe'").where("age = 32").show()

The conditions are automatically joined with an AND operator.

When using relationship.node.map = true or query the PushdownFilters support is automatically disabled. In that case, the filters are applied by Spark and not by Neo4j.

Aggregation

The Neo4j Connector for Apache Spark implements the SupportsPushDownAggregates interface, that allows you to push Spark aggregations down to the Neo4j layer. In this way the data that Spark receives have been already aggregate by Neo4j, decreasing the amount of data transferred from Neo4j to Spark.

You can manually disable the PushdownAggregate support using the pushdown.aggregate.enabled option and set it to false (default is true).

// Given a DB populated with the following query
"""
  CREATE (pe:Person {id: 1, fullName: 'Jane Doe'})
  WITH pe
  UNWIND range(1, 10) as id
  CREATE (pr:Product {id: id * rand(), name: 'Product ' + id, price: id})
  CREATE (pe)-[:BOUGHT{when: rand(), quantity: rand() * 1000}]->(pr)
  RETURN *
"""
import org.apache.spark.sql.{SaveMode, SparkSession}
val spark = SparkSession.builder().getOrCreate()

(spark.read.format("org.neo4j.spark.DataSource")
  .option("url", "neo4j://localhost:7687")
  .option("authentication.basic.username", "neo4j")
  .option("authentication.basic.password", "letmein!")
  .option("relationship", "BOUGHT")
  .option("relationship.source.labels", "Person")
  .option("relationship.target.labels", "Product")
  .load
  .createTempView("BOUGHT"))


val df = spark.sql(
  """SELECT `source.fullName`, MAX(`target.price`) AS max, MIN(`target.price`) AS min
    |FROM BOUGHT
    |GROUP BY `source.fullName`""".stripMargin)

df.show()

The MAX and MIN operators are applied directly on Neo4j.

Push-down limit

The Neo4j Connector for Apache Spark implements the SupportsPushDownLimit interface. That allows you to push Spark limits down to the Neo4j layer. In this way the data that Spark receives have been already limited by Neo4j. This decreases the amount of data transferred from Neo4j to Spark.

You can manually disable the PushdownLimit support using the pushdown.limit.enabled option and set it to false (default is true).

// Given a DB populated with the following query
"""
  CREATE (pe:Person {id: 1, fullName: 'Jane Doe'})
  WITH pe
  UNWIND range(1, 10) as id
  CREATE (pr:Product {id: id * rand(), name: 'Product ' + id, price: id})
  CREATE (pe)-[:BOUGHT{when: rand(), quantity: rand() * 1000}]->(pr)
  RETURN *
"""
import org.apache.spark.sql.{SaveMode, SparkSession}
val spark = SparkSession.builder().getOrCreate()

val df = (spark.read
      .format("org.neo4j.spark.DataSource")
      .option("url", "neo4j://localhost:7687")
      .option("authentication.basic.username", "neo4j")
      .option("authentication.basic.password", "letmein!")
      .option("relationship", "BOUGHT")
      .option("relationship.source.labels", "Person")
      .option("relationship.target.labels", "Product")
      .load
      .select("`target.name`", "`target.id`")
      .limit(10))


df.show()

The limit value will be pushed down to Neo4j.

Push-down top N

The Neo4j Connector for Apache Spark implements the SupportsPushDownTopN interface. That allows you to push top N aggregations down to the Neo4j layer. In this way the data that Spark receives have been already aggregated and limited by Neo4j. This decreases the amount of data transferred from Neo4j to Spark.

You can manually disable the PushDownTopN support using the pushdown.topN.enabled option and set it to false (default is true).

// Given a DB populated with the following query
"""
  CREATE (pe:Person {id: 1, fullName: 'Jane Doe'})
  WITH pe
  UNWIND range(1, 10) as id
  CREATE (pr:Product {id: id * rand(), name: 'Product ' + id, price: id})
  CREATE (pe)-[:BOUGHT{when: rand(), quantity: rand() * 1000}]->(pr)
  RETURN *
"""
import org.apache.spark.sql.{SaveMode, SparkSession}
val spark = SparkSession.builder().getOrCreate()

val df = (spark.read
      .format("org.neo4j.spark.DataSource")
      .option("url", "neo4j://localhost:7687")
      .option("authentication.basic.username", "neo4j")
      .option("authentication.basic.password", "letmein!")
      .option("relationship", "BOUGHT")
      .option("relationship.source.labels", "Person")
      .option("relationship.target.labels", "Product")
      .load
      .select("`target.name`", "`target.id`")
      .sort(col("`target.name`").desc)
      .limit(10))


df.show()

The limit value will be pushed down to Neo4j.

Partitioning

When the connector fetches data from Neo4j during read operations you can partition the query. The partitions run in parallel as separate Spark tasks.

Consider the following job:

import org.apache.spark.sql.SaveMode
import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder().getOrCreate()

val df = (spark.read.format("org.neo4j.spark.DataSource")
        .option("url", "neo4j://localhost:7687")
        .option("authentication.basic.username", "neo4j")
        .option("authentication.basic.password", "letmein!")
        .option("labels", "Person")
        .option("partitions", "5")
        .load())

This means that if the total count of the nodes with label Person into Neo4j is 100 we are creating 5 partitions and each one manages 20 records (we use SKIP / LIMIT queries).

Partitioning the dataset makes sense when you are working with a large dataset, for example where you have more than 10 million records.

How to parallelize query execution

Partitions option is supported by three of our modes. GDS is not supported by this option. You can therefore partition these modes:

  1. Node extraction.

  2. Relationship extraction.

  3. Query extraction.

A dynamic count on what you are trying to fetch is provided and a query with SKIP / LIMIT approach over each partition is built. For a dataset of 100 nodes (Person) with a partition size of 5 the following queries are generated (one for each partition):

MATCH (p:Person) RETURN p SKIP 0 LIMIT 20
MATCH (p:Person) RETURN p SKIP 20 LIMIT 20
MATCH (p:Person) RETURN p SKIP 40 LIMIT 20
MATCH (p:Person) RETURN p SKIP 60 LIMIT 20
MATCH (p:Person) RETURN p SKIP 80 LIMIT 20

For node and relationship extractions, the connector automatically leverages the Neo4j count store in order to retrieve the total count of the nodes and/or relationships you are trying to fetch. However, for custom query extractions, you need to provide an additional optimized count query, or a static number if you already know the count.

The count query is executed once and should be considered as additional overhead for when partitions is enabled. It is just as important to optimize your count query as it is to optimize the actual query. By providing a custom count query, you gain the flexibility to optimize it against your domain.

The connector will opt for one of three possible approaches to determine a global partition count. By setting option query.count you influence the approach that is used:

  1. Compute the global count using a second optimized query that leverages indexes.

    • Do this by setting .option("query.count", "<your cypher count query>").

  2. Provide a static integer value as count if you know the value.

    • Do this by setting option query.count to an integer, e.g. .option("query.count", 100)

  3. If you do not set a value for query.count a value is automatically generated.

    • Generated on the form of CALL { <your query> } RETURN count(*) AS count for you.

A count query specified by query.count option must return a single integer as count, consider this example:

MATCH (p:Person)-[r:BOUGHT]->(pr:Product)
WHERE pr.name = 'An Awesome Product'
RETURN count(p) AS count

Failing to provide such a query or count will fall back to approach (3) which has no guarantee to be an optimized query. It is therefore strongly recommended to use indexes and custom count queries in favor of falling back to (3).

Deterministic ordering of partitions

To guarantee that there is no data overlap between the partitions, you must provide a deterministic ordering in your query. To do this, use ORDER BY in your query to provide explicit ordering. If you do not provide an explicit order, it can lead to incomplete extractions, where some data is never included in any of the partitions.

This is a documented Cypher® behavior, please see SKIP in the Cypher manual for more details.

Reading partitions when under heavy write-load

The partition optimization feature assumes that it is safe to run each individual partition of work in their own driver transactions. This is not the case if your instance is under heavy write-load. When your database is under heavy write-load it sees frequent state changes.

The count query runs in one transaction before creating the partitions, which then all run under individual transactions. Any data written after partitions are determined and before actual reading operation manage to start execution may not be read. To mitigate this risk, be mindful of your database state before a read operation.