跳到主要内容
版本：Next

Neo4j

Neo4j Sink 连接器

描述​

Neo4j Sink 连接器通过执行 Cypher 语句把 SeaTunnel 数据写入 Neo4j。它支持逐条写入， 也支持使用 Cypher UNWIND 批量写入。

neo4j-java-driver 版本：4.4.9

主要特性​

Sink 选项​

名称类型是否必填默认值描述
uriString是-Neo4j 连接地址，例如 neo4j://localhost:7687 或 bolt://localhost:7687。
usernameString否-Neo4j 用户名，需要和 password 一起使用。username、bearer_token、kerberos_ticket 三种认证方式至少配置一种。
passwordString否-Neo4j 密码。配置 username 时必须配置。
bearer_tokenString否-用于 Neo4j 认证的 bearer token。
kerberos_ticketString否-用于 Neo4j 认证的 Kerberos ticket。
databaseString是-Neo4j 数据库名。
queryString是-写入数据使用的 Cypher 语句。ONE_BY_ONE 模式使用 $name 这类占位符；BATCH 模式使用 UNWIND $batch AS row。
queryParamPositionObject是-Cypher 参数名和输入行字段位置的映射。连接器配置校验要求必须填写。
max_batch_sizeInteger否500write_mode = "BATCH" 时，单个事务最多写入的数据条数，必须大于 0。
write_modeString否ONE_BY_ONE写入模式。可选值为 ONE_BY_ONE 和 BATCH。
max_transaction_retry_timeLong否30最大事务重试时间，单位为秒。
max_connection_timeoutLong否30建立 TCP 连接的最大等待时间，单位为秒。
common-optionsconfig否-Sink 通用选项，详见 Sink 通用选项。

注意事项​

  • 认证方式只选一种：用户名密码、bearer token 或 Kerberos ticket。
  • ONE_BY_ONE 模式下，queryParamPosition 用来把 Cypher 占位符映射到输入行的字段位置。
  • BATCH 模式下，查询语句应使用 UNWIND $batch AS row，连接器会通过 batch 变量传入一批数据。
  • BATCH 模式虽然从 row 中取值，但连接器配置校验仍要求填写 queryParamPosition。
  • queryParamPosition 中的字段位置从 0 开始，顺序对应上游输入表结构。
  • BATCH 模式下，每个 row 使用上游字段名取值，因此 Cypher 语句里的字段名需要和上游表结构一致。

逐条写入示例​

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
    }
  }
}

批量写入示例​

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"
  }
}

变更日志​

Change Log
ChangeCommitVersion
[improve] neo4j options (#9164)https://github.com/apache/seatunnel/commit/1eb81e7f882.3.11
[Improve] restruct connector common options (#8634)https://github.com/apache/seatunnel/commit/f3499a6eeb2.3.10
[Feature][Restapi] Allow metrics information to be associated to logical plan nodes (#7786)https://github.com/apache/seatunnel/commit/6b7c53d03c2.3.9
[Feature][Doris] Add Doris type converter (#6354)https://github.com/apache/seatunnel/commit/51899918432.3.6
[Feature][Core] Upgrade flink source translation (#5100)https://github.com/apache/seatunnel/commit/5aabb14a942.3.4
[Improve][Common] Introduce new error define rule (#5793)https://github.com/apache/seatunnel/commit/9d1b2582b22.3.4
[Improve] Remove use SeaTunnelSink::getConsumedType method and mark it as deprecated (#5755)https://github.com/apache/seatunnel/commit/8de74081002.3.4
Support config column/primaryKey/constraintKey in schema (#5564)https://github.com/apache/seatunnel/commit/eac76b4e502.3.4
[Improve] Documentation and partial word optimization. (#4936)https://github.com/apache/seatunnel/commit/6e8de0e2a62.3.3
[Improve][connector-V2-Neo4j]Supports neo4j sink batch write and update docs (#4841)https://github.com/apache/seatunnel/commit/580276a8bd2.3.3
Merge branch 'dev' into merge/cdchttps://github.com/apache/seatunnel/commit/4324ee19122.3.1
[Improve][Project] Code format with spotless plugin.https://github.com/apache/seatunnel/commit/423b5830382.3.1
[improve][api] Refactoring schema parse (#4157)https://github.com/apache/seatunnel/commit/b2f573a13e2.3.1
[Improve][build] Give the maven module a human readable name (#4114)https://github.com/apache/seatunnel/commit/d7cd6010512.3.1
[Improve][Project] Code format with spotless plugin. (#4101)https://github.com/apache/seatunnel/commit/a2ab1665612.3.1
[Feature][Connector] add get source method to all source connector (#3846)https://github.com/apache/seatunnel/commit/417178fb842.3.1
[Feature][API & Connector & Doc] add parallelism and column projection interface (#3829)https://github.com/apache/seatunnel/commit/b9164b8ba12.3.1
[Hotfix][OptionRule] Fix option rule about all connectors (#3592)https://github.com/apache/seatunnel/commit/226dc6a1192.3.0
[Improve][Connector-V2][Neo4j] Unified exception for Neo4j source & sink connector (#3565)https://github.com/apache/seatunnel/commit/58584eefb12.3.0
[Feature][Connector][Neo4j] expose configurable options in Neo4j (#3342)https://github.com/apache/seatunnel/commit/efa04b38fe2.3.0
[Connector-V2][ElasticSearch] Add ElasticSearch Source/Sink Factory (#3325)https://github.com/apache/seatunnel/commit/38254e3f262.3.0
[Feature][Connector-v2] Neo4j source connector (#2777)https://github.com/apache/seatunnel/commit/38b0daf8b72.3.0
[#2606]Dependency management split (#2630)https://github.com/apache/seatunnel/commit/fc047be69b2.2.0-beta
[Feature][Connector-v2] Neo4j sink connector (#2434)https://github.com/apache/seatunnel/commit/950b27d1322.2.0-beta