Options and configuration

Spark options are useful not only to control how the connector reads to and writes from a Neo4j database, but also to manage multiple connections and manage Neo4j driver options.

Multiple connections

The connector allows you to use more than one connection within a single Spark Session. For example, you can read data from a database and write it to another database in the same session.

Reading from a database and writing to a different one
import org.apache.spark.sql.{SaveMode, SparkSession}

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

val df = spark.read.format("org.neo4j.spark.DataSource")
  .option("url", "neo4j://first.host.com:7687")
  .option("labels", "Person")
  .load()

df.write.format("org.neo4j.spark.DataSource")
  .mode(SaveMode.Append)
  .option("url", "neo4j://second.host.com:7687")
  .option("labels", "Person")
  .save()

Another case to use multiple connections is when you want to merge two data sources.

Merge data from two databases
import org.apache.spark.sql.{SaveMode, SparkSession}

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

val dfOne = spark.read.format("org.neo4j.spark.DataSource")
  .option("url", "neo4j://first.host.com:7687")
  .option("labels", "Person")
  .load()

val dfTwo = spark.read.format("org.neo4j.spark.DataSource")
  .option("url", "neo4j://second.host.com:7687")
  .option("labels", "Person")
  .load()

val dfJoin = dfOne.join(dfTwo, dfOne("name") === dfTwo("name"))

Custom authentication supplier

The connector provides an option to use a custom authentication supplier in addition to already supported authentication types, such as none, basic, kerberos, custom and bearer.

The authentication supplier needs to implement interface org.neo4j.connectors.authn.AuthenticationTokenSupplierFactory from the org.neo4j.connectors:commons-authn-spi library. The interface requires implementing two methods:

  • String getName() - returns the name of the supplier. Must be unique and not clash with existing authentication types.

  • Supplier<AuthenticationToken> create(String, String, Map<String, String>) - accepts username, password and additional parameters and return a supplier of AuthenticationToken.

An AuthenticationToken instance is an abstract representation of an arbitrary authentication to be presented to the Neo4j server and can contain a principal and its credentials (username and password), a token or other ways of confirming an identity.

The interface itself offers the following factories:

  • AuthenticationToken#bearer

  • AuthenticationToken#kerberos

  • AuthenticationToken#none

  • AuthenticationToken#usernameAndPassword

  • AuthenticationToken#custom

The typical use-case for an authentication supplier is supporting expiring tokens issued by OAuth 2.0 or OIDC providers.

Custom authentication supplier can be used via the service loader mechanism and by setting configuration parameter authentication.type to the name of the supplier. Additional parameters can be specified as configuration options prefixed with authentication.$name..

Keycloak authentication supplier example

This supplier is based on the Keycloak client, which requires Java 11 or later.

An authentication supplier for Keycloak is available as a ready-to-use, optional artifact: the org.neo4j.connectors:commons-authn-keycloak library. Instead of implementing the supplier yourself, add this library as a dependency to your application.

Check Maven Central for the latest available version.

 
<dependency>
  <groupId>org.neo4j.connectors</groupId>
  <artifactId>commons-authn-keycloak</artifactId>
  <version>{connectors-commons-version}</version>
</dependency>

The library provides:

  • A factory class that creates the authentication token supplier and registers it under the name keycloak through the service loader mechanism.

  • A supplier that obtains an access token from a Keycloak server and refreshes it on expiration.

Once the library is on the classpath, you can configure Spark to use Keycloak authentication as follows:

val df = spark.read
  .format("org.neo4j.spark.DataSource")
  .option("url", s"$NEO4J_URL")
  .option("authentication.type", "keycloak")
  .option("authentication.keycloak.username", s"$KEYCLOAK_USERNAME")
  .option("authentication.keycloak.password", s"$KEYCLOAK_PASSWORD")
  .option("authentication.keycloak.authServerUrl", s"$KEYCLOAK_URL")
  .option("authentication.keycloak.realm", s"$KEYCLOAK_REALM")
  .option("authentication.keycloak.clientId", s"$KEYCLOAK_CLIENT_ID")
  .option("authentication.keycloak.clientSecret", s"$KEYCLOAK_CLIENT_SECRET")
  .option("query", "MATCH (n:Person) WITH n LIMIT 2 RETURN id(n) as id, n.age as age")
  .load()

Neo4j driver options

Under the covers, the Spark connector uses the official Neo4j Java Driver. When using the connector, you can set any valid Neo4j driver option using the option method:

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.type", "basic")
        .option("authentication.basic.username", "myuser")
        .option("authentication.basic.password", "neo4jpassword")
        .option("labels", "Person")
        .load()

Alternatively, you can specify a global configuration in the Spark Session to avoid retyping connection options every time. You can set any Neo4j Connector option adding the neo4j. prefix.

For example, if you want to set the authentication.type option for the session, you have to add neo4j.authentication.type. Here is a full example:

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

val spark = SparkSession.builder()
    .config("neo4j.url", "neo4j://localhost:7687")
    .config("neo4j.authentication.type", "basic")
    .config("neo4j.authentication.basic.username", "myuser")
    .config("neo4j.authentication.basic.password", "neo4jpassword")
    .getOrCreate()

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

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

The following table captures the most common configuration settings to use with the Neo4j driver. For full documentation on all possible configuration options for Neo4j Drivers, see the Neo4j Java Driver manual.

Table 1. List of available configuration settings
Setting name Description Default value Required

Driver options

url

The url of the Neo4j instance to connect to.

When provided with a comma-separated list of URIs, the resolver function feature of the driver will be activated. The first URI will be used as original host while the rest are treated as resolver function outputs.

(none)

Yes

authentication.type

The authentication methods to be used:

See Authentication for more information.

basic

No

authentication.basic.username

Username to use for basic authentication type

(Neo4j Driver default)

No

authentication.basic.password

Username to use for basic authentication type

(Neo4j Driver default)

No

authentication.kerberos.ticket

Kerberos Auth Ticket

(Neo4j Driver default)

No

authentication.custom.principal

This is used to identify who this token represents

(Neo4j Driver default)

No

authentication.custom.credentials

These are the credentials authenticating the principal

(Neo4j Driver default)

No

authentication.custom.realm

This is the "realm" string specifying the authentication provider

(Neo4j Driver default)

No

authentication.bearer.token

This is the token to provide for the bearer authentication scheme

(Neo4j Driver default)

No

encryption.enabled

Specify if encryption should be enabled. This setting is ignored if you use a URI scheme with +s or +ssc

false

No

encryption.trust.strategy

Set certificate trust strategy, it is ignored if the connection URI uses +s or +ssc as suffix. Available values are:

  • TRUST_SYSTEM_CA_SIGNED_CERTIFICATES

  • TRUST_CUSTOM_CA_SIGNED_CERTIFICATES

  • TRUST_ALL_CERTIFICATES.

(Neo4j Driver default)

No

encryption.ca.certificate.path

Set certificate path for TRUST_CUSTOM_CA_SIGNED_CERTIFICATES trust strategy

(Neo4j Driver default)

No

connection.max.lifetime.msecs

Connection lifetime in milliseconds

(Neo4j Driver default)

No

connection.liveness.timeout.msecs

Liveness check timeout in milliseconds

(Neo4j Driver default)

No

connection.acquisition.timeout.msecs

Connection acquisition timeout in milliseconds

(Neo4j Driver default)

No

connection.timeout.msecs

Connection timeout in milliseconds

(Neo4j Driver default)

No

db.transaction.timeout

Transaction timeout in milliseconds

(Neo4j Driver default)

No

db.transaction.metadata

Tag custom transaction metadata to the driver, works on read and write, details below

(empty)

No

Session options

database

Database name to connect to. The driver allows to define the database in the URL, yet in case you set this option, it has the priority compared to the one defined in the URL.

(Neo4j Driver default)

No

access.mode

Possible values are:

  • read

  • write

Used only while you’re pulling data from Neo4j. In case of read, the connector in a cluster environment routes the requests to the followers, otherwise to the leader.

read

No

Driver transaction metadata
Introduced in 6.0

You can use transaction metadata to label your transactions in the target Neo4j database query log, to help identifying them as Cypher® queries generated by the Spark connector.

Metadata is sent as a object/map in the driver, and you populate it using options prefix db.transaction.metadata. Keys are dot-separated and interpreted right after the prefix:

Example
val df = spark.read
    .format("org.neo4j.spark.DataSource")
    .option("query", "RETURN 42 AS example")
    .option("db.transaction.metadata.fromMySparkApp", true)
    .option("db.transaction.metadata.anotherMetadata", "free text goes here")
    .load()
Example
df = (
    spark.read.format("org.neo4j.spark.DataSource")
    .option("query", "RETURN 42 AS example")
    .option("db.transaction.metadata.fromMySparkApp", True)
    .option("db.transaction.metadata.withMessage", "free text goes here")
    .load()
)
Send the object to the driver via transaction metadata.
{
  "fromMySparkApp": true,
  "anotherMetadata": "free text goes here"
}

You can create nested structures by assigning more options. Each dot represents one step deeper into the nested map/object.

Example
val df = spark.read
    .format("org.neo4j.spark.DataSource")
    .option("query", "RETURN 42 AS example")
    .option("db.transaction.metadata.fromMySparkApp", true)
    .option("db.transaction.metadata.example.id", 55)
    .option("db.transaction.metadata.example.message", "another message")
    .load()
Example
df = (
    spark.read.format("org.neo4j.spark.DataSource")
    .option("query", "RETURN 42 AS example")
    .option("db.transaction.metadata.fromMySparkApp", True)
    .option("db.transaction.metadata.example.id", 55)
    .option("db.transaction.metadata.example.message", "another message")
    .load()
)
Send the object to the driver via transaction metadata.
{
  "fromMySparkApp": true,
  "example": {
    "id": 55,
    "message": "another message"
  }
}