This is an automated email from the ASF dual-hosted git repository.
zhangshenghang pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 04d28d011e [Docs][Connector-V2] Improve Neo4j connector docs (#11261)
04d28d011e is described below
commit 04d28d011e0b5d06de61e30b81c30c31f60e17cc
Author: Daniel Carter <[email protected]>
AuthorDate: Fri Jul 3 08:14:16 2026 +0800
[Docs][Connector-V2] Improve Neo4j connector docs (#11261)
Co-authored-by: DanielCarter-stack
<[email protected]>
Co-authored-by: Shenghang <[email protected]>
---
docs/en/connectors/sink/Neo4j.md | 165 ++++++++++++++---------------------
docs/en/connectors/source/Neo4j.md | 157 +++++++++++++++++-----------------
docs/zh/connectors/sink/Neo4j.md | 170 ++++++++++++++++---------------------
docs/zh/connectors/source/Neo4j.md | 160 +++++++++++++++++-----------------
4 files changed, 296 insertions(+), 356 deletions(-)
diff --git a/docs/en/connectors/sink/Neo4j.md b/docs/en/connectors/sink/Neo4j.md
index 58811c91cb..bfe288a1a0 100644
--- a/docs/en/connectors/sink/Neo4j.md
+++ b/docs/en/connectors/sink/Neo4j.md
@@ -6,102 +6,48 @@ import ChangeLog from '../changelog/connector-neo4j.md';
## Description
-Write data to Neo4j.
+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
+`neo4j-java-driver` version: 4.4.9
-## Key features
+## Key Features
- [ ] [exactly-once](../../introduction/concepts/connector-v2-features.md)
-## Options
-
-| name | type | required | default value |
-|----------------------------|---------|----------|---------------|
-| uri | String | Yes | - |
-| username | String | No | - |
-| password | String | No | - |
-| max_batch_size | Integer | No | - |
-| write_mode | String | No | OneByOne |
-| bearer_token | String | No | - |
-| kerberos_ticket | String | No | - |
-| database | String | Yes | - |
-| query | String | Yes | - |
-| queryParamPosition | Object | Yes | - |
-| max_transaction_retry_time | Long | No | 30 |
-| max_connection_timeout | Long | No | 30 |
-| common-options | config | no | - |
-
-### uri [string]
-
-The URI of the Neo4j database. Refer to a case: `neo4j://localhost:7687`
-
-### username [string]
-
-username of the Neo4j
-
-### password [string]
-
-password of the Neo4j. required if `username` is provided
-
-### max_batch_size [Integer]
-
-max_batch_size refers to the maximum number of data entries that can be
written in a single transaction when writing to a database.
-
-### write_mode
-
-The default value is oneByOne, or set it to "Batch" if you want to have the
ability to write in batches
-
-```cypher
-unwind $ttt as row create (n:Label) set n.name = row.name,n.age = rw.age
-```
-
-"ttt" represents a batch of data.,"ttt" can be any arbitrary string as long as
it matches the configured "batch_data_variable".
-
-### bearer_token [string]
-
-base64 encoded bearer token of the Neo4j. for Auth.
-
-### kerberos_ticket [string]
-
-base64 encoded kerberos ticket of the Neo4j. for Auth.
-
-### database [string]
-
-database name.
-
-### query [string]
-
-Query statement. contain parameter placeholders that are substituted with the
corresponding values at runtime
-
-### queryParamPosition [object]
-
-position mapping information for query parameters.
-
-key name is parameter placeholder name.
-
-associated value is position of field in input data row.
-
-### max_transaction_retry_time [long]
-
-maximum transaction retry time(seconds). transaction fail if exceeded
-
-### max_connection_timeout [long]
-
-The maximum amount of time to wait for a TCP connection to be established
(seconds)
-
-### common options
-
-Sink plugin common parameters, please refer to [Sink Common
Options](../common-options/sink-common-options.md) for details
-
-## WriteOneByOneExample
-
-```
+## 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](../common-options/sink-common-options.md).
|
+
+## 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`.
+
+## Write One Row At A Time
+
+```bash
sink {
Neo4j {
uri = "neo4j://localhost:7687"
username = "neo4j"
- password = "1234"
+ password = "password"
database = "neo4j"
max_transaction_retry_time = 10
@@ -109,36 +55,57 @@ sink {
query = "CREATE (a:Person {name: $name, age: $age})"
queryParamPosition = {
- name = 0
- age = 1
+ name = 0
+ age = 1
}
}
}
```
-## WriteBatchExample
-> The unwind keyword provided by cypher supports batch writing, and the
default variable for a batch of data is batch. If you write a batch write
statement, then you should declare cypher:unwind $batch as row to do someting
-
+## Write In Batches
+
+```bash
+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 = "bolt://localhost:7687"
+ uri = "neo4j://localhost:7687"
username = "neo4j"
- password = "neo4j"
+ password = "password"
database = "neo4j"
- max_batch_size = 1000
- write_mode = "BATCH"
+ write_mode = "BATCH"
+ max_batch_size = 500
max_transaction_retry_time = 3
max_connection_timeout = 10
- query = "unwind $batch as row create(n:MyLabel) set n.name =
row.name,n.age = row.age"
+ queryParamPosition = {
+ name = 0
+ age = 1
+ }
+ query = "UNWIND $batch AS row CREATE (n:BatchLabel) SET n.name = row.name,
n.age = row.age"
}
}
```
## Changelog
-<ChangeLog />
\ No newline at end of file
+<ChangeLog />
diff --git a/docs/en/connectors/source/Neo4j.md
b/docs/en/connectors/source/Neo4j.md
index 21b6e11a11..0ce4c4906e 100644
--- a/docs/en/connectors/source/Neo4j.md
+++ b/docs/en/connectors/source/Neo4j.md
@@ -6,11 +6,12 @@ import ChangeLog from '../changelog/connector-neo4j.md';
## Description
-Read data from Neo4j.
+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
+`neo4j-java-driver` version: 4.4.9
-## Key features
+## Key Features
- [x] [batch](../../introduction/concepts/connector-v2-features.md)
- [ ] [stream](../../introduction/concepts/connector-v2-features.md)
@@ -19,85 +20,83 @@ Read data from Neo4j.
- [ ] [parallelism](../../introduction/concepts/connector-v2-features.md)
- [ ] [support user-defined
split](../../introduction/concepts/connector-v2-features.md)
-## Options
-
-| name | type | required | default value |
-|----------------------------|--------|----------|---------------|
-| uri | String | Yes | - |
-| username | String | No | - |
-| password | String | No | - |
-| bearer_token | String | No | - |
-| kerberos_ticket | String | No | - |
-| database | String | Yes | - |
-| query | String | Yes | - |
-| schema | Object | Yes | - |
-| max_transaction_retry_time | Long | No | 30 |
-| max_connection_timeout | Long | No | 30 |
-
-### uri [string]
-
-The URI of the Neo4j database. Refer to a case: `neo4j://localhost:7687`
-
-### username [string]
-
-username of the Neo4j
-
-### password [string]
-
-password of the Neo4j. required if `username` is provided
-
-### bearer_token [string]
-
-base64 encoded bearer token of the Neo4j. for Auth.
-
-### kerberos_ticket [string]
-
-base64 encoded kerberos ticket of the Neo4j. for Auth.
-
-### database [string]
-
-database name.
-
-### query [string]
-
-Query statement.
-
-### schema.fields [string]
-
-returned fields of `query`
-
-see [column projection](../../introduction/concepts/connector-v2-features.md)
-
-### max_transaction_retry_time [long]
-
-maximum transaction retry time(seconds). transaction fail if exceeded
-
-### max_connection_timeout [long]
-
-The maximum amount of time to wait for a TCP connection to be established
(seconds)
-
-## Example
+## Data Type Mapping
+
+| Neo4j Value Type | SeaTunnel Data Type |
+|------------------|---------------------|
+| String | STRING |
+| Boolean | BOOLEAN |
+| Integer | INT / BIGINT |
+| Float | FLOAT / DOUBLE |
+| ByteArray | BYTES |
+| Date | DATE |
+| 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.
+
+## Task Example
+
+```bash
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
-```
source {
- Neo4j {
- uri = "neo4j://localhost:7687"
- username = "neo4j"
- password = "1234"
- database = "neo4j"
-
- max_transaction_retry_time = 1
- max_connection_timeout = 1
-
- query = "MATCH (a:Person) RETURN a.name, a.age"
-
- schema {
- fields {
- a.age=INT
- a.name=STRING
- }
- }
+ 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 {}
}
```
diff --git a/docs/zh/connectors/sink/Neo4j.md b/docs/zh/connectors/sink/Neo4j.md
index ce4b94f8ee..85f2b09089 100644
--- a/docs/zh/connectors/sink/Neo4j.md
+++ b/docs/zh/connectors/sink/Neo4j.md
@@ -2,134 +2,106 @@ import ChangeLog from '../changelog/connector-neo4j.md';
# Neo4j
-> Neo4j 写连接器
+> Neo4j Sink 连接器
## 描述
-写数据到 `Neo4j`。
+Neo4j Sink 连接器通过执行 Cypher 语句把 SeaTunnel 数据写入 Neo4j。它支持逐条写入，
+也支持使用 Cypher `UNWIND` 批量写入。
-`neo4j-java-driver` version 4.4.9
+`neo4j-java-driver` 版本：4.4.9
-## 主要功能
+## 主要特性
- [ ] [精确一次](../../introduction/concepts/connector-v2-features.md)
-## 配置选项
-
-| 名称 | 类型 | 是否必须 | 默认值 |
-|----------------------------|---------|------|----------|
-| uri | String | 是 | - |
-| username | String | 否 | - |
-| password | String | 否 | - |
-| max_batch_size | Integer | 否 | - |
-| write_mode | String | 否 | OneByOne |
-| bearer_token | String | 否 | - |
-| kerberos_ticket | String | 否 | - |
-| database | String | 是 | - |
-| query | String | 是 | - |
-| queryParamPosition | Object | 是 | - |
-| max_transaction_retry_time | Long | 否 | 30 |
-| max_connection_timeout | Long | 否 | 30 |
-| common-options | config | 否 | - |
-
-### uri [string]
-
-`Neo4j`数据库的URI，参考配置： `neo4j://localhost:7687`。
-
-### username [string]
-
-`Neo4j`用户名。
-
-### password [string]
-
-`Neo4j`密码。如果提供了“用户名”，则需要。
-
-### max_batch_size [Integer]
-
-`max_batch_size` 是指写入数据时，单个事务中可以写入的最大数据条目数。
-
-### write_mode
-
-默认值为 `oneByOne` ，如果您想批量写入，请将其设置为`Batch`
-
-```cypher
-unwind $ttt as row create (n:Label) set n.name = row.name,n.age = rw.age
-```
-
-`ttt`代表一批数据。，`ttt`可以是任意字符串，只要它与配置的`batch_data_variable` 匹配。
-
-### bearer_token [string]
-
-`Neo4j`的`base64`编码`bearer token`用于鉴权。
-
-### kerberos_ticket [string]
-
-`Neo4j`的`base64`编码`kerberos ticket`用于鉴权。
-
-### database [string]
-
-数据库名称。
-
-### query [string]
-
-查询语句。包含在运行时用相应值替换的参数占位符。
-
-### queryParamPosition [object]
-
-查询参数的位置映射信息。
-
-键名是参数占位符名称。
-
-关联值是字段在输入数据行中的位置。
-
-### max_transaction_retry_time [long]
-
-最大事务重试时间（秒）。如果超过，则交易失败。
-
-### max_connection_timeout [long]
-
-等待TCP连接建立的最长时间（秒）。
-
-### common options
-
-Sink插件常用参数， 详细信息请参考 [Sink公共配置](../common-options/sink-common-options.md)
-
-## OneByOne模式写示例
-
-```
+## Sink 选项
+
+| 名称 | 类型 | 是否必填 | 默认值 | 描述
|
+|----------------------------|---------|------|------------|--------------------------------------------------------------------------------------------------------------|
+| uri | String | 是 | - | Neo4j 连接地址，例如
`neo4j://localhost:7687` 或 `bolt://localhost:7687`。
|
+| username | String | 否 | - | Neo4j 用户名，需要和
`password` 一起使用。`username`、`bearer_token`、`kerberos_ticket` 三种认证方式至少配置一种。
|
+| password | String | 否 | - | Neo4j 密码。配置
`username` 时必须配置。
|
+| bearer_token | String | 否 | - | 用于 Neo4j 认证的
bearer token。
|
+| kerberos_ticket | String | 否 | - | 用于 Neo4j 认证的
Kerberos ticket。
|
+| database | String | 是 | - | Neo4j 数据库名。
|
+| query | String | 是 | - | 写入数据使用的 Cypher
语句。`ONE_BY_ONE` 模式使用 `$name` 这类占位符；`BATCH` 模式使用 `UNWIND $batch AS row`。
|
+| queryParamPosition | Object | 是 | - | Cypher
参数名和输入行字段位置的映射。连接器配置校验要求必须填写。
|
+| max_batch_size | Integer | 否 | 500 | `write_mode =
"BATCH"` 时，单个事务最多写入的数据条数，必须大于 0。
|
+| write_mode | String | 否 | ONE_BY_ONE | 写入模式。可选值为
`ONE_BY_ONE` 和 `BATCH`。
|
+| max_transaction_retry_time | Long | 否 | 30 | 最大事务重试时间，单位为秒。
|
+| max_connection_timeout | Long | 否 | 30 | 建立 TCP
连接的最大等待时间，单位为秒。
|
+| common-options | config | 否 | - | Sink 通用选项，详见
[Sink 通用选项](../common-options/sink-common-options.md)。
|
+
+## 注意事项
+
+- 认证方式只选一种：用户名密码、bearer token 或 Kerberos ticket。
+- `ONE_BY_ONE` 模式下，`queryParamPosition` 用来把 Cypher 占位符映射到输入行的字段位置。
+- `BATCH` 模式下，查询语句应使用 `UNWIND $batch AS row`，连接器会通过 `batch` 变量传入一批数据。
+- `BATCH` 模式虽然从 `row` 中取值，但连接器配置校验仍要求填写 `queryParamPosition`。
+
+## 逐条写入示例
+
+```bash
sink {
Neo4j {
uri = "neo4j://localhost:7687"
username = "neo4j"
- password = "1234"
+ 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
+ name = 0
+ age = 1
}
}
}
```
-## Batch模式写示例
-> cypher提供的`unwind`关键字支持批量写入，
-> 批量数据的默认变量是batch。如果你写一个批处理写语句，
-> 那么你应该声明 cypher `unwind $batch` 作为行
-```
+## 批量写入示例
+
+```bash
+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 = "bolt://localhost:7687"
+ uri = "neo4j://localhost:7687"
username = "neo4j"
- password = "neo4j"
+ password = "password"
database = "neo4j"
- max_batch_size = 1000
+
write_mode = "BATCH"
+ max_batch_size = 500
max_transaction_retry_time = 3
max_connection_timeout = 10
- query = "unwind $batch as row create(n:MyLabel) set n.name =
row.name,n.age = row.age"
+
+ queryParamPosition = {
+ name = 0
+ age = 1
+ }
+
+ query = "UNWIND $batch AS row CREATE (n:BatchLabel) SET n.name = row.name,
n.age = row.age"
}
}
```
diff --git a/docs/zh/connectors/source/Neo4j.md
b/docs/zh/connectors/source/Neo4j.md
index de12e16f89..35c57d6509 100644
--- a/docs/zh/connectors/source/Neo4j.md
+++ b/docs/zh/connectors/source/Neo4j.md
@@ -2,99 +2,101 @@ import ChangeLog from '../changelog/connector-neo4j.md';
# Neo4j
-> Neo4j 源连接器器
+> Neo4j 源连接器
## 描述
-从 `Neo4j` 读取数据
+Neo4j 源连接器通过执行 Cypher 查询从 Neo4j 读取数据，并把查询返回字段映射成 SeaTunnel
+表结构。
-`neo4j-java-driver` 版本 4.4.9
+`neo4j-java-driver` 版本：4.4.9
-## 主要功能
+## 主要特性
- [x] [批处理](../../introduction/concepts/connector-v2-features.md)
- [ ] [流处理](../../introduction/concepts/connector-v2-features.md)
- [ ] [精确一次](../../introduction/concepts/connector-v2-features.md)
- [x] [列投影](../../introduction/concepts/connector-v2-features.md)
- [ ] [并行度](../../introduction/concepts/connector-v2-features.md)
-- [ ] [支持用户定义拆分](../../introduction/concepts/connector-v2-features.md)
-
-## 配置选项
-
-| 名称 | 类型 | 是否必须 | 默认值 |
-|----------------------------|--------|------|-----|
-| uri | String | 是 | - |
-| username | String | 否 | - |
-| password | String | 否 | - |
-| bearer_token | String | 否 | - |
-| kerberos_ticket | String | 否 | - |
-| database | String | 是 | - |
-| query | String | 是 | - |
-| schema | Object | 是 | - |
-| max_transaction_retry_time | Long | 否 | 30 |
-| max_connection_timeout | Long | 否 | 30 |
-
-### uri [string]
-
-`Neo4j`数据库的URI，参考配置： `neo4j://localhost:7687`。
-
-### username [string]
-
-`Neo4j`用户名。
-
-### password [string]
-
-`Neo4j`密码。如果提供了“用户名”，则需要。
-
-### bearer_token [string]
-
-`Neo4j`的`base64`编码`bearer token`用于鉴权。
-
-### kerberos_ticket [string]
-
-`Neo4j`的`base64`编码`kerberos ticket`用于鉴权。
-
-### database [string]
-
-数据库名。
-
-### query [string]
-
-查询语句。
-
-### schema.fields [string]
-
-返回`query` 的字段。
-
-查看 [列投影](../../introduction/concepts/connector-v2-features.md)
-
-### max_transaction_retry_time [long]
-
-最大事务重试时间（秒）。如果超过，则事务失败。
-
-### max_connection_timeout [long]
-
-等待TCP连接建立的最长时间（秒）。
-
-## 示例
+- [ ] [支持用户自定义切分](../../introduction/concepts/connector-v2-features.md)
+
+## 数据类型映射
+
+| Neo4j 值类型 | SeaTunnel 数据类型 |
+|---------------|--------------------|
+| String | STRING |
+| Boolean | BOOLEAN |
+| Integer | INT / BIGINT |
+| Float | FLOAT / DOUBLE |
+| ByteArray | BYTES |
+| Date | DATE |
+| LocalDateTime | TIMESTAMP |
+| List | ARRAY |
+| Map | MAP |
+| Null | NULL |
+
+## 源选项
+
+| 名称 | 类型 | 是否必填 | 默认值 | 描述
|
+|----------------------------|--------|------|-----|-----------------------------------------------------------------------------------------------------|
+| uri | String | 是 | - | Neo4j 连接地址，例如
`neo4j://localhost:7687` 或 `bolt://localhost:7687`。
|
+| username | String | 否 | - | Neo4j 用户名，需要和 `password`
一起使用。`username`、`bearer_token`、`kerberos_ticket` 三种认证方式至少配置一种。 |
+| password | String | 否 | - | Neo4j 密码。配置 `username`
时必须配置。 |
+| bearer_token | String | 否 | - | 用于 Neo4j 认证的 bearer
token。 |
+| kerberos_ticket | String | 否 | - | 用于 Neo4j 认证的 Kerberos
ticket。 |
+| database | String | 是 | - | Neo4j 数据库名。
|
+| query | String | 是 | - | 读取数据使用的 Cypher
查询语句。查询返回字段必须和 `schema.fields` 对应。 |
+| schema | Object | 是 | - | 查询结果对应的 SeaTunnel 表结构，在
`schema.fields` 中配置字段名和类型。 |
+| max_transaction_retry_time | Long | 否 | 30 | 最大事务重试时间，单位为秒。
|
+| max_connection_timeout | Long | 否 | 30 | 建立 TCP 连接的最大等待时间，单位为秒。
|
+
+## 注意事项
+
+- 认证方式只选一种：用户名密码、bearer token 或 Kerberos ticket。
+- `query` 决定返回哪些字段，`schema.fields` 必须写清这些返回字段和对应类型。
+- 查询返回字段名可以包含点号，例如从节点属性返回的 `t.string`。
+
+## 任务示例
+
+```bash
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
-```
source {
- Neo4j {
- uri = "neo4j://localhost:7687"
- username = "neo4j"
- password = "1234"
- database = "neo4j"
- max_transaction_retry_time = 1
- max_connection_timeout = 1
- query = "MATCH (a:Person) RETURN a.name, a.age"
- schema {
- fields {
- a.age=INT
- a.name=STRING
- }
- }
+ 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 {}
}
```
