Neo4j

Neo4j source connector

Description

The Neo4j source connector reads data from Neo4j by running a Cypher query and mapping the returned fields to a SeaTunnel schema.

neo4j-java-driver version: 4.4.9

Key Features

• batch
• stream
• exactly-once
• column projection
• parallelism
• support user-defined split

Data Type Mapping

Neo4j Value Type	SeaTunnel Data Type
String	STRING
Boolean	BOOLEAN
Integer	INT / BIGINT
Float	FLOAT / DOUBLE
ByteArray	BYTES
Date	DATE
LocalTime	TIME
LocalDateTime	TIMESTAMP
List	ARRAY
Map	MAP
Null	NULL

Source 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 query used to read data. The fields returned by this query must match schema.fields.
schema	Object	Yes	-	SeaTunnel schema of the query result. Configure it under schema.fields.
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.

Notes

• Use exactly one authentication method: username/password, bearer token, or Kerberos ticket.
• query controls which fields are returned. schema.fields must list the returned field names and their SeaTunnel types.
• Returned field names can contain dots, such as t.string, when the Cypher query returns properties from a node.
• MAP fields must use STRING keys, for example MAP<STRING, INT>.
• Neo4j integer and floating-point values are converted according to the SeaTunnel type declared in schema.fields. Use BIGINT/DOUBLE when the value may exceed the range of INT/FLOAT.

Task Example

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

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

    max_transaction_retry_time = 1
    max_connection_timeout = 1

    query = "MATCH (t:Test) WITH *, t{.int} AS _map RETURN t.string, t.boolean, t.long, t.double, t.byteArray, t.date, t.localDateTime, _map, t.list, t.int, t.float"

    schema {
      fields {
        t.string = STRING
        t.boolean = BOOLEAN
        t.long = BIGINT
        t.double = DOUBLE
        t.null = NULL
        t.byteArray = BYTES
        t.date = DATE
        t.localDateTime = TIMESTAMP
        _map = "MAP<STRING, INT>"
        t.list = "ARRAY<INT>"
        t.int = INT
        t.float = FLOAT
      }
    }
  }
}

sink {
  Console {}
}

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