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 2b05efa91a [Docs][Connector-V2] Improve GraphQL connector docs (#11265)
2b05efa91a is described below
commit 2b05efa91a237f3341b6d216aaac475ae1d3bd0d
Author: Daniel Carter <[email protected]>
AuthorDate: Fri Jul 3 18:14:51 2026 +0800
[Docs][Connector-V2] Improve GraphQL connector docs (#11265)
Co-authored-by: DanielCarter-stack
<[email protected]>
Co-authored-by: Shenghang <[email protected]>
---
docs/en/connectors/sink/GraphQL.md | 237 ++++++++++++++----------------
docs/en/connectors/source/GraphQL.md | 276 ++++++++++++++++-------------------
docs/zh/connectors/sink/GraphQL.md | 239 ++++++++++++++----------------
docs/zh/connectors/source/GraphQL.md | 269 ++++++++++++++++------------------
4 files changed, 477 insertions(+), 544 deletions(-)
diff --git a/docs/en/connectors/sink/GraphQL.md
b/docs/en/connectors/sink/GraphQL.md
index fe61172e32..0852316125 100644
--- a/docs/en/connectors/sink/GraphQL.md
+++ b/docs/en/connectors/sink/GraphQL.md
@@ -10,6 +10,12 @@ import ChangeLog from '../changelog/connector-graphql.md';
> Flink<br/>
> SeaTunnel Zeta<br/>
+## Description
+
+The GraphQL sink connector writes SeaTunnel rows to a GraphQL service by
sending a GraphQL
+`mutation` over HTTP POST. For each input row, the sink adds row fields to
GraphQL `variables`
+with the same field names, then sends the configured mutation.
+
## Key Features
- [ ] [exactly-once](../../introduction/concepts/connector-v2-features.md)
@@ -17,42 +23,43 @@ import ChangeLog from '../changelog/connector-graphql.md';
- [x] [support multiple table
write](../../introduction/concepts/connector-v2-features.md)
- [ ] [timer flush](../../introduction/concepts/connector-v2-features.md)
-## Description
-
-Used to launch web hooks using data.
-
-> For example, if the data from upstream is [`label: {"__name__": "test1"},
value: 1.2.3,time:2024-08-15T17:00:00`], the body content is the following:
`{"label":{"__name__": "test1"}, "value":"1.23","time":"2024-08-15T17:00:00"}`
-
-**Tips: GraphQL sink only support `post json` webhook and the data from source
will be treated as body content in web hook.And does not support passing past
data**
-
## Supported DataSource Info
-In order to use the GraphQL connector, the following dependencies are required.
-They can be downloaded via install-plugin.sh or from the Maven central
repository.
+In order to use the GraphQL connector, the following dependency is required.
+It can be downloaded via install-plugin.sh or from Maven central repository.
-| Datasource | Supported Versions |
Dependency |
-|------------|--------------------|------------------------------------------------------------------------------------------------------------------|
-| Http | universal |
[Download](https://mvnrepository.com/artifact/org.apache.seatunnel/seatunnel-connectors-v2/connector-http)
|
+| Datasource | Supported Versions | Dependency |
+|------------|--------------------|------------|
+| GraphQL | universal |
[Download](https://mvnrepository.com/artifact/org.apache.seatunnel/connector-graphql)
|
## Sink Options
-| Name | Type | Required | Default | Description
|
-|-----------------------------|--------|----------|---------|-------------------------------------------------------------------------------------------------------------|
-| url | String | Yes | - | Http request url
|
-| query | String | Yes | - | GraphQL query |
-| variables | String | No | - | GraphQL variables |
-| valueCover | Boolean | No | - | Whether the data overwrites the variable
value |
-| headers | Map | No | - | Http headers
|
-| retry | Int | No | - | The max retry
times if request http return to `IOException`
|
-| retry_backoff_multiplier_ms | Int | No | 100 | The
retry-backoff times(millis) multiplier if request http failed
|
-| retry_backoff_max_ms | Int | No | 10000 | The maximum
retry-backoff times(millis) if request http failed
|
-| connect_timeout_ms | Int | No | 12000 | Connection
timeout setting, default 12s.
|
-| socket_timeout_ms | Int | No | 60000 | Socket timeout
setting, default 60s.
|
-| common-options | | No | - | Sink plugin
common parameters, please refer to [Sink Common
Options](../common-options/sink-common-options.md) for details |
-
-## Example
-
-simple:
+| Name | Type | Required | Default | Description |
+|---------------|---------|----------|---------|-------------|
+| url | String | Yes | - | GraphQL HTTP endpoint. Sink
mode requires `http://` or `https://`. |
+| query | String | Yes | - | GraphQL mutation statement.
Sink only supports `mutation`. |
+| variables | Map | No | - | Initial GraphQL variables.
Input row fields are added to this map before each request. |
+| valueCover | Boolean | No | false | When false, row fields
overwrite variables with the same name. When true, existing variables keep
their configured values. |
+| timeout | Long | No | - | Request timeout value passed
to the HTTP client parameter map. |
+| headers | Map | No | - | HTTP request headers, such as
authentication headers. |
+| params | Map | No | - | HTTP request parameters. |
+| retry | Int | No | - | Maximum retry times when an
HTTP request returns an IOException. |
+| retry_backoff_multiplier_ms | Int | No | 100 | HTTP retry backoff multiplier
in milliseconds. |
+| retry_backoff_max_ms | Int | No | 10000 | Maximum HTTP retry backoff in
milliseconds. |
+| connect_timeout_ms | Int | No | 12000 | HTTP connection timeout in
milliseconds. |
+| socket_timeout_ms | Int | No | 60000 | HTTP socket timeout in milliseconds. |
+| common-options | Config | No | - | Sink common options. See [Sink Common
Options](../common-options/sink-common-options.md). |
+
+## Notes
+
+- The sink validates that `query` is a GraphQL `mutation`.
+- Field names in the upstream row should match variable names used by the
mutation.
+- If `variables` already contains a key and `valueCover = true`, the
configured variable value is kept.
+- If `valueCover = false`, row field values replace variables with the same
name.
+
+## Task Example
+
+### Write Rows with a GraphQL Mutation
```hocon
env {
@@ -62,112 +69,90 @@ env {
source {
FakeSource {
- tables_configs = [
- {
- schema = {
- table = "graphql_sink_1"
- fields {
- id = int
- val_bool = boolean
- val_int8 = tinyint
- val_int16 = smallint
- val_int32 = int
- val_int64 = bigint
- val_float = float
- val_double = double
- val_decimal = "decimal(16, 1)"
- val_string = string
- val_unixtime_micros = timestamp
+ plugin_output = "fake"
+ schema = {
+ fields {
+ id = int
+ val_bool = boolean
+ val_int8 = tinyint
+ val_int16 = smallint
+ val_int32 = int
+ val_int64 = bigint
+ val_float = float
+ val_double = double
+ val_decimal = "decimal(16, 1)"
+ val_string = string
+ val_unixtime_micros = timestamp
}
- }
- rows = [
- {
- kind = INSERT
- fields = [1, true, 1, 2, 3, 4, 4.3,5.3,6.3, "NEW",
"2020-02-02T02:02:02"]
- }
- ]
- },
- {
- schema = {
- table = "graphql_sink_2"
- fields {
- id = int
- val_bool = boolean
- val_int8 = tinyint
- val_int16 = smallint
- val_int32 = int
- val_int64 = bigint
- val_float = float
- val_double = double
- val_decimal = "decimal(16, 1)"
- val_string = string
- val_unixtime_micros = timestamp
- }
- }
- rows = [
- {
- kind = INSERT
- fields = [2, true, 1, 2, 3, 4, 4.3,5.3,6.3, "NEW",
"2020-02-02T02:02:02"]
- }
- ]
+ }
+ rows = [
+ {
+ kind = INSERT
+ fields = [1, true, 1, 2, 3, 4, 4.3, 5.3, 6.3, "NEW",
"2020-02-02T02:02:02"]
}
]
}
}
sink {
- GraphQL {
- url = "http://192.168.1.103:9081/v1/graphql";
- query = """
- mutation MyMutation(
- $id: Int!
- $val_bool: Boolean!
- $val_int8: smallint!
- $val_int16: smallint!
- $val_int32: Int!
- $val_int64: bigint!
- $val_float: Float!
- $val_double: Float!
- $val_decimal: numeric!
- $val_string: String!
- $val_unixtime_micros: timestamp!
- ) {
- insert_sink(objects: {
- id: $id,
- val_bool: $val_bool,
- val_int8: $val_int8,
- val_int16: $val_int16,
- val_int32: $val_int32,
- val_int64: $val_int64,
- val_float: $val_float,
- val_double: $val_double,
- val_decimal: $val_decimal,
- val_string: $val_string,
- val_unixtime_micros: $val_unixtime_micros
- }) {
- affected_rows
- returning {
- id
- val_bool
- val_decimal
- val_double
- val_float
- val_int16
- val_int32
- val_int64
- val_int8
- val_string
- val_unixtime_micros
- }
- }
- }
- """
- variables = {
- "val_bool": True
+ GraphQL {
+ plugin_input = "fake"
+ url = "http://graphql:8080/v1/graphql";
+ query = """
+ mutation MyMutation(
+ $id: Int!
+ $val_bool: Boolean!
+ $val_int8: smallint!
+ $val_int16: smallint!
+ $val_int32: Int!
+ $val_int64: bigint!
+ $val_float: Float!
+ $val_double: Float!
+ $val_decimal: numeric!
+ $val_string: String!
+ $val_unixtime_micros: timestamp!
+ ) {
+ insert_sink(objects: {
+ id: $id,
+ val_bool: $val_bool,
+ val_int8: $val_int8,
+ val_int16: $val_int16,
+ val_int32: $val_int32,
+ val_int64: $val_int64,
+ val_float: $val_float,
+ val_double: $val_double,
+ val_decimal: $val_decimal,
+ val_string: $val_string,
+ val_unixtime_micros: $val_unixtime_micros
+ }) {
+ affected_rows
}
- }
+ }
+ """
+ }
}
+```
+
+### Keep a Configured Variable Value
+```hocon
+sink {
+ GraphQL {
+ plugin_input = "fake"
+ url = "http://graphql:8080/v1/graphql";
+ valueCover = true
+ variables = {
+ val_bool = true
+ }
+ query = """
+ mutation MyMutation($id: Int!, $val_bool: Boolean!) {
+ insert_sink(objects: {id: $id, val_bool: $val_bool}) {
+ affected_rows
+ }
+ }
+ """
+ }
+}
```
## Changelog
diff --git a/docs/en/connectors/source/GraphQL.md
b/docs/en/connectors/source/GraphQL.md
index 73b228c71f..b8df97da35 100644
--- a/docs/en/connectors/source/GraphQL.md
+++ b/docs/en/connectors/source/GraphQL.md
@@ -4,175 +4,157 @@ import ChangeLog from '../changelog/connector-graphql.md';
> GraphQL source connector
+## Support Those Engines
+
+> Spark<br/>
+> Flink<br/>
+> SeaTunnel Zeta<br/>
+
## Description
-Used to read data from GraphQL.
+The GraphQL source connector reads data from a GraphQL endpoint. It supports:
-## Key features
+- HTTP batch or polling reads with a GraphQL `query`.
+- WebSocket subscription reads with a GraphQL `subscription`.
+- JSON response parsing by `content_field` and `schema.fields`.
+- Request headers, request parameters, GraphQL variables, and timeout settings.
+
+## Key Features
- [x] [batch](../../introduction/concepts/connector-v2-features.md)
- [x] [stream](../../introduction/concepts/connector-v2-features.md)
- [ ] [parallelism](../../introduction/concepts/connector-v2-features.md)
-## Options
-
-| name | type | required | default value |
-| --------------------------- | ------- | -------- | ----------------------- |
-| url | String | Yes | - |
-| query | String | Yes | - |
-| variables | Config | No | - |
-| enable_subscription | boolean | No | false |
-| timeout | Long | No | - |
-| content_field | String | Yes | $.data.{query_object}.* |
-| schema.fields | Config | Yes | - |
-| params | Map | Yes | - |
-| poll_interval_millis | int | No | - |
-| retry | int | No | - |
-| retry_backoff_multiplier_ms | int | No | 100 |
-| retry_backoff_max_ms | int | No | 10000 |
-| enable_multi_lines | boolean | No | false |
-| common-options | config | No | - |
-
-### url [String]
-
-http request url
-
-### query [String]
-
-GraphQL expression query string
-
-### variables [String]
-
-GraphQL Variables
-
-for example
-
-```
-variables = {
- limit = 2
-}
-```
-
-### enable_subscription [boolean]
-
-1. true : Enable streaming subscription mode (WebSocket)
-2. false : Enable batch query mode (HTTP)
-
-### timeout [Long]
-
-Time-out Period
-
-### content_field [String]
-
-JSONPath wildcard
-
-### params [Map]
-
-http request params
-
-### poll_interval_millis [int]
-
-request http api interval(millis) in stream mode
-
-### retry [int]
-
-The max retry times if request http return to `IOException`
-
-### retry_backoff_multiplier_ms [int]
-
-The retry-backoff times(millis) multiplier if request http failed
-
-### retry_backoff_max_ms [int]
-
-The maximum retry-backoff times(millis) if request http failed
-
-### schema [Config]
-
-Fill in a fixed value
+## Supported DataSource Info
+
+In order to use the GraphQL connector, the following dependency is required.
+It can be downloaded via install-plugin.sh or from Maven central repository.
+
+| Datasource | Supported Versions | Dependency |
+|------------|--------------------|------------|
+| GraphQL | universal |
[Download](https://mvnrepository.com/artifact/org.apache.seatunnel/connector-graphql)
|
+
+## Source Options
+
+| Name | Type | Required | Default | Description |
+|---------------------|---------|----------|---------|-------------|
+| url | String | Yes | - | GraphQL endpoint. Use
`http://` or `https://` for query mode. Use `ws://` or `wss://` for
subscription mode. |
+| query | String | Yes | - | GraphQL statement.
Source supports `query`, and supports `subscription` only when
`enable_subscription = true`. |
+| variables | Map | No | - | GraphQL variables sent
with the request body. |
+| enable_subscription | Boolean | No | false | Whether to use
WebSocket subscription mode. When false, the connector uses HTTP POST. |
+| timeout | Long | No | - | Request timeout value
passed to the HTTP client parameter map. |
+| headers | Map | No | - | HTTP request headers,
such as authentication headers. |
+| params | Map | No | - | HTTP request
parameters. |
+| format | String | No | TEXT | Response format
inherited from HTTP source. Use `json` when `schema.fields` is configured. |
+| content_field | String | No | - | JSONPath used to pick
the data array or object from the GraphQL response, for example
`$.data.source`. |
+| schema.fields | Config | No | - | Output fields and
SeaTunnel data types. Configure it when reading structured JSON rows. |
+| poll_interval_millis | Int | No | - | Interval between HTTP
requests in streaming query mode. |
+| max_retries | Int | No | 5 | Maximum reconnect
attempts for WebSocket subscription mode. |
+| retry_delay_ms | Int | No | 5000 | Delay between WebSocket
reconnect attempts, in milliseconds. |
+| retry | Int | No | - | Maximum HTTP retry
times when an HTTP request returns an IOException. |
+| retry_backoff_multiplier_ms | Int | No | 100 | HTTP retry backoff
multiplier in milliseconds. |
+| retry_backoff_max_ms | Int | No | 10000 | Maximum HTTP retry
backoff in milliseconds. |
+| enable_multi_lines | Boolean | No | false | Whether to split the
HTTP response by line before parsing. |
+| connect_timeout_ms | Int | No | 12000 | HTTP connection timeout
in milliseconds. |
+| socket_timeout_ms | Int | No | 60000 | HTTP socket timeout in
milliseconds. |
+| common-options | Config | No | - | Source common options.
See [Source Common Options](../common-options/source-common-options.md). |
+
+## Notes
+
+- In normal query mode, `url` must start with `http://` or `https://`.
+- In subscription mode, set `enable_subscription = true` and use a `ws://` or
`wss://` URL.
+- Source queries cannot be GraphQL `mutation` operations.
+- `content_field` is usually needed because GraphQL responses are commonly
wrapped under `data`.
+
+## Task Example
+
+### Query GraphQL Data
```hocon
- schema = {
- fields {
- metric = "map<string, string>"
- value = double
- time = long
- }
- }
-
-```
-
-#### fields [Config]
-
-the schema fields of upstream data
-
-### common options
-
-Source plugin common parameters, please refer to [Source Common
Options](../common-options/source-common-options.md) for details
-
-## Example
-
-### Query
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
-```hocon
source {
- GraphQL {
- url = "http://192.168.1.103:9081/v1/graphql";
- content_field = "$.data.source"
- query = """
- query MyQuery($limit: Int) {
- source(limit: $limit) {
- id
- val_bool
- val_double
- val_float
- }
- }
- """
- variables = {
- limit = 2
- }
- schema = {
- fields {
- id = "int"
- val_bool = "boolean"
- val_double = "double"
- val_float = "float"
- }
+ GraphQL {
+ plugin_output = "graphql_source"
+ url = "http://graphql:8080/v1/graphql";
+ format = "json"
+ content_field = "$.data.source"
+ query = """
+ query MyQuery($limit: Int) {
+ source(limit: $limit) {
+ id
+ val_bool
+ val_double
+ val_float
}
+ }
+ """
+ variables = {
+ limit = 2
}
+ schema = {
+ fields {
+ id = "int"
+ val_bool = "boolean"
+ val_double = "double"
+ val_float = "float"
+ }
+ }
+ }
+}
+
+sink {
+ Console {
+ plugin_input = "graphql_source"
+ }
}
```
-### Subscription
+### Subscribe to GraphQL Data
```hocon
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+}
+
source {
- GraphQL {
- url = "http://192.168.1.103:9081/v1/graphql";
- content_field = "$.data.source"
- query = """
- query MyQuery($limit: Int) {
- source(limit: $limit) {
- id
- val_bool
- val_double
- val_float
- }
- }
- """
- variables = {
- limit = 2
- }
- enable_subscription = true
- schema = {
- fields {
- id = "int"
- val_bool = "boolean"
- val_double = "double"
- val_float = "float"
- }
+ GraphQL {
+ plugin_output = "graphql_subscription"
+ url = "ws://graphql:8080/v1/graphql"
+ format = "json"
+ content_field = "$.data.source"
+ enable_subscription = true
+ max_retries = 5
+ retry_delay_ms = 5000
+ query = """
+ subscription MySubscription {
+ source {
+ id
+ val_bool
+ val_double
+ val_float
}
+ }
+ """
+ schema = {
+ fields {
+ id = "int"
+ val_bool = "boolean"
+ val_double = "double"
+ val_float = "float"
+ }
}
+ }
+}
+
+sink {
+ Console {
+ plugin_input = "graphql_subscription"
+ }
}
```
diff --git a/docs/zh/connectors/sink/GraphQL.md
b/docs/zh/connectors/sink/GraphQL.md
index 8abf9854c2..0d23e35d55 100644
--- a/docs/zh/connectors/sink/GraphQL.md
+++ b/docs/zh/connectors/sink/GraphQL.md
@@ -2,56 +2,63 @@ import ChangeLog from '../changelog/connector-graphql.md';
# GraphQL
-> GraphQL sink 连接器
+> GraphQL Sink 连接器
-## 支持的引擎
+## 支持引擎
> Spark<br/>
> Flink<br/>
> SeaTunnel Zeta<br/>
+## 描述
+
+GraphQL Sink 连接器通过 HTTP POST 向 GraphQL 服务写入数据。它会对每一行输入数据发送一条
+GraphQL `mutation`：输入行里的字段会按同名字段放入 GraphQL `variables`，然后随配置的
+mutation 一起发送。
+
## 主要特性
-- [ ] [[精确一次]](../../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)
-- [ ] [timer flush](../../introduction/concepts/connector-v2-features.md)
-
-## 描述
-
-接收Source端传入的数据，利用数据触发 web hooks。
+- [ ] [定时刷新](../../introduction/concepts/connector-v2-features.md)
-> 例如，来自上游的数据为 [`label: {"__name__": "test1"}, value:
1.2.3,time:2024-08-15T17:00:00`], 则body内容如下: `{"label":{"__name__": "test1"},
"value":"1.23","time":"2024-08-15T17:00:00"}`
+## 支持的数据源信息
-**Tips: GraphQL 数据接收器 仅支持 `post json` 类型的 web hook，source 数据将被视为 webhook 中的
body 内容。并且不支持传递过去太久的数据**
+使用 GraphQL 连接器需要安装下面的依赖。可以通过 install-plugin.sh 安装，也可以从 Maven 中央仓库下载。
-## 支持的数据源信息
+| 数据源 | 支持版本 | 依赖 |
+|---------|----------|------|
+| GraphQL | 通用 |
[下载](https://mvnrepository.com/artifact/org.apache.seatunnel/connector-graphql)
|
-想使用 GraphQL 连接器，需要安装以下必要的依赖。可以通过运行 install-plugin.sh 脚本或者从 Maven 中央仓库下载这些依赖
+## Sink 选项
-| 数据源 | 支持版本 | 依赖 |
-| ------ | --------- |
------------------------------------------------------------ |
-| Http | universal |
[Download](https://mvnrepository.com/artifact/org.apache.seatunnel/seatunnel-connectors-v2/connector-http)
|
+| 名称 | 类型 | 是否必填 | 默认值 | 描述 |
+|---------------|---------|----------|--------|------|
+| url | String | 是 | - | GraphQL HTTP 服务地址。Sink 模式要求使用
`http://` 或 `https://`。 |
+| query | String | 是 | - | GraphQL mutation 语句。Sink 只支持
`mutation`。 |
+| variables | Map | 否 | - | 初始 GraphQL
变量。每次请求前，输入行字段会加入这个变量表。 |
+| valueCover | Boolean | 否 | false | 为 false 时，同名输入行字段会覆盖配置里的变量；为
true 时，保留配置里的变量值。 |
+| timeout | Long | 否 | - | 传给 HTTP 客户端参数的请求超时时间。 |
+| headers | Map | 否 | - | HTTP 请求头，例如鉴权请求头。 |
+| params | Map | 否 | - | HTTP 请求参数。 |
+| retry | Int | 否 | - | HTTP 请求出现 IOException 时的最大重试次数。
|
+| retry_backoff_multiplier_ms | Int | 否 | 100 | HTTP 请求失败后的重试退避倍率，单位毫秒。 |
+| retry_backoff_max_ms | Int | 否 | 10000 | HTTP 请求失败后的最大重试退避时间，单位毫秒。 |
+| connect_timeout_ms | Int | 否 | 12000 | HTTP 连接超时时间，单位毫秒。 |
+| socket_timeout_ms | Int | 否 | 60000 | HTTP socket 超时时间，单位毫秒。 |
+| common-options | Config | 否 | - | Sink 通用参数，详见 [Sink Common
Options](../common-options/sink-common-options.md)。 |
-## 接收器选项
+## 注意事项
-| Name | Type | Required | Default | Description
|
-|-----------------------------|--------|----------|---------|-------------------------------------------------------------------------------------------------------------|
-| url | String | Yes | - | Http request url
|
-| query | String | Yes | - | GraphQL query |
-| variables | String | No | - | GraphQL variables |
-| valueCover | Boolean | No | - | Whether the data overwrites the variable
value |
-| headers | Map | No | - | Http headers
|
-| retry | Int | No | - | The max retry
times if request http return to `IOException`
|
-| retry_backoff_multiplier_ms | Int | No | 100 | The
retry-backoff times(millis) multiplier if request http failed
|
-| retry_backoff_max_ms | Int | No | 10000 | The maximum
retry-backoff times(millis) if request http failed
|
-| connect_timeout_ms | Int | No | 12000 | Connection
timeout setting, default 12s.
|
-| socket_timeout_ms | Int | No | 60000 | Socket timeout
setting, default 60s.
|
-| common-options | | No | - | Sink plugin
common parameters, please refer to [Sink Common
Options](../common-options/sink-common-options.md) for details |
+- Sink 会校验 `query` 必须是 GraphQL `mutation`。
+- 上游字段名应和 mutation 里使用的变量名保持一致。
+- 如果 `variables` 已有某个 key，并且设置 `valueCover = true`，会保留配置里的变量值。
+- 如果 `valueCover = false`，同名输入行字段会覆盖配置里的变量值。
-## 示例
+## 任务示例
-简单示例:
+### 使用 GraphQL Mutation 写入数据
```hocon
env {
@@ -61,112 +68,90 @@ env {
source {
FakeSource {
- tables_configs = [
- {
- schema = {
- table = "graphql_sink_1"
- fields {
- id = int
- val_bool = boolean
- val_int8 = tinyint
- val_int16 = smallint
- val_int32 = int
- val_int64 = bigint
- val_float = float
- val_double = double
- val_decimal = "decimal(16, 1)"
- val_string = string
- val_unixtime_micros = timestamp
+ plugin_output = "fake"
+ schema = {
+ fields {
+ id = int
+ val_bool = boolean
+ val_int8 = tinyint
+ val_int16 = smallint
+ val_int32 = int
+ val_int64 = bigint
+ val_float = float
+ val_double = double
+ val_decimal = "decimal(16, 1)"
+ val_string = string
+ val_unixtime_micros = timestamp
}
- }
- rows = [
- {
- kind = INSERT
- fields = [1, true, 1, 2, 3, 4, 4.3,5.3,6.3, "NEW",
"2020-02-02T02:02:02"]
- }
- ]
- },
- {
- schema = {
- table = "graphql_sink_2"
- fields {
- id = int
- val_bool = boolean
- val_int8 = tinyint
- val_int16 = smallint
- val_int32 = int
- val_int64 = bigint
- val_float = float
- val_double = double
- val_decimal = "decimal(16, 1)"
- val_string = string
- val_unixtime_micros = timestamp
- }
- }
- rows = [
- {
- kind = INSERT
- fields = [2, true, 1, 2, 3, 4, 4.3,5.3,6.3, "NEW",
"2020-02-02T02:02:02"]
- }
- ]
+ }
+ rows = [
+ {
+ kind = INSERT
+ fields = [1, true, 1, 2, 3, 4, 4.3, 5.3, 6.3, "NEW",
"2020-02-02T02:02:02"]
}
]
}
}
sink {
- GraphQL {
- url = "http://192.168.1.103:9081/v1/graphql";
- query = """
- mutation MyMutation(
- $id: Int!
- $val_bool: Boolean!
- $val_int8: smallint!
- $val_int16: smallint!
- $val_int32: Int!
- $val_int64: bigint!
- $val_float: Float!
- $val_double: Float!
- $val_decimal: numeric!
- $val_string: String!
- $val_unixtime_micros: timestamp!
- ) {
- insert_sink(objects: {
- id: $id,
- val_bool: $val_bool,
- val_int8: $val_int8,
- val_int16: $val_int16,
- val_int32: $val_int32,
- val_int64: $val_int64,
- val_float: $val_float,
- val_double: $val_double,
- val_decimal: $val_decimal,
- val_string: $val_string,
- val_unixtime_micros: $val_unixtime_micros
- }) {
- affected_rows
- returning {
- id
- val_bool
- val_decimal
- val_double
- val_float
- val_int16
- val_int32
- val_int64
- val_int8
- val_string
- val_unixtime_micros
- }
- }
- }
- """
- variables = {
- "val_bool": True
+ GraphQL {
+ plugin_input = "fake"
+ url = "http://graphql:8080/v1/graphql";
+ query = """
+ mutation MyMutation(
+ $id: Int!
+ $val_bool: Boolean!
+ $val_int8: smallint!
+ $val_int16: smallint!
+ $val_int32: Int!
+ $val_int64: bigint!
+ $val_float: Float!
+ $val_double: Float!
+ $val_decimal: numeric!
+ $val_string: String!
+ $val_unixtime_micros: timestamp!
+ ) {
+ insert_sink(objects: {
+ id: $id,
+ val_bool: $val_bool,
+ val_int8: $val_int8,
+ val_int16: $val_int16,
+ val_int32: $val_int32,
+ val_int64: $val_int64,
+ val_float: $val_float,
+ val_double: $val_double,
+ val_decimal: $val_decimal,
+ val_string: $val_string,
+ val_unixtime_micros: $val_unixtime_micros
+ }) {
+ affected_rows
}
- }
+ }
+ """
+ }
}
+```
+### 保留配置里的变量值
+
+```hocon
+sink {
+ GraphQL {
+ plugin_input = "fake"
+ url = "http://graphql:8080/v1/graphql";
+ valueCover = true
+ variables = {
+ val_bool = true
+ }
+ query = """
+ mutation MyMutation($id: Int!, $val_bool: Boolean!) {
+ insert_sink(objects: {id: $id, val_bool: $val_bool}) {
+ affected_rows
+ }
+ }
+ """
+ }
+}
```
## 变更日志
diff --git a/docs/zh/connectors/source/GraphQL.md
b/docs/zh/connectors/source/GraphQL.md
index 82595a0943..2b6a4a33b9 100644
--- a/docs/zh/connectors/source/GraphQL.md
+++ b/docs/zh/connectors/source/GraphQL.md
@@ -2,177 +2,158 @@ import ChangeLog from '../changelog/connector-graphql.md';
# GraphQL
-> GraphQL Source 连接器
+> GraphQL 源连接器
+
+## 支持引擎
+
+> Spark<br/>
+> Flink<br/>
+> SeaTunnel Zeta<br/>
## 描述
-用于读取GraphQL数据。
+GraphQL 源连接器用于从 GraphQL 服务读取数据。它支持：
+
+- 通过 HTTP 批量读取或轮询读取 GraphQL `query`。
+- 通过 WebSocket 读取 GraphQL `subscription`。
+- 使用 `content_field` 和 `schema.fields` 解析 JSON 响应。
+- 配置请求头、请求参数、GraphQL 变量和超时时间。
## 主要特性
- [x] [批处理](../../introduction/concepts/connector-v2-features.md)
- [x] [流处理](../../introduction/concepts/connector-v2-features.md)
-- [ ] [并行](../../introduction/concepts/connector-v2-features.md)
-
-## 源选项
-
-| 名称 | 类型 | 是否必填 | 默认值 |
-| --------------------------- | ------- | -------- | ----------------------- |
-| url | String | Yes | - |
-| query | String | Yes | - |
-| variables | Config | No | - |
-| enable_subscription | boolean | No | false |
-| timeout | Long | No | - |
-| content_field | String | Yes | $.data.{query_object}.* |
-| schema.fields | Config | Yes | - |
-| params | Map | Yes | - |
-| poll_interval_millis | int | No | - |
-| retry | int | No | - |
-| retry_backoff_multiplier_ms | int | No | 100 |
-| retry_backoff_max_ms | int | No | 10000 |
-| enable_multi_lines | boolean | No | false |
-| common-options | config | No | - |
-
-### url [String]
-
-http 请求路径。
-
-### query [String]
-
-GraphQL 表达式查询字符串
-
-### variables [String]
-
-GraphQL 变量
-
-比如
-
-```
-variables = {
- limit = 2
-}
-```
-
-### enable_subscription [boolean]
-
-1. true : 开启流式订阅模式（WebSocket）
-2. false : 开启批处理查询模式（HTTP）
-
-### timeout [Long]
-
-超时时间
-
-### content_field [String]
-
-SONPath通配符
-
-### params [Map]
-
-HTTP请求参数
-
-### poll_interval_millis [int]
-
-流模式下请求HTTP API间隔（毫秒）
+- [ ] [并行度](../../introduction/concepts/connector-v2-features.md)
-### retry [int]
+## 支持的数据源信息
-如果请求http返回到‘ IOException ’的最大重试次数
+使用 GraphQL 连接器需要安装下面的依赖。可以通过 install-plugin.sh 安装，也可以从 Maven 中央仓库下载。
-### retry_backoff_multiplier_ms [int]
+| 数据源 | 支持版本 | 依赖 |
+|---------|----------|------|
+| GraphQL | 通用 |
[下载](https://mvnrepository.com/artifact/org.apache.seatunnel/connector-graphql)
|
-如果请求http失败，则重试回退时间（毫秒）倍率
-
-### retry_backoff_max_ms [int]
-
-如果http请求失败，最大重试回退时间（毫秒）
-
-### schema [Config]
+## 源选项
-填写一个固定值
+| 名称 | 类型 | 是否必填 | 默认值 | 描述 |
+|---------------------|---------|----------|--------|------|
+| url | String | 是 | - | GraphQL 服务地址。查询模式使用
`http://` 或 `https://`；订阅模式使用 `ws://` 或 `wss://`。 |
+| query | String | 是 | - | GraphQL 语句。源连接器支持
`query`；只有设置 `enable_subscription = true` 时才支持 `subscription`。 |
+| variables | Map | 否 | - | 随请求体一起发送的 GraphQL 变量。 |
+| enable_subscription | Boolean | 否 | false | 是否使用 WebSocket 订阅模式。为
false 时使用 HTTP POST。 |
+| timeout | Long | 否 | - | 传给 HTTP 客户端参数的请求超时时间。 |
+| headers | Map | 否 | - | HTTP 请求头，例如鉴权请求头。 |
+| params | Map | 否 | - | HTTP 请求参数。 |
+| format | String | 否 | TEXT | 继承自 HTTP source 的响应格式。配置
`schema.fields` 读取结构化 JSON 时，通常设置为 `json`。 |
+| content_field | String | 否 | - | 从 GraphQL 响应中提取数据数组或对象的
JSONPath，例如 `$.data.source`。 |
+| schema.fields | Config | 否 | - | 输出字段和 SeaTunnel
数据类型。读取结构化 JSON 行时配置。 |
+| poll_interval_millis | Int | 否 | - | 流式查询模式下，两次 HTTP
请求之间的间隔，单位毫秒。 |
+| max_retries | Int | 否 | 5 | WebSocket 订阅模式下的最大重连次数。 |
+| retry_delay_ms | Int | 否 | 5000 | WebSocket
订阅模式下两次重连之间的等待时间，单位毫秒。 |
+| retry | Int | 否 | - | HTTP 请求出现 IOException
时的最大重试次数。 |
+| retry_backoff_multiplier_ms | Int | 否 | 100 | HTTP 请求失败后的重试退避倍率，单位毫秒。 |
+| retry_backoff_max_ms | Int | 否 | 10000 | HTTP 请求失败后的最大重试退避时间，单位毫秒。
|
+| enable_multi_lines | Boolean | 否 | false | 是否按行拆分 HTTP 响应后再解析。 |
+| connect_timeout_ms | Int | 否 | 12000 | HTTP 连接超时时间，单位毫秒。 |
+| socket_timeout_ms | Int | 否 | 60000 | HTTP socket 超时时间，单位毫秒。 |
+| common-options | Config | 否 | - | Source 通用参数，详见 [Source
Common Options](../common-options/source-common-options.md)。 |
+
+## 注意事项
+
+- 普通查询模式下，`url` 必须以 `http://` 或 `https://` 开头。
+- 订阅模式下，需要设置 `enable_subscription = true`，并使用 `ws://` 或 `wss://` 地址。
+- Source 不支持 GraphQL `mutation` 操作。
+- GraphQL 响应通常包在 `data` 字段下面，所以一般需要配置 `content_field`。
+
+## 任务示例
+
+### 查询 GraphQL 数据
```hocon
- schema = {
- fields {
- metric = "map<string, string>"
- value = double
- time = long
- }
- }
-
-```
-
-#### fields [Config]
-
-上游数据的模式字段
-
-### common options
-
-源插件常用参数，请参考 [Source Common
Options](../common-options/source-common-options.md) 获取详细信息
-
-## 示例
-
-### Query
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
-```hocon
source {
- GraphQL {
- url = "http://192.168.1.103:9081/v1/graphql";
- content_field = "$.data.source"
- query = """
- query MyQuery($limit: Int) {
- source(limit: $limit) {
- id
- val_bool
- val_double
- val_float
- }
- }
- """
- variables = {
- limit = 2
- }
- schema = {
- fields {
- id = "int"
- val_bool = "boolean"
- val_double = "double"
- val_float = "float"
- }
+ GraphQL {
+ plugin_output = "graphql_source"
+ url = "http://graphql:8080/v1/graphql";
+ format = "json"
+ content_field = "$.data.source"
+ query = """
+ query MyQuery($limit: Int) {
+ source(limit: $limit) {
+ id
+ val_bool
+ val_double
+ val_float
}
+ }
+ """
+ variables = {
+ limit = 2
}
+ schema = {
+ fields {
+ id = "int"
+ val_bool = "boolean"
+ val_double = "double"
+ val_float = "float"
+ }
+ }
+ }
+}
+
+sink {
+ Console {
+ plugin_input = "graphql_source"
+ }
}
```
-### Subscription
+### 订阅 GraphQL 数据
```hocon
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+}
+
source {
- GraphQL {
- url = "http://192.168.1.103:9081/v1/graphql";
- content_field = "$.data.source"
- query = """
- query MyQuery($limit: Int) {
- source(limit: $limit) {
- id
- val_bool
- val_double
- val_float
- }
- }
- """
- variables = {
- limit = 2
- }
- enable_subscription = true
- schema = {
- fields {
- id = "int"
- val_bool = "boolean"
- val_double = "double"
- val_float = "float"
- }
+ GraphQL {
+ plugin_output = "graphql_subscription"
+ url = "ws://graphql:8080/v1/graphql"
+ format = "json"
+ content_field = "$.data.source"
+ enable_subscription = true
+ max_retries = 5
+ retry_delay_ms = 5000
+ query = """
+ subscription MySubscription {
+ source {
+ id
+ val_bool
+ val_double
+ val_float
}
+ }
+ """
+ schema = {
+ fields {
+ id = "int"
+ val_bool = "boolean"
+ val_double = "double"
+ val_float = "float"
+ }
}
+ }
+}
+
+sink {
+ Console {
+ plugin_input = "graphql_subscription"
+ }
}
```
