Neo4j

Neo4j sink connector

Description

The Neo4j sink connector writes SeaTunnel rows to Neo4j by running a Cypher statement. It supports one-row-at-a-time writes and batch writes with Cypher UNWIND.

neo4j-java-driver version: 4.4.9

Key Features

• exactly-once

Sink Options

Name	Type	Required	Default	Description
uri	String	Yes	-	Neo4j connection URI, for example neo4j://localhost:7687 or bolt://localhost:7687.
username	String	No	-	Neo4j username. Use it together with password. One of username, bearer_token, or kerberos_ticket must be configured.
password	String	No	-	Neo4j password. Required when username is configured.
bearer_token	String	No	-	Bearer token used for Neo4j authentication.
kerberos_ticket	String	No	-	Kerberos ticket used for Neo4j authentication.
database	String	Yes	-	Neo4j database name.
query	String	Yes	-	Cypher statement used to write data. In ONE_BY_ONE mode, use placeholders such as $name; in BATCH mode, use UNWIND $batch AS row.
queryParamPosition	Object	Yes	-	Mapping between Cypher parameter names and input row field positions. It is required by the connector configuration.
max_batch_size	Integer	No	500	Maximum number of rows written in one transaction when write_mode = "BATCH". Must be greater than 0.
write_mode	String	No	ONE_BY_ONE	Write mode. Supported values are ONE_BY_ONE and BATCH.
max_transaction_retry_time	Long	No	30	Maximum transaction retry time, in seconds.
max_connection_timeout	Long	No	30	Maximum time to wait for a TCP connection to be established, in seconds.
common-options	config	No	-	Sink common options. See Sink Common Options.

Notes

• Use exactly one authentication method: username/password, bearer token, or Kerberos ticket.
• In ONE_BY_ONE mode, queryParamPosition maps each Cypher placeholder to a field position in the input row.
• In BATCH mode, the query should use UNWIND $batch AS row. The connector passes the rows through the batch variable.
• queryParamPosition is still required by the connector configuration in BATCH mode, even though the batch query reads values from row.
• Field positions in queryParamPosition start from 0, following the input schema field order.
• In BATCH mode, each row entry uses the input field names, so the names referenced in the Cypher statement must match the upstream schema.

Write One Row At A Time

sink {
  Neo4j {
    uri = "neo4j://localhost:7687"
    username = "neo4j"
    password = "password"
    database = "neo4j"

    max_transaction_retry_time = 10
    max_connection_timeout = 10

    query = "CREATE (a:Person {name: $name, age: $age})"
    queryParamPosition = {
      name = 0
      age = 1
    }
  }
}

Write In Batches

env {
  parallelism = 1
  job.mode = "BATCH"
}

source {
  FakeSource {
    plugin_output = "fake"
    parallelism = 1
    row.num = 1000
    schema = {
      fields {
        name = "string"
        age = "int"
      }
    }
  }
}

sink {
  Neo4j {
    uri = "neo4j://localhost:7687"
    username = "neo4j"
    password = "password"
    database = "neo4j"

    write_mode = "BATCH"
    max_batch_size = 500
    max_transaction_retry_time = 3
    max_connection_timeout = 10

    queryParamPosition = {
      name = 0
      age = 1
    }

    query = "UNWIND $batch AS row CREATE (n:BatchLabel) SET n.name = row.name, n.age = row.age"
  }
}

Changelog

Change Log

Change	Commit	Version
[improve] neo4j options (#9164)	https://github.com/apache/seatunnel/commit/1eb81e7f88	2.3.11
[Improve] restruct connector common options (#8634)	https://github.com/apache/seatunnel/commit/f3499a6eeb	2.3.10
[Feature][Restapi] Allow metrics information to be associated to logical plan nodes (#7786)	https://github.com/apache/seatunnel/commit/6b7c53d03c	2.3.9
[Feature][Doris] Add Doris type converter (#6354)	https://github.com/apache/seatunnel/commit/5189991843	2.3.6
[Feature][Core] Upgrade flink source translation (#5100)	https://github.com/apache/seatunnel/commit/5aabb14a94	2.3.4
[Improve][Common] Introduce new error define rule (#5793)	https://github.com/apache/seatunnel/commit/9d1b2582b2	2.3.4
[Improve] Remove use SeaTunnelSink::getConsumedType method and mark it as deprecated (#5755)	https://github.com/apache/seatunnel/commit/8de7408100	2.3.4
Support config column/primaryKey/constraintKey in schema (#5564)	https://github.com/apache/seatunnel/commit/eac76b4e50	2.3.4
[Improve] Documentation and partial word optimization. (#4936)	https://github.com/apache/seatunnel/commit/6e8de0e2a6	2.3.3
[Improve][connector-V2-Neo4j]Supports neo4j sink batch write and update docs (#4841)	https://github.com/apache/seatunnel/commit/580276a8bd	2.3.3
Merge branch 'dev' into merge/cdc	https://github.com/apache/seatunnel/commit/4324ee1912	2.3.1
[Improve][Project] Code format with spotless plugin.	https://github.com/apache/seatunnel/commit/423b583038	2.3.1
[improve][api] Refactoring schema parse (#4157)	https://github.com/apache/seatunnel/commit/b2f573a13e	2.3.1
[Improve][build] Give the maven module a human readable name (#4114)	https://github.com/apache/seatunnel/commit/d7cd601051	2.3.1
[Improve][Project] Code format with spotless plugin. (#4101)	https://github.com/apache/seatunnel/commit/a2ab166561	2.3.1
[Feature][Connector] add get source method to all source connector (#3846)	https://github.com/apache/seatunnel/commit/417178fb84	2.3.1
[Feature][API & Connector & Doc] add parallelism and column projection interface (#3829)	https://github.com/apache/seatunnel/commit/b9164b8ba1	2.3.1
[Hotfix][OptionRule] Fix option rule about all connectors (#3592)	https://github.com/apache/seatunnel/commit/226dc6a119	2.3.0
[Improve][Connector-V2][Neo4j] Unified exception for Neo4j source & sink connector (#3565)	https://github.com/apache/seatunnel/commit/58584eefb1	2.3.0
[Feature][Connector][Neo4j] expose configurable options in Neo4j (#3342)	https://github.com/apache/seatunnel/commit/efa04b38fe	2.3.0
[Connector-V2][ElasticSearch] Add ElasticSearch Source/Sink Factory (#3325)	https://github.com/apache/seatunnel/commit/38254e3f26	2.3.0
[Feature][Connector-v2] Neo4j source connector (#2777)	https://github.com/apache/seatunnel/commit/38b0daf8b7	2.3.0
[#2606]Dependency management split (#2630)	https://github.com/apache/seatunnel/commit/fc047be69b	2.2.0-beta
[Feature][Connector-v2] Neo4j sink connector (#2434)	https://github.com/apache/seatunnel/commit/950b27d132	2.2.0-beta