This is an automated email from the ASF dual-hosted git repository.
corgy-w 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 6ec96ef22f [Docs][Connector-V2] Improve Pulsar TDengine HugeGraph
AmazonSqs and BigQuery connector docs (#11736)
6ec96ef22f is described below
commit 6ec96ef22f353324555095706bcd392cc3ac0f6f
Author: Daniel Carter <[email protected]>
AuthorDate: Sun Aug 16 13:06:20 2026 +0800
[Docs][Connector-V2] Improve Pulsar TDengine HugeGraph AmazonSqs and
BigQuery connector docs (#11736)
---
docs/en/connectors/sink/AmazonSqs.md | 29 ++++++++++
docs/en/connectors/sink/BigQuery.md | 43 ++++++++++++++
docs/en/connectors/sink/HugeGraph.md | 89 ++++++++++++++++++++++++++++
docs/en/connectors/sink/Pulsar.md | 50 ++++++++++++++++
docs/en/connectors/sink/TDengine.md | 75 +++++++++++++++++-------
docs/en/connectors/source/AmazonSqs.md | 49 ++++++++++++++--
docs/en/connectors/source/HugeGraph.md | 60 +++++++++++++++++++
docs/en/connectors/source/Pulsar.md | 88 ++++++++++++++++++++++++++++
docs/en/connectors/source/TDengine.md | 81 +++++++++++++++++---------
docs/zh/connectors/sink/AmazonSqs.md | 14 +++++
docs/zh/connectors/sink/BigQuery.md | 32 +++++++++++
docs/zh/connectors/sink/HugeGraph.md | 84 ++++++++++++++++++++++++++-
docs/zh/connectors/sink/Pulsar.md | 50 ++++++++++++++++
docs/zh/connectors/sink/TDengine.md | 60 ++++++++++++-------
docs/zh/connectors/source/AmazonSqs.md | 31 ++++++++++
docs/zh/connectors/source/HugeGraph.md | 61 +++++++++++++++++++-
docs/zh/connectors/source/Pulsar.md | 88 ++++++++++++++++++++++++++++
docs/zh/connectors/source/TDengine.md | 102 +++++++++++++++++++--------------
18 files changed, 965 insertions(+), 121 deletions(-)
diff --git a/docs/en/connectors/sink/AmazonSqs.md
b/docs/en/connectors/sink/AmazonSqs.md
index 3aa659fb64..24b5c335ca 100644
--- a/docs/en/connectors/sink/AmazonSqs.md
+++ b/docs/en/connectors/sink/AmazonSqs.md
@@ -48,6 +48,19 @@ is serialized by the configured `format`, and the serialized
value is sent as th
- `access_key_id` and `secret_access_key` are optional, but they must be
configured together when static AWS credentials are used.
- The sink sends each SeaTunnel row as one SQS message. It does not batch
multiple rows into one SQS request.
+## Authentication
+
+The connector resolves AWS credentials in the following order:
+
+1. `access_key_id` and `secret_access_key` if both are configured.
+2. Otherwise, the AWS default credential provider chain (environment
variables, instance profile, etc.).
+
+For local testing against an SQS-compatible service such as LocalStack or
ElasticMQ,
+point `url` at the local endpoint (for example `http://sqs-host:4566/...`) and
provide any
+non-empty `access_key_id` / `secret_access_key`. SQS-compatible test services
typically
+do not validate the SigV4 signature on incoming requests, so any static
credential pair is
+accepted by them.
+
## Task Examples
### Copy Messages Between Local-Compatible Queues
@@ -142,6 +155,22 @@ sink {
}
```
+### Write Canal JSON Messages
+
+Set `format = canal_json` so each SeaTunnel row is serialized as a Canal JSON
+change event. Useful when the downstream consumer is a Canal JSON consumer
(such
+as a Canal → Kafka bridge or a Canal-compatible BigQuery loader).
+
+```hocon
+sink {
+ AmazonSqs {
+ url = "https://sqs.us-east-1.amazonaws.com/123456789012/sink_queue";
+ region = "us-east-1"
+ format = canal_json
+ }
+}
+```
+
## Changelog
<ChangeLog />
diff --git a/docs/en/connectors/sink/BigQuery.md
b/docs/en/connectors/sink/BigQuery.md
index e1ab3a026e..801d05cff5 100644
--- a/docs/en/connectors/sink/BigQuery.md
+++ b/docs/en/connectors/sink/BigQuery.md
@@ -87,6 +87,10 @@ If `sequence_number_column` is not configured,
`_CHANGE_SEQUENCE_NUMBER` is not
### Simple Batch Example
+This example shows a local end-to-end batch test against the BigQuery emulator.
+Production jobs should provide a real service-account key (or rely on default
+application credentials) and target a real GCP project.
+
```hocon
env {
parallelism = 1
@@ -175,6 +179,27 @@ sink {
}
```
+When the upstream CDC source already produces a monotonically increasing column
+(such as an `updated_at` epoch millis or a row version), wire it to
+`sequence_number_column` so BigQuery can dedup retried batches. The target
+table must define a primary key (the example above uses `PRIMARY KEY (uuid)
+NOT ENFORCED`), otherwise BigQuery treats every write as an append and skips
+deduplication.
+
+```hocon
+sink {
+ BigQuery {
+ project_id = "my-gcp-project"
+ dataset_id = "cdc_dataset"
+ table_id = "orders"
+ service_account_key_path = "/path/to/key.json"
+ write_mode = "streaming"
+ sequence_number_column = "updated_at"
+ batch_size = 500
+ }
+}
+```
+
### Complex Data Types Example
```hocon
@@ -207,6 +232,24 @@ sink {
}
```
+### Inline Service Account Key
+
+For environments where a key file is inconvenient (CI runners, Kubernetes
+secrets mounted as env vars), inline the JSON content with
+`service_account_key_json`.
+
+```hocon
+sink {
+ BigQuery {
+ project_id = "my-gcp-project"
+ dataset_id = "orders"
+ table_id = "customer_orders"
+ service_account_key_json = "${GCP_SA_KEY_JSON}"
+ batch_size = 500
+ }
+}
+```
+
### Testing
This connector uses the BigQuery Storage Write API. The current local BigQuery
emulator does not fully support the write path used by this connector.
diff --git a/docs/en/connectors/sink/HugeGraph.md
b/docs/en/connectors/sink/HugeGraph.md
index e8ac7a39c1..9866733e91 100644
--- a/docs/en/connectors/sink/HugeGraph.md
+++ b/docs/en/connectors/sink/HugeGraph.md
@@ -4,6 +4,12 @@ import ChangeLog from '../changelog/connector-hugegraph.md';
`Sink: HugeGraph`
+## Support Those Engines
+
+> Spark<br/>
+> Flink<br/>
+> SeaTunnel Zeta<br/>
+
## Description
The HugeGraph sink connector allows you to write data from SeaTunnel to Apache
HugeGraph, a fast and scalable graph database.
@@ -282,6 +288,89 @@ sink {
}
```
+### 3. Writing DELETE rows
+
+The sink honours the row kind. `DELETE` rows only need to carry the columns
+required to rebuild the element id; other columns can be omitted. With
+`delete_vertex_with_edges = true`, deleting a vertex also deletes its incident
+edges.
+
+```hocon
+source {
+ FakeSource {
+ schema = {
+ fields = {
+ name = "string"
+ }
+ }
+ rows = [
+ {
+ kind = DELETE
+ fields = ["bob"]
+ }
+ ]
+ }
+}
+
+sink {
+ HugeGraph {
+ host = "localhost"
+ port = 8080
+ graph_name = "hugegraph"
+ delete_vertex_with_edges = true
+ mappings = [
+ {
+ type = "VERTEX"
+ label = "person"
+ idStrategy = "PRIMARY_KEY"
+ idFields = ["name"]
+ }
+ ]
+ }
+}
+```
+
+### 4. Cloning from a HugeGraph Source
+
+When the upstream is the HugeGraph Source, each row already carries the
+reserved columns with the assembled element ids (`~id` for a vertex;
+`~source_id` / `~target_id` for an edge). Reusing these ids preserves the
+original graph exactly. The example below sets `multi_table_sink_replica` so
the
+sink can fan out across parallel writers when the source reads multiple labels.
+
+```hocon
+env {
+ job.mode = "BATCH"
+}
+
+source {
+ HugeGraph {
+ host = "src-host"
+ port = 8080
+ graph_name = "hugegraph"
+ label_type = "VERTEX"
+ }
+}
+
+sink {
+ HugeGraph {
+ host = "dst-host"
+ port = 8080
+ graph_name = "hugegraph"
+ multi_table_sink_replica = 2
+ batch_size = 500
+ mappings = [
+ {
+ type = "VERTEX"
+ label = "person"
+ idStrategy = "CUSTOMIZE_STRING"
+ idFields = ["~id"]
+ }
+ ]
+ }
+}
+```
+
## Changelog
<ChangeLog />
diff --git a/docs/en/connectors/sink/Pulsar.md
b/docs/en/connectors/sink/Pulsar.md
index af97ca3275..80ea4e9c2a 100644
--- a/docs/en/connectors/sink/Pulsar.md
+++ b/docs/en/connectors/sink/Pulsar.md
@@ -213,6 +213,56 @@ sink {
}
```
+### Write Text Messages With a Custom Delimiter
+
+Set `format = text` and configure `field_delimiter` to serialize each row into
a delimited text payload. Useful when the downstream consumer expects a simple
flat format.
+
+```hocon
+sink {
+ Pulsar {
+ topic = "text_events"
+ client.service-url = "pulsar://localhost:6650"
+ admin.service-url = "http://localhost:8080";
+ format = text
+ field_delimiter = "|"
+ }
+}
+```
+
+### Write Avro Messages
+
+Set `format = avro`. The connector derives the Avro schema from the upstream
row type, so no sink-side `schema` option is required.
+
+```hocon
+sink {
+ Pulsar {
+ topic = "test_avro_topic_fake_source"
+ client.service-url = "pulsar://localhost:6650"
+ admin.service-url = "http://localhost:8080";
+ format = avro
+ }
+}
+```
+
+### Write With Pulsar Producer Properties
+
+Pass extra producer properties via `pulsar.config`. This is forwarded to the
Pulsar producer client and can be used for tuning (timeouts, batching,
compression, etc.).
+
+```hocon
+sink {
+ Pulsar {
+ topic = "topic_test"
+ client.service-url = "pulsar://localhost:6650"
+ admin.service-url = "http://localhost:8080";
+ format = json
+ pulsar.config = {
+ sendTimeoutMs = 30000
+ batchingMaxMessages = 1000
+ }
+ }
+}
+```
+
## Changelog
<ChangeLog />
diff --git a/docs/en/connectors/sink/TDengine.md
b/docs/en/connectors/sink/TDengine.md
index e0bd099621..cf32eac44a 100644
--- a/docs/en/connectors/sink/TDengine.md
+++ b/docs/en/connectors/sink/TDengine.md
@@ -4,6 +4,12 @@ import ChangeLog from '../changelog/connector-tdengine.md';
> TDengine sink connector
+## Support Those Engines
+
+> Spark<br/>
+> Flink<br/>
+> SeaTunnel Zeta<br/>
+
## Description
Write data to TDengine.
@@ -30,50 +36,55 @@ sub table name.
## Options
-| name | type | required | default value |
-|--------------|--------|----------|---------------|
-| url | string | yes | - |
-| username | string | yes | - |
-| password | string | yes | - |
-| database | string | yes | - |
-| stable | string | yes | - |
-| timezone | string | no | UTC |
-| write_columns | list | no | - |
-| common-options | | no | - |
+| Name | Type | Required | Default | Description |
+|---------------|--------|----------|---------|-------------|
+| url | String | Yes | - | TDengine REST JDBC URL. For
example `jdbc:TAOS-RS://localhost:6041/`. |
+| username | String | Yes | - | Username used to connect to
TDengine. |
+| password | String | Yes | - | Password used to connect to
TDengine. |
+| database | String | Yes | - | TDengine database name. |
+| stable | String | Yes | - | TDengine super table name. For
multi-table writes, this value can contain placeholders such as
`${table_name}`. |
+| timezone | String | No | UTC | TDengine server timezone used
for timestamp conversion. |
+| write_columns | List | No | - | Normal TDengine column names
to insert. When unset, the target super table column order is used. Do not
include TAGS columns or the sub-table-name field. |
+| common-options | | No | - | Sink plugin common
parameters. See [Sink Common
Options](../common-options/sink-common-options.md). |
-### url [string]
+### url [String]
The TDengine REST JDBC URL.
-e.g.
+For example:
```
jdbc:TAOS-RS://localhost:6041/
```
-### username [string]
+### username [String]
The username used to connect to TDengine.
-### password [string]
+### password [String]
The password used to connect to TDengine.
-### database [string]
+### database [String]
-The TDengine database name.
+The TDengine database name. The database must already exist on the server.
-### stable [string]
+### stable [String]
-The TDengine super table name. For multi-table writes, this value can contain
-placeholders, for example `${table_name}`.
+The TDengine super table name. The value is used verbatim by the sink writer;
+the TDengine connector itself does not perform placeholder substitution, so
+`${table_name}` and similar tokens are not rewritten at runtime. For
multi-table
+writes, SeaTunnel's upstream framework (`TablePlaceholderProcessor`) may
+substitute `stable` once during job setup based on the upstream `CatalogTable`
+identifier, but this depends on the upstream framework wiring and is not a
+TDengine-specific feature.
-### timezone [string]
+### timezone [String]
The TDengine server timezone used for timestamp conversion. The default value
is
-`UTC`.
+`UTC`. Set this to match your TDengine server if it is not running in UTC.
-### write_columns [list]
+### write_columns [List]
The normal TDengine column names to insert. If it is not set, TDengine uses the
column order of the target super table. Do not include the first input field
@@ -87,6 +98,20 @@ Sink plugin common parameters, please refer to
For multi-table writes, `multi_table_sink_replica` can be used with the common
sink options.
+## Input Row Shape
+
+The connector expects every input row to follow the super-table write shape:
+
+1. The first field is the target sub table name (string). The sink creates the
+ sub table on demand if it does not exist.
+2. The next fields are the normal columns declared in `write_columns` (or in
the
+ target super table column order when `write_columns` is unset).
+3. The last fields are TAGS values. The number of TAGS fields is read from the
+ target super table metadata.
+
+If the target super table has two TAGS fields, the last two input fields are
+TAGS values and the first input field is the sub table name.
+
## Examples
### Write to one super table
@@ -177,6 +202,12 @@ sink {
}
```
+Here `${table_name}` is treated as a literal string by the TDengine sink writer
+(the connector does not substitute it per row), so this example only works if
+the upstream framework substitutes `stable` once at job setup with the
+`CatalogTable` identifier from the upstream source. The target super tables
+must already exist with matching TAGS columns.
+
## Changelog
<ChangeLog />
diff --git a/docs/en/connectors/source/AmazonSqs.md
b/docs/en/connectors/source/AmazonSqs.md
index b44775f11d..f571cb4551 100644
--- a/docs/en/connectors/source/AmazonSqs.md
+++ b/docs/en/connectors/source/AmazonSqs.md
@@ -4,6 +4,12 @@ import ChangeLog from '../changelog/connector-amazonsqs.md';
> Amazon SQS source connector
+## Support Those Engines
+
+> Spark<br/>
+> Flink<br/>
+> SeaTunnel Zeta<br/>
+
## Description
The Amazon SQS source connector reads messages from one Amazon SQS queue URL.
Each message body is
@@ -14,12 +20,6 @@ processed. It is suitable for bounded reads from a queue.
Each receive request asks SQS for up to 10 messages. If more messages are
waiting in the queue, run another job or use a streaming-style upstream design
that repeatedly starts bounded reads.
-## Support Those Engines
-
-> Spark<br/>
-> Flink<br/>
-> SeaTunnel Zeta<br/>
-
## Key Features
- [x] [batch](../../introduction/concepts/connector-v2-features.md)
@@ -57,6 +57,19 @@ Each receive request asks SQS for up to 10 messages. If more
messages are waitin
- `access_key_id` and `secret_access_key` are optional, but they must be
configured together when static AWS credentials are used.
- The source performs one receive request, with up to 10 messages, and then
finishes the bounded job.
+## Authentication
+
+The connector resolves AWS credentials in the following order:
+
+1. `access_key_id` and `secret_access_key` if both are configured.
+2. Otherwise, the AWS default credential provider chain (environment
variables, instance profile, etc.).
+
+For local testing against an SQS-compatible service such as LocalStack or
ElasticMQ,
+point `url` at the local endpoint (for example `http://sqs-host:4566/...`) and
provide any
+non-empty `access_key_id` / `secret_access_key`. SQS-compatible test services
typically
+do not validate the SigV4 signature on incoming requests, so any static
credential pair is
+accepted by them.
+
## Task Examples
### Copy Messages Between Local-Compatible Queues
@@ -143,6 +156,30 @@ sink {
}
```
+### Read Debezium JSON Messages
+
+When the upstream system (such as Debezium or a CDC source) publishes change
+events as Debezium envelopes, set `format = debezium_json` and use
+`debezium_record_include_schema` to control whether the schema field is
expected.
+
+```hocon
+source {
+ AmazonSqs {
+ url = "https://sqs.us-east-1.amazonaws.com/123456789012/cdc_events";
+ region = "us-east-1"
+ format = debezium_json
+ debezium_record_include_schema = true
+ schema = {
+ fields {
+ id = bigint
+ name = string
+ score = double
+ }
+ }
+ }
+}
+```
+
## Changelog
<ChangeLog />
diff --git a/docs/en/connectors/source/HugeGraph.md
b/docs/en/connectors/source/HugeGraph.md
index 2059cf2027..7f76a037bf 100644
--- a/docs/en/connectors/source/HugeGraph.md
+++ b/docs/en/connectors/source/HugeGraph.md
@@ -4,6 +4,12 @@ import ChangeLog from '../changelog/connector-hugegraph.md';
`Source: HugeGraph`
+## Support Those Engines
+
+> Spark<br/>
+> Flink<br/>
+> SeaTunnel Zeta<br/>
+
## Description
The HugeGraph source connector reads graph data from Apache HugeGraph through
the HugeGraph REST API.
@@ -88,6 +94,8 @@ Notes:
## Example
+### Read a vertex label with explicit schema
+
```hocon
source {
HugeGraph {
@@ -107,6 +115,58 @@ source {
}
```
+### Read an edge label
+
+To read edges instead of vertices, set `label_type = "EDGE"` and list the edge
+properties in `schema.fields`. The output also includes the reserved columns
+`~source_id`, `~source_label`, `~target_id`, `~target_label`.
+
+```hocon
+source {
+ HugeGraph {
+ host = "localhost"
+ port = 8080
+ graph_name = "hugegraph"
+ label = "knows"
+ label_type = "EDGE"
+ schema = {
+ fields = {
+ since = "int"
+ }
+ }
+ }
+}
+```
+
+### Read with a server-side property filter
+
+`filter` requires `parallelism = 1` (the job fails fast if `filter` is combined
+with `parallelism > 1`); with that setting, configure `filter` to push a
+property-equality condition to the server. Only elements whose properties match
+every entry are returned.
+
+```hocon
+source {
+ HugeGraph {
+ host = "localhost"
+ port = 8080
+ graph_name = "hugegraph"
+ label = "person"
+ label_type = "VERTEX"
+ filter = {
+ country = "US"
+ active = "true"
+ }
+ schema = {
+ fields = {
+ name = "string"
+ age = "int"
+ }
+ }
+ }
+}
+```
+
## Schema auto-discovery
`schema` is optional. When omitted, the connector connects to the server at
job build time, reads the definition of `label`, and produces one output column
per property key (types from the [Type Mapping](#type-mapping) table,
`LIST`/`SET` as `array<T>`), ordered by property name. This is convenient for a
full-label dump when you do not want to hand-declare every field.
diff --git a/docs/en/connectors/source/Pulsar.md
b/docs/en/connectors/source/Pulsar.md
index 90bb263765..be83ebcbe8 100644
--- a/docs/en/connectors/source/Pulsar.md
+++ b/docs/en/connectors/source/Pulsar.md
@@ -273,6 +273,94 @@ source {
For batch jobs, use a bounded stop mode such as `LATEST` or `TIMESTAMP`. Use
`cursor.stop.mode = "NEVER"` for streaming jobs.
+### Read Avro Messages
+
+When the topic carries Avro-encoded records, set `format = avro` and declare
the field types in `schema`. The connector decodes the Avro payload using the
SeaTunnel type system.
+
+```hocon
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ Pulsar {
+ topic = "test_avro_topic"
+ subscription.name = "seatunnel-avro"
+ client.service-url = "pulsar://localhost:6650"
+ admin.service-url = "http://localhost:8080";
+ cursor.startup.mode = "EARLIEST"
+ cursor.stop.mode = "LATEST"
+ format = avro
+ schema = {
+ fields {
+ id = bigint
+ c_string = string
+ c_int = int
+ c_double = double
+ c_timestamp = timestamp
+ }
+ }
+ }
+}
+```
+
+### Streaming Read From a Topic
+
+Use `cursor.stop.mode = "NEVER"` to keep reading new messages until the job is
stopped. Pair it with `cursor.startup.mode = "LATEST"` to start from the latest
message and avoid replaying history.
+
+```hocon
+env {
+ parallelism = 2
+ job.mode = "STREAMING"
+ checkpoint.interval = 10000
+}
+
+source {
+ Pulsar {
+ topic = "persistent://public/default/events"
+ subscription.name = "seatunnel-stream"
+ client.service-url = "pulsar://localhost:6650"
+ admin.service-url = "http://localhost:8080";
+ cursor.startup.mode = "LATEST"
+ cursor.stop.mode = "NEVER"
+ format = json
+ schema = {
+ fields {
+ event_id = string
+ user_id = bigint
+ payload = string
+ }
+ }
+ }
+}
+```
+
+### Read From a Topic Pattern With Discovery
+
+When the topic list grows over time, combine `topic-pattern` with
`topic-discovery.interval` to pick up newly created topics automatically.
+
+```hocon
+source {
+ Pulsar {
+ topic-pattern = "persistent://public/default/orders-.*"
+ subscription.name = "seatunnel-orders"
+ client.service-url = "pulsar://localhost:6650"
+ admin.service-url = "http://localhost:8080";
+ topic-discovery.interval = 30000
+ cursor.startup.mode = "EARLIEST"
+ cursor.stop.mode = "LATEST"
+ format = json
+ schema = {
+ fields {
+ order_id = bigint
+ amount = double
+ }
+ }
+ }
+}
+```
+
## Changelog
<ChangeLog />
diff --git a/docs/en/connectors/source/TDengine.md
b/docs/en/connectors/source/TDengine.md
index cccdad9848..f01d8f394e 100644
--- a/docs/en/connectors/source/TDengine.md
+++ b/docs/en/connectors/source/TDengine.md
@@ -4,6 +4,12 @@ import ChangeLog from '../changelog/connector-tdengine.md';
> TDengine source connector
+## Support Those Engines
+
+> Spark<br/>
+> Flink<br/>
+> SeaTunnel Zeta<br/>
+
## Description
Read data from TDengine super tables.
@@ -23,69 +29,70 @@ TDengine sink when writing rows back to TDengine.
- [ ] [stream](../../introduction/concepts/connector-v2-features.md)
- [x] [exactly-once](../../introduction/concepts/connector-v2-features.md)
- [x] [column projection](../../introduction/concepts/connector-v2-features.md)
-
- [x] [parallelism](../../introduction/concepts/connector-v2-features.md)
- [ ] [support user-defined
split](../../introduction/concepts/connector-v2-features.md)
## Options
-| name | type | required | default value |
-|--------------|--------|----------|---------------|
-| url | string | yes | - |
-| username | string | yes | - |
-| password | string | yes | - |
-| database | string | yes | - |
-| stable | string | yes | - |
-| sub_tables | list | no | - |
-| lower_bound | string | yes | - |
-| upper_bound | string | yes | - |
-| read_columns | list | no | - |
-
-### url [string]
+| Name | Type | Required | Default | Description |
+|--------------|--------|----------|---------|-------------|
+| url | String | Yes | - | TDengine REST JDBC URL. For
example `jdbc:TAOS-RS://localhost:6041/`. |
+| username | String | Yes | - | Username used to connect to
TDengine. |
+| password | String | Yes | - | Password used to connect to
TDengine. |
+| database | String | Yes | - | TDengine database name. |
+| stable | String | Yes | - | TDengine super table name. |
+| sub_tables | List | No | - | Sub table names to read. When
unset, all sub tables under `stable` are read. |
+| lower_bound | String | Yes | - | Inclusive lower bound of the
query time range. Use a TDengine-compatible timestamp such as `2018-10-03
14:38:05.000`. |
+| upper_bound | String | Yes | - | Exclusive upper bound of the
query time range. Use a TDengine-compatible timestamp such as `2018-10-03
14:38:16.801`. |
+| read_columns | List | No | - | Columns to read. When unset,
all columns are read. Put TAGS columns at the end of the list; do not include
`subtable_name`. |
+| common-options | | No | - | Source plugin common
parameters. See [Source Common
Options](../common-options/source-common-options.md). |
+
+### url [String]
The TDengine REST JDBC URL.
-e.g.
+For example:
```
jdbc:TAOS-RS://localhost:6041/
```
-### username [string]
+### username [String]
The username used to connect to TDengine.
-### password [string]
+### password [String]
The password used to connect to TDengine.
-### database [string]
+### database [String]
-The TDengine database name.
+The TDengine database name. The database must already exist on the server.
-### stable [string]
+### stable [String]
-The TDengine super table name.
+The TDengine super table name. The connector splits the read into one source
+split per sub table under this super table.
-### sub_tables [list]
+### sub_tables [List]
A list of sub table names. If it is not configured, all sub tables under the
configured super table are read. If it is configured, only the listed sub
tables
-are read.
+are read. Sub table names must match the server exactly.
-### lower_bound [string]
+### lower_bound [String]
The inclusive lower bound of the query time range. The connector adds
`timestamp_column >= lower_bound` to each sub table query. Use a
TDengine-compatible timestamp string, for example `2018-10-03 14:38:05.000`.
-### upper_bound [string]
+### upper_bound [String]
The exclusive upper bound of the query time range. The connector adds
`timestamp_column < upper_bound` to each sub table query. Use a
TDengine-compatible timestamp string, for example `2018-10-03 14:38:16.801`.
-### read_columns [list]
+### read_columns [List]
A list of column names to read. If it is not configured, all columns are read.
When reading from a super table, put TAGS columns at the end of the list. Do
not
@@ -97,6 +104,19 @@ The order of `read_columns` decides the output field order
after
before TAGS columns so the sink can split column values and TAGS values
correctly.
+### common options
+
+Source plugin common parameters, please refer to
+[Source Common Options](../common-options/source-common-options.md) for
details.
+
+## Output Schema
+
+The output table always starts with the reserved `subtable_name` column, which
+carries the TDengine sub table name each row came from. Subsequent columns are
+the columns declared in `read_columns` (or all columns when `read_columns` is
+unset), in the order specified by `read_columns`. TAGS columns follow the
+normal columns in the same order you list them.
+
## Examples
### Read all sub tables in a time range
@@ -139,6 +159,11 @@ source {
}
```
+This reads only sub tables `d1001` and `d1002` under super table `meters`. The
+TAGS columns (`location`, `groupid`) are placed at the end of `read_columns` so
+that a downstream TDengine sink can correctly split normal columns and TAGS
+values.
+
### Read from TDengine and write to TDengine
```hocon
@@ -172,6 +197,10 @@ sink {
}
```
+The sink relies on `subtable_name` from the source as the target sub table name
+when writing back. The number of TAGS values is read from the target super
+table metadata, so the target super table must already exist.
+
## Changelog
<ChangeLog />
diff --git a/docs/zh/connectors/sink/AmazonSqs.md
b/docs/zh/connectors/sink/AmazonSqs.md
index c76450acc9..9a3be28722 100644
--- a/docs/zh/connectors/sink/AmazonSqs.md
+++ b/docs/zh/connectors/sink/AmazonSqs.md
@@ -142,6 +142,20 @@ sink {
}
```
+### 写入 Canal JSON 消息
+
+将 `format` 设为 `canal_json`，每行 SeaTunnel 数据会被序列化为 Canal JSON 变更事件。下游是 Canal
JSON 消费者（如 Canal → Kafka 桥接或兼容 Canal 的 BigQuery 加载器）时非常有用。
+
+```hocon
+sink {
+ AmazonSqs {
+ url = "https://sqs.us-east-1.amazonaws.com/123456789012/sink_queue";
+ region = "us-east-1"
+ format = canal_json
+ }
+}
+```
+
## 变更日志
<ChangeLog />
diff --git a/docs/zh/connectors/sink/BigQuery.md
b/docs/zh/connectors/sink/BigQuery.md
index 23f161ada5..63e248e2e0 100644
--- a/docs/zh/connectors/sink/BigQuery.md
+++ b/docs/zh/connectors/sink/BigQuery.md
@@ -174,6 +174,22 @@ sink {
}
```
+如果上游 CDC 源能产生单调递增的列（例如 `updated_at` 毫秒时间戳或行版本号），可以把它配到
`sequence_number_column`，让 BigQuery 端对重试批次做去重。目标表必须定义主键（上例使用 `PRIMARY KEY
(uuid) NOT ENFORCED`），否则 BigQuery 会把每次写入都当作 append，跳过去重。
+
+```hocon
+sink {
+ BigQuery {
+ project_id = "my-gcp-project"
+ dataset_id = "cdc_dataset"
+ table_id = "orders"
+ service_account_key_path = "/path/to/key.json"
+ write_mode = "streaming"
+ sequence_number_column = "updated_at"
+ batch_size = 500
+ }
+}
+```
+
### 复杂数据类型示例
```hocon
@@ -206,6 +222,22 @@ sink {
}
```
+### 内联服务账号密钥
+
+如果不便挂载密钥文件（例如 CI runner、把密钥放在 Kubernetes Secret 中以环境变量形式注入），可以直接把 JSON 内容放到
`service_account_key_json` 中。
+
+```hocon
+sink {
+ BigQuery {
+ project_id = "my-gcp-project"
+ dataset_id = "orders"
+ table_id = "customer_orders"
+ service_account_key_json = "${GCP_SA_KEY_JSON}"
+ batch_size = 500
+ }
+}
+```
+
### 测试
该连接器使用 BigQuery Storage Write API。当前本地 BigQuery emulator 不能完整支持该连接器使用的写入路径。
diff --git a/docs/zh/connectors/sink/HugeGraph.md
b/docs/zh/connectors/sink/HugeGraph.md
index 7f5e94c22c..d47f748159 100644
--- a/docs/zh/connectors/sink/HugeGraph.md
+++ b/docs/zh/connectors/sink/HugeGraph.md
@@ -4,6 +4,12 @@ import ChangeLog from '../changelog/connector-hugegraph.md';
`Sink: HugeGraph`
+## 支持的引擎
+
+> Spark<br/>
+> Flink<br/>
+> SeaTunnel Zeta<br/>
+
## 描述
HugeGraph sink连接器允许您将数据从SeaTunnel写入Apache HugeGraph，这是一个快速且可扩展的图数据库。
@@ -14,7 +20,7 @@ HugeGraph sink连接器允许您将数据从SeaTunnel写入Apache HugeGraph，
- [x] [批处理](../../introduction/concepts/connector-v2-features.md)
- [ ] [精确一次](../../introduction/concepts/connector-v2-features.md)
-- [ ] [CDC](../../introduction/concepts/connector-v2-features.md)
+- [ ] [变更数据捕获](../../introduction/concepts/connector-v2-features.md)
- [x] [支持多表写入](../../introduction/concepts/connector-v2-features.md)
- [x] [定时刷新](../../introduction/concepts/connector-v2-features.md)
@@ -282,6 +288,82 @@ sink {
}
```
+### 3. 写入 DELETE 行
+
+Sink 会按行 kind 处理数据。`DELETE` 行只需要提供重建元素 id 所需的字段，其他列可以省略。把
`delete_vertex_with_edges` 设为 `true` 后，删除顶点时会同时删除其相连的边。
+
+```hocon
+source {
+ FakeSource {
+ schema = {
+ fields = {
+ name = "string"
+ }
+ }
+ rows = [
+ {
+ kind = DELETE
+ fields = ["bob"]
+ }
+ ]
+ }
+}
+
+sink {
+ HugeGraph {
+ host = "localhost"
+ port = 8080
+ graph_name = "hugegraph"
+ delete_vertex_with_edges = true
+ mappings = [
+ {
+ type = "VERTEX"
+ label = "person"
+ idStrategy = "PRIMARY_KEY"
+ idFields = ["name"]
+ }
+ ]
+ }
+}
+```
+
+### 4. 从 HugeGraph Source 整体克隆
+
+当上游是 HugeGraph Source 时，每行已经带有预留列（顶点为 `~id`，边端点为 `~source_id` /
`~target_id`），直接复用这些 id 即可完整还原图。下面的示例把 `multi_table_sink_replica` 调大，让 sink 在
source 读取多 label 时能并行写出。
+
+```hocon
+env {
+ job.mode = "BATCH"
+}
+
+source {
+ HugeGraph {
+ host = "src-host"
+ port = 8080
+ graph_name = "hugegraph"
+ label_type = "VERTEX"
+ }
+}
+
+sink {
+ HugeGraph {
+ host = "dst-host"
+ port = 8080
+ graph_name = "hugegraph"
+ multi_table_sink_replica = 2
+ batch_size = 500
+ mappings = [
+ {
+ type = "VERTEX"
+ label = "person"
+ idStrategy = "CUSTOMIZE_STRING"
+ idFields = ["~id"]
+ }
+ ]
+ }
+}
+```
+
## Changelog
<ChangeLog />
diff --git a/docs/zh/connectors/sink/Pulsar.md
b/docs/zh/connectors/sink/Pulsar.md
index baee0e785c..78857c0fd5 100644
--- a/docs/zh/connectors/sink/Pulsar.md
+++ b/docs/zh/connectors/sink/Pulsar.md
@@ -212,6 +212,56 @@ sink {
}
```
+### 使用自定义分隔符写入文本消息
+
+将 `format` 设置为 `text`，并通过 `field_delimiter`
指定分隔符，把每行序列化成定界文本。下游消费者需要简单扁平格式时可以使用这种方式。
+
+```hocon
+sink {
+ Pulsar {
+ topic = "text_events"
+ client.service-url = "pulsar://localhost:6650"
+ admin.service-url = "http://localhost:8080";
+ format = text
+ field_delimiter = "|"
+ }
+}
+```
+
+### 写入 Avro 消息
+
+将 `format` 设置为 `avro` 即可。Avro schema 由上游行类型推导生成，不需要在 Sink 端额外配置 `schema`。
+
+```hocon
+sink {
+ Pulsar {
+ topic = "test_avro_topic_fake_source"
+ client.service-url = "pulsar://localhost:6650"
+ admin.service-url = "http://localhost:8080";
+ format = avro
+ }
+}
+```
+
+### 自定义 Pulsar Producer 属性
+
+通过 `pulsar.config` 传入额外的 producer 属性，这些配置会透传给 Pulsar producer
客户端，可以用来调整超时、批大小、压缩等参数。
+
+```hocon
+sink {
+ Pulsar {
+ topic = "topic_test"
+ client.service-url = "pulsar://localhost:6650"
+ admin.service-url = "http://localhost:8080";
+ format = json
+ pulsar.config = {
+ sendTimeoutMs = 30000
+ batchingMaxMessages = 1000
+ }
+ }
+}
+```
+
## 变更日志
<ChangeLog />
diff --git a/docs/zh/connectors/sink/TDengine.md
b/docs/zh/connectors/sink/TDengine.md
index b87bd37799..74e2f126a6 100644
--- a/docs/zh/connectors/sink/TDengine.md
+++ b/docs/zh/connectors/sink/TDengine.md
@@ -4,6 +4,12 @@ import ChangeLog from '../changelog/connector-tdengine.md';
> TDengine 数据接收器
+## 支持的引擎
+
+> Spark<br/>
+> Flink<br/>
+> SeaTunnel Zeta<br/>
+
## 描述
用于将数据写入 TDengine。
@@ -23,48 +29,48 @@ import ChangeLog from '../changelog/connector-tdengine.md';
## 选项
-| 名称 | 类型 | 是否必传 | 默认值 |
-|----------------|--------|----------|--------|
-| url | string | 是 | - |
-| username | string | 是 | - |
-| password | string | 是 | - |
-| database | string | 是 | - |
-| stable | string | 是 | - |
-| timezone | string | 否 | UTC |
-| write_columns | list | 否 | - |
-| common-options | | 否 | - |
+| 名称 | 类型 | 是否必传 | 默认值 | 说明 |
+|----------------|--------|----------|--------|------|
+| url | String | 是 | - | TDengine REST JDBC 连接地址，例如
`jdbc:TAOS-RS://localhost:6041/`。 |
+| username | String | 是 | - | 连接 TDengine 使用的用户名。 |
+| password | String | 是 | - | 连接 TDengine 使用的密码。 |
+| database | String | 是 | - | TDengine 数据库名称。 |
+| stable | String | 是 | - | TDengine 超级表名称。多表写入时可以使用占位符，例如
`${table_name}`。 |
+| timezone | String | 否 | UTC | TDengine 服务端时区，用于时间戳转换。 |
+| write_columns | List | 否 | - | 要写入 TDengine
的普通列名列表。不配置时按目标超级表的列顺序写入；不要包含子表名列或 TAGS 字段。 |
+| common-options | | 否 | - | Sink 插件通用参数，请参考 [Sink Common
Options](../common-options/sink-common-options.md)。 |
-### url [string]
+### url [String]
TDengine REST JDBC 连接地址。
-例如
+例如：
```
jdbc:TAOS-RS://localhost:6041/
```
-### username [string]
+### username [String]
连接 TDengine 使用的用户名。
-### password [string]
+### password [String]
连接 TDengine 使用的密码。
-### database [string]
+### database [String]
-TDengine 数据库名称。
+TDengine 数据库名称，数据库必须已经在服务端存在。
-### stable [string]
+### stable [String]
-TDengine 超级表名称。多表写入时可以使用占位符，例如 `${table_name}`。
+TDengine 超级表名称。该值会被 Sink Writer 原样使用，TDengine 连接器本身不会执行占位符替换，因此
`${table_name}` 等占位符不会在运行时被替换。在多表写入场景下，上游 SeaTunnel
框架（`TablePlaceholderProcessor`）可能在任务初始化阶段根据上游 `CatalogTable` 的标识替换一次
`stable`，但这取决于上游框架的接线，并不是 TDengine 连接器自身的能力。
-### timezone [string]
+### timezone [String]
-TDengine 服务端时区，用于时间戳转换，默认值为 `UTC`。
+TDengine 服务端时区，用于时间戳转换，默认值为 `UTC`。如果服务端不是 UTC 时区，请把该项设置为与服务端一致的时区。
-### write_columns [list]
+### write_columns [List]
要写入 TDengine 的普通列名列表。不配置时，TDengine 会按目标超级表的列顺序写入。这里不要包含第一列子表名，也不要包含 TAGS
字段；连接器会自动从输入数据末尾取出 TAGS 值。
@@ -73,6 +79,16 @@ TDengine 服务端时区，用于时间戳转换，默认值为 `UTC`。
Sink 插件通用参数，请参考 [Sink Common
Options](../common-options/sink-common-options.md)。
多表写入时，可以配合通用参数中的 `multi_table_sink_replica` 使用。
+## 输入数据格式
+
+连接器要求每行输入数据符合超级表写入结构：
+
+1. 第一列为目标子表名（字符串）。若该子表不存在，Sink 会按目标超级表的 schema 自动创建。
+2. 接下来的列为 `write_columns` 中声明的普通列（未配置时使用目标超级表的列顺序）。
+3. 末尾几列为 TAGS 值。TAGS 字段的数量会从目标超级表的元数据中读取。
+
+例如，目标超级表有 2 个 TAGS 字段时，输入行最后 2 列会作为 TAGS 值，第一列会作为子表名。
+
## 示例
### 写入单个超级表
@@ -163,6 +179,8 @@ sink {
}
```
+这里的 `${table_name}` 会被 TDengine Sink Writer
当作普通字符串原样使用（连接器不会按行替换它），因此本示例只有在任务初始化阶段由上游框架依据 `CatalogTable` 的标识替换 `stable`
时才能生效。目标超级表必须已经存在并具有匹配的 TAGS 列。
+
## 变更日志
<ChangeLog />
diff --git a/docs/zh/connectors/source/AmazonSqs.md
b/docs/zh/connectors/source/AmazonSqs.md
index 736cc4ca6b..159d5f4ecc 100644
--- a/docs/zh/connectors/source/AmazonSqs.md
+++ b/docs/zh/connectors/source/AmazonSqs.md
@@ -56,6 +56,15 @@ Amazon SQS 源连接器用于从一个 Amazon SQS 队列 URL 读取消息。连
- `access_key_id` 和 `secret_access_key` 是可选项；如果使用静态 AWS 凭证，需要两个一起配置。
- 该源连接器只执行一次 receive 请求，最多读取 10 条消息，然后结束这个有界任务。
+## 认证
+
+连接器按以下顺序解析 AWS 凭证：
+
+1. 如果同时配置了 `access_key_id` 和 `secret_access_key`，则使用这对静态凭证。
+2. 否则，回退到 AWS 默认凭证链（环境变量、实例角色等）。
+
+针对 LocalStack、ElasticMQ 等 SQS 兼容本地服务进行测试时，把 `url` 指向本地端点（例如
`http://sqs-host:4566/...`），并提供任意非空的 `access_key_id` / `secret_access_key`
即可。SQS 兼容的测试服务通常不会校验请求里的 SigV4 签名，因此任意一对静态凭证都会被接受。
+
## 任务示例
### 在本地兼容队列之间复制消息
@@ -142,6 +151,28 @@ sink {
}
```
+### 读取 Debezium JSON 消息
+
+当上游系统（例如 Debezium 或其他 CDC 源）以 Debezium 信封形式发布变更事件时，把 `format` 设为
`debezium_json`，并通过 `debezium_record_include_schema` 控制消息中是否包含 schema 字段。
+
+```hocon
+source {
+ AmazonSqs {
+ url = "https://sqs.us-east-1.amazonaws.com/123456789012/cdc_events";
+ region = "us-east-1"
+ format = debezium_json
+ debezium_record_include_schema = true
+ schema = {
+ fields {
+ id = bigint
+ name = string
+ score = double
+ }
+ }
+ }
+}
+```
+
## 变更日志
<ChangeLog />
diff --git a/docs/zh/connectors/source/HugeGraph.md
b/docs/zh/connectors/source/HugeGraph.md
index e030146d76..c3c37869c8 100644
--- a/docs/zh/connectors/source/HugeGraph.md
+++ b/docs/zh/connectors/source/HugeGraph.md
@@ -4,6 +4,12 @@ import ChangeLog from '../changelog/connector-hugegraph.md';
`Source: HugeGraph`
+## 支持的引擎
+
+> Spark<br/>
+> Flink<br/>
+> SeaTunnel Zeta<br/>
+
## 描述
HugeGraph Source Connector 通过 HugeGraph REST API 读取 Apache HugeGraph 图数据。
@@ -16,9 +22,9 @@ HugeGraph Source Connector 通过 HugeGraph REST API 读取 Apache
HugeGraph 图
## 主要特性
-- [x] [batch](../../introduction/concepts/connector-v2-features.md)
-- [x] [parallelism](../../introduction/concepts/connector-v2-features.md)
-- [ ] [cdc](../../introduction/concepts/connector-v2-features.md)
+- [x] [批处理](../../introduction/concepts/connector-v2-features.md)
+- [x] [并行度](../../introduction/concepts/connector-v2-features.md)
+- [ ] [变更数据捕获](../../introduction/concepts/connector-v2-features.md)
## 参数
@@ -88,6 +94,8 @@ cardinality 为 `LIST` 或 `SET` 的 HugeGraph 属性会被读为 SeaTunnel 的
## 示例
+### 显式声明 schema 读取顶点 label
+
```hocon
source {
HugeGraph {
@@ -107,6 +115,53 @@ source {
}
```
+### 读取边 label
+
+读取边时，将 `label_type` 设为 `EDGE`，并在 `schema.fields` 中列出边的属性。输出还会包含保留列
`~source_id`、`~source_label`、`~target_id`、`~target_label`。
+
+```hocon
+source {
+ HugeGraph {
+ host = "localhost"
+ port = 8080
+ graph_name = "hugegraph"
+ label = "knows"
+ label_type = "EDGE"
+ schema = {
+ fields = {
+ since = "int"
+ }
+ }
+ }
+}
+```
+
+### 使用服务端属性等值过滤
+
+`parallelism = 1` 时，可以通过 `filter` 把属性等值条件推到服务端，只返回匹配全部条件的元素。
+
+```hocon
+source {
+ HugeGraph {
+ host = "localhost"
+ port = 8080
+ graph_name = "hugegraph"
+ label = "person"
+ label_type = "VERTEX"
+ filter = {
+ country = "US"
+ active = "true"
+ }
+ schema = {
+ fields = {
+ name = "string"
+ age = "int"
+ }
+ }
+ }
+}
+```
+
## Schema 自动发现
`schema` 为可选项。省略时，connector 会在作业构建阶段连接服务端，读取 `label`
的定义，为每个属性键生成一个输出列（类型见[类型映射](#类型映射)表，`LIST`/`SET` 映射为 `array<T>`），并按属性名排序。适合「整
label 全字段 dump」而又不想逐字段手写声明的场景。
diff --git a/docs/zh/connectors/source/Pulsar.md
b/docs/zh/connectors/source/Pulsar.md
index 055f72a92b..55700815c8 100644
--- a/docs/zh/connectors/source/Pulsar.md
+++ b/docs/zh/connectors/source/Pulsar.md
@@ -266,6 +266,94 @@ source {
用于 batch 作业时，请使用 `LATEST` 或 `TIMESTAMP` 这类有界停止模式；`cursor.stop.mode = "NEVER"`
适合流式作业。
+### 读取 Avro 消息
+
+当 topic 中是 Avro 编码的消息时，使用 `format = avro`，并在 `schema` 中声明字段类型，连接器会按 SeaTunnel
类型系统解码 Avro 数据。
+
+```hocon
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ Pulsar {
+ topic = "test_avro_topic"
+ subscription.name = "seatunnel-avro"
+ client.service-url = "pulsar://localhost:6650"
+ admin.service-url = "http://localhost:8080";
+ cursor.startup.mode = "EARLIEST"
+ cursor.stop.mode = "LATEST"
+ format = avro
+ schema = {
+ fields {
+ id = bigint
+ c_string = string
+ c_int = int
+ c_double = double
+ c_timestamp = timestamp
+ }
+ }
+ }
+}
+```
+
+### 流式读取 topic
+
+使用 `cursor.stop.mode = "NEVER"` 可以持续读取新消息直到作业停止。配合 `cursor.startup.mode =
"LATEST"` 可以从最新消息开始，避免重放历史消息。
+
+```hocon
+env {
+ parallelism = 2
+ job.mode = "STREAMING"
+ checkpoint.interval = 10000
+}
+
+source {
+ Pulsar {
+ topic = "persistent://public/default/events"
+ subscription.name = "seatunnel-stream"
+ client.service-url = "pulsar://localhost:6650"
+ admin.service-url = "http://localhost:8080";
+ cursor.startup.mode = "LATEST"
+ cursor.stop.mode = "NEVER"
+ format = json
+ schema = {
+ fields {
+ event_id = string
+ user_id = bigint
+ payload = string
+ }
+ }
+ }
+}
+```
+
+### 按 topic-pattern 自动发现新 topic
+
+当 topic 列表会随时间增长时，可以组合使用 `topic-pattern` 与
`topic-discovery.interval`，让连接器自动发现新增的 topic。
+
+```hocon
+source {
+ Pulsar {
+ topic-pattern = "persistent://public/default/orders-.*"
+ subscription.name = "seatunnel-orders"
+ client.service-url = "pulsar://localhost:6650"
+ admin.service-url = "http://localhost:8080";
+ topic-discovery.interval = 30000
+ cursor.startup.mode = "EARLIEST"
+ cursor.stop.mode = "LATEST"
+ format = json
+ schema = {
+ fields {
+ order_id = bigint
+ amount = double
+ }
+ }
+ }
+}
+```
+
## 变更日志
<ChangeLog />
diff --git a/docs/zh/connectors/source/TDengine.md
b/docs/zh/connectors/source/TDengine.md
index fe460860a7..3a305d98af 100644
--- a/docs/zh/connectors/source/TDengine.md
+++ b/docs/zh/connectors/source/TDengine.md
@@ -4,6 +4,12 @@ import ChangeLog from '../changelog/connector-tdengine.md';
> TDengine 源端连接器
+## 支持的引擎
+
+> Spark<br/>
+> Flink<br/>
+> SeaTunnel Zeta<br/>
+
## 描述
从 TDengine 超级表读取数据。
@@ -18,25 +24,25 @@ import ChangeLog from '../changelog/connector-tdengine.md';
- [ ] [流式](../../introduction/concepts/connector-v2-features.md)
- [x] [精确一次](../../introduction/concepts/connector-v2-features.md)
- [x] [列投影](../../introduction/concepts/connector-v2-features.md)
-
- [x] [并行度](../../introduction/concepts/connector-v2-features.md)
- [ ] [支持用户自定义分片](../../introduction/concepts/connector-v2-features.md)
## 配置项
-| 名称 | 类型 | 必填 | 默认值 |
-|----------------|--------|------|----------------|
-| url | string | 是 | - |
-| username | string | 是 | - |
-| password | string | 是 | - |
-| database | string | 是 | - |
-| stable | string | 是 | - |
-| sub_tables | list | 否 | - |
-| lower_bound | string | 是 | - |
-| upper_bound | string | 是 | - |
-| read_columns | list | 否 | - |
-
-### url [string]
+| 名称 | 类型 | 必填 | 默认值 | 说明 |
+|------------------|--------|------|--------|------|
+| url | String | 是 | - | TDengine REST JDBC 连接地址，例如
`jdbc:TAOS-RS://localhost:6041/`。 |
+| username | String | 是 | - | 连接 TDengine 使用的用户名。 |
+| password | String | 是 | - | 连接 TDengine 使用的密码。 |
+| database | String | 是 | - | TDengine 数据库名称。 |
+| stable | String | 是 | - | TDengine 超级表名称。 |
+| sub_tables | List | 否 | - |
子表名称列表。不配置时读取该超级表下所有子表；配置后只读取指定子表。 |
+| lower_bound | String | 是 | - | 查询时间范围的下界（包含），例如 `2018-10-03
14:38:05.000`。 |
+| upper_bound | String | 是 | - | 查询时间范围的上界（不包含），例如 `2018-10-03
14:38:16.801`。 |
+| read_columns | List | 否 | - | 要读取的列名列表。不配置时读取所有列。读取超级表时请将 TAGS
列放在列表末尾；不要包含 `subtable_name`。 |
+| common-options | | 否 | - | Source 通用参数，请参考 [Source Common
Options](../common-options/source-common-options.md)。 |
+
+### url [String]
TDengine REST JDBC 连接地址。
@@ -46,40 +52,48 @@ TDengine REST JDBC 连接地址。
jdbc:TAOS-RS://localhost:6041/
```
-### username [string]
+### username [String]
连接 TDengine 使用的用户名。
-### password [string]
+### password [String]
连接 TDengine 使用的密码。
-### database [string]
+### database [String]
-TDengine 数据库名称。
+TDengine 数据库名称，数据库必须已经在服务端存在。
-### stable [string]
+### stable [String]
-TDengine 超级表名称。
+TDengine 超级表名称。连接器会按该超级表下的每个子表生成一个 source 分片。
-### sub_tables [list]
+### sub_tables [List]
-TDengine 子表名称列表。不配置时读取该超级表下的所有子表；配置后只读取指定子表。
+TDengine 子表名称列表。不配置时读取该超级表下的所有子表；配置后只读取指定子表。子表名需与服务端保持完全一致。
-### lower_bound [string]
+### lower_bound [String]
查询时间范围的下界，包含该时间点。连接器会为每个子表查询加上 `timestamp_column >= lower_bound` 条件。请使用
TDengine 可识别的时间字符串，例如 `2018-10-03 14:38:05.000`。
-### upper_bound [string]
+### upper_bound [String]
查询时间范围的上界，不包含该时间点。连接器会为每个子表查询加上 `timestamp_column < upper_bound` 条件。请使用
TDengine 可识别的时间字符串，例如 `2018-10-03 14:38:16.801`。
-### read_columns [list]
+### read_columns [List]
要读取的列名列表。不配置时读取所有列。读取超级表时，请将 TAGS 字段放在列表末尾。不要在这里配置
`subtable_name`，连接器会自动把它作为第一列输出。
`read_columns` 的顺序会决定 `subtable_name` 之后的输出字段顺序。如果后续继续写入 TDengine
Sink，请保持普通列在前、TAGS 字段在后，这样 Sink 才能正确区分普通列和 TAGS 值。
+### 通用选项
+
+Source 插件通用参数，请参考 [Source Common
Options](../common-options/source-common-options.md)。
+
+## 输出 Schema
+
+输出表的第一列固定为预留字段 `subtable_name`，用来标识该行来自哪个 TDengine 子表。后续字段为 `read_columns`
中声明的列（未设置时为所有列），顺序与 `read_columns` 保持一致；TAGS 列请按声明的顺序排在普通列之后。
+
## 示例
### 按时间范围读取所有子表
@@ -104,6 +118,26 @@ source {
}
```
+### 读取指定子表和指定列
+
+```hocon
+source {
+ TDengine {
+ url = "jdbc:TAOS-RS://localhost:6041/"
+ username = "root"
+ password = "taosdata"
+ database = "power"
+ stable = "meters"
+ lower_bound = "2018-10-03 14:38:05.000"
+ upper_bound = "2018-10-03 14:38:16.801"
+ sub_tables = ["d1001", "d1002"]
+ read_columns = ["ts", "current", "voltage", "phase", "off", "nc",
"location", "groupid"]
+ }
+}
+```
+
+该配置只会读取超级表 `meters` 下的子表 `d1001` 和 `d1002`。TAGS 字段（`location`、`groupid`）放在
`read_columns` 末尾，方便下游 TDengine Sink 正确切分普通列与 TAGS 值。
+
### 从 TDengine 读取并写入 TDengine
```hocon
@@ -137,23 +171,7 @@ sink {
}
```
-### 读取指定子表和指定列
-
-```hocon
-source {
- TDengine {
- url = "jdbc:TAOS-RS://localhost:6041/"
- username = "root"
- password = "taosdata"
- database = "power"
- stable = "meters"
- lower_bound = "2018-10-03 14:38:05.000"
- upper_bound = "2018-10-03 14:38:16.801"
- sub_tables = ["d1001", "d1002"]
- read_columns = ["ts", "current", "voltage", "phase", "off", "nc",
"location", "groupid"]
- }
-}
-```
+写入时，Sink 会使用源端产生的 `subtable_name` 作为目标子表名。TAGS
字段的数量会从目标超级表的元数据中读取，因此目标超级表必须先在服务端创建好。
## 变更日志
