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 917690446c [Docs][Connector-V2] Improve Pulsar connector docs (#11283)
917690446c is described below
commit 917690446c3abb28efe5b9d38033c4c3f9bede26
Author: Daniel Carter <[email protected]>
AuthorDate: Fri Jul 3 08:19:36 2026 +0800
[Docs][Connector-V2] Improve Pulsar connector docs (#11283)
Co-authored-by: DanielCarter-stack
<[email protected]>
Co-authored-by: Shenghang <[email protected]>
---
docs/en/connectors/sink/Pulsar.md | 188 ++++++++++++++++++---------------
docs/en/connectors/source/Pulsar.md | 86 +++++++++++----
docs/zh/connectors/sink/Pulsar.md | 201 ++++++++++++++++++++++--------------
docs/zh/connectors/source/Pulsar.md | 86 +++++++++++----
4 files changed, 360 insertions(+), 201 deletions(-)
diff --git a/docs/en/connectors/sink/Pulsar.md
b/docs/en/connectors/sink/Pulsar.md
index 68d30a1670..7e70db1774 100644
--- a/docs/en/connectors/sink/Pulsar.md
+++ b/docs/en/connectors/sink/Pulsar.md
@@ -8,15 +8,17 @@ import ChangeLog from '../changelog/connector-pulsar.md';
> Spark<br/>
> Flink<br/>
-> Seatunnel Zeta<br/>
+> SeaTunnel Zeta<br/>
## Key features
- [x] [exactly-once](../../introduction/concepts/connector-v2-features.md)
+- [x] [support multiple table
write](../../introduction/concepts/connector-v2-features.md)
+- [ ] [timer flush](../../introduction/concepts/connector-v2-features.md)
## Description
-Sink connector for Apache Pulsar.
+The Pulsar sink writes SeaTunnel rows to Apache Pulsar topics. It can write to
one configured topic, or route multi-table records by the table id carried in
each row.
## Supported DataSource Info
@@ -26,35 +28,40 @@ Sink connector for Apache Pulsar.
## Sink Options
-| Name | Type | Required | Default |
Description
|
-|----------------------|--------|----------|---------------------|------------------------------------------------------------------------------------------------------------------|
-| topic | String | No | - | Sink Pulsar
topic. Required for single-table writes and optional when records provide
`SeaTunnelRow.tableId`. |
-| client.service-url | String | Yes | - | Service URL
provider for Pulsar service.
|
-| admin.service-url | String | Yes | - | The Pulsar
service HTTP URL for the admin endpoint.
|
-| auth.plugin-class | String | No | - | Name of the
authentication plugin.
|
-| auth.params | String | No | - | Parameters
for the authentication plugin.
|
-| format | String | No | json | Data
format. The default format is json. Optional text format.
|
-| field_delimiter | String | No | , | Customize
the field delimiter for data format.
|
-| semantics | Enum | No | AT_LEAST_ONCE | Consistency
semantics for writing to pulsar.
|
-| transaction_timeout | Int | No | 600 | The
transaction timeout is specified as 10 minutes by default.
|
-| pulsar.config | Map | No | - | In addition
to the above parameters that must be specified by the Pulsar producer client.
|
-| message.routing.mode | Enum | No | RoundRobinPartition | Default
routing mode for messages to partition.
|
-| partition_key_fields | array | No | - | Configure
which fields are used as the key of the pulsar message.
|
-| common-options | config | no | - | Sink plugin
common parameters, please refer to [Sink Common
Options](../common-options/sink-common-options.md) for details. |
+| Name | Type | Required | Default |
Description
|
+|--------------------------|--------|----------|---------------------|------------------------------------------------------------------------------------------------------------------|
+| topic | String | No | - | Target
Pulsar topic. Required for normal single-table writes. Optional only when
multi-table rows carry table ids. |
+| client.service-url | String | Yes | - | Pulsar
client service URL, for example `pulsar://localhost:6650`.
|
+| admin.service-url | String | Yes | - | Pulsar
admin HTTP URL, for example `http://localhost:8080`.
|
+| auth.plugin-class | String | No | - | Pulsar
authentication plugin class name.
|
+| auth.params | String | No | - |
Parameters for the authentication plugin. Configure it together with
`auth.plugin-class`. |
+| format | String | No | json | Data
format. Supports `json` and `text`.
|
+| field_delimiter | String | No | , | Field
delimiter used when `format = "text"`.
|
+| semantics | Enum | No | AT_LEAST_ONCE | Write
consistency. Valid values: `NON`, `AT_LEAST_ONCE`, `EXACTLY_ONCE`.
|
+| transaction_timeout | Int | No | 600 | Pulsar
transaction timeout in seconds. Used by `EXACTLY_ONCE`.
|
+| pulsar.config | Map | No | - | Extra
producer properties passed to the Pulsar producer client.
|
+| message.routing.mode | Enum | No | RoundRobinPartition | Routing
mode for partitioned topics. Valid values: `SinglePartition`,
`RoundRobinPartition`. |
+| partition_key_fields | Array | No | - | Fields
used to build the Pulsar message key.
|
+| multi_table_sink_replica | Int | No | 1 | Writer
replica count for multi-table writes.
|
+| common-options | Config | No | - | Sink
plugin common parameters. See [Sink Common
Options](../common-options/sink-common-options.md). |
## Parameter Interpretation
-### client.service-url [String]
+### topic [String]
+
+The default Pulsar topic used by the sink.
+
+For single-table pipelines, configure `topic`. For multi-table pipelines, the
sink uses `SeaTunnelRow.getTableId()` as the target topic when the row has a
table id, and falls back to `topic` only when the row does not carry one.
-Service URL provider for Pulsar service.
-To connect to Pulsar using client libraries, you need to specify a Pulsar
protocol URL.
-You can assign Pulsar protocol URLs to specific clusters and use the Pulsar
scheme.
+If neither the row table id nor `topic` is available, the sink fails fast with
a configuration error.
-For example, `localhost`: `pulsar://localhost:6650,localhost:6651`.
+### client.service-url [String]
+
+Pulsar client service URL. Use the Pulsar protocol, for example
`pulsar://localhost:6650`.
### admin.service-url [String]
-The Pulsar service HTTP URL for the admin endpoint.
+Pulsar admin HTTP URL.
For example, `http://my-broker.example.com:8080`, or
`https://my-broker.example.com:8443` for TLS.
@@ -66,74 +73,50 @@ Name of the authentication plugin.
Parameters for the authentication plugin.
-For example, `key1:val1,key2:val2`
+For example, `key1:val1,key2:val2`.
### format [String]
-Data format. The default format is json. Optional text format. The default
field separator is ",".
-If you customize the delimiter, add the "field_delimiter" option.
+Data format. The default format is `json`. You can also use `text`. When using
`text`, configure `field_delimiter` if the default comma delimiter is not
suitable.
### field_delimiter [String]
-Customize the field delimiter for data format.The default field_delimiter is
','.
-
-### topic [String]
-
-The default Pulsar topic used by the sink.
-
-For single-table pipelines, this option is required.
-For multi-table pipelines, the sink will use `SeaTunnelRow.getTableId()` as
the target topic when it is present, and fall back to `topic` only when the row
does not carry a table id.
-
-If neither `SeaTunnelRow.getTableId()` nor `topic` is available, the sink
fails fast with a configuration error.
+Field delimiter used by the `text` format. The default value is `,`.
### semantics [Enum]
-Consistency semantics for writing to pulsar.
-Available options are EXACTLY_ONCE,NON,AT_LEAST_ONCE, default AT_LEAST_ONCE.
-If semantic is specified as EXACTLY_ONCE, we will use 2pc to guarantee the
message is sent to pulsar exactly once.
-If semantic is specified as NON, we will directly send the message to pulsar,
the data may duplicat/lost if
-job restart/retry or network error.
+Write consistency semantics.
+
+- `AT_LEAST_ONCE`: the default. Messages may be duplicated after job restart
or retry.
+- `EXACTLY_ONCE`: writes messages through Pulsar transactions. The Pulsar
cluster must enable transaction support, and `transaction_timeout` should be
greater than the checkpoint interval.
+- `NON`: sends messages directly without checkpoint coordination. Data may be
duplicated or lost after restart, retry, or network errors.
### transaction_timeout [Int]
-The transaction timeout is specified as 10 minutes by default.
-If the transaction does not commit within the specified timeout, the
transaction will be automatically aborted.
-So you need to ensure that the timeout is greater than the checkpoint interval.
+Pulsar transaction timeout in seconds. The default value is `600`.
+
+This option is only used when `semantics = "EXACTLY_ONCE"`. If a transaction
is not committed before the timeout, Pulsar aborts it automatically.
### pulsar.config [Map]
-In addition to the above parameters that must be specified by the Pulsar
producer client,
-the user can also specify multiple non-mandatory parameters for the producer
client,
-covering all the producer parameters specified in the official Pulsar document.
+Extra Pulsar producer properties. These properties are passed to the Pulsar
producer client.
### message.routing.mode [Enum]
-Default routing mode for partitioned messages.
-Available options are `SinglePartition` and `RoundRobinPartition`.
-When `SinglePartition` is selected and no key is provided, the partitioned
producer randomly selects one partition and publishes all messages to that
partition. When a key is provided, the producer hashes the key and sends the
message to the selected partition.
-When `RoundRobinPartition` is selected and no key is provided, the producer
publishes messages across all partitions in round-robin order to achieve
maximum throughput.
-Round-robin routing is applied at the batching delay boundary rather than per
individual message, so batching remains effective.
-
-### partition_key_fields [String]
-
-Configure which fields are used as the key of the pulsar message.
+Routing mode for partitioned topics.
-For example, if you want to use value of fields from upstream data as key, you
can assign field names to this property.
+- `SinglePartition`: without a message key, one partition is selected and all
messages are sent to it. With a key, Pulsar hashes the key and sends the
message to the selected partition.
+- `RoundRobinPartition`: without a message key, messages are sent across
partitions in round-robin order. Pulsar applies round-robin routing at the
batching-delay boundary, not per individual message.
-Upstream data is the following:
+### partition_key_fields [Array]
-| name | age | data |
-|------|-----|---------------|
-| Jack | 16 | data-example1 |
-| Mary | 23 | data-example2 |
+Fields used to build the Pulsar message key.
-If name is set as the key, then the hash value of the name column will
determine which partition the message is sent to.
+For example, if the upstream row has `name` and `age`, setting
`partition_key_fields = ["name"]` builds a JSON key such as `{"name":"Jack"}`.
The selected fields must exist in the upstream schema.
-If not set partition key fields, the null message key will be sent to.
+### multi_table_sink_replica [Int]
-The format of the message key is json, If name is set as the key, for example
'{"name":"Jack"}'.
-
-The selected field must be an existing field in the upstream.
+Replica count for multi-table sink writers. It applies to all target topics in
the multi-table sink.
### common options
@@ -141,27 +124,25 @@ Sink plugin common parameters, please refer to [Sink
Common Options](../common-o
## Task Example
-### Simple
-
-> This example defines a SeaTunnel synchronization task that automatically
generates data through FakeSource and sends it to Pulsar Sink. FakeSource
generates a total of 16 rows of data (row.num=16), with each row having two
fields, name (string type) and age (int type). The final target topic is
test_topic will also be 16 rows of data in the topic. And if you have not yet
installed and deployed SeaTunnel, you need to follow the instructions in
[Install SeaTunnel](../../getting-started/l [...]
+### Write FakeSource To Pulsar
```hocon
-# Defining the runtime environment
env {
- # You can set flink configuration here
- execution.parallelism = 1
+ parallelism = 1
job.mode = "BATCH"
}
source {
FakeSource {
- parallelism = 1
plugin_output = "fake"
- row.num = 16
+ row.num = 10
schema = {
fields {
- name = "string"
- age = "int"
+ c_string = string
+ c_int = int
+ c_bigint = bigint
+ c_double = double
+ c_timestamp = timestamp
}
}
}
@@ -169,27 +150,64 @@ source {
sink {
Pulsar {
- topic = "example"
- client.service-url = "localhost:pulsar://localhost:6650"
- admin.service-url = "http://my-broker.example.com:8080";
- plugin_output = "test"
+ topic = "topic_test"
+ client.service-url = "pulsar://localhost:6650"
+ admin.service-url = "http://localhost:8080";
+ format = json
pulsar.config = {
- sendTimeoutMs = 30000
+ sendTimeoutMs = 30000
}
}
}
```
-### Multi-table
+### Write With Message Key
+
+```hocon
+sink {
+ Pulsar {
+ topic = "orders"
+ client.service-url = "pulsar://localhost:6650"
+ admin.service-url = "http://localhost:8080";
+ partition_key_fields = ["order_id"]
+ message.routing.mode = "SinglePartition"
+ format = json
+ }
+}
+```
+
+### Exactly-Once Write
+
+```hocon
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+ checkpoint.interval = 10000
+}
+
+sink {
+ Pulsar {
+ topic = "orders"
+ client.service-url = "pulsar://localhost:6650"
+ admin.service-url = "http://localhost:8080";
+ semantics = "EXACTLY_ONCE"
+ transaction_timeout = 600
+ format = json
+ }
+}
+```
+
+### Multi-Table Write
-> This example routes each row to the Pulsar topic carried in
`SeaTunnelRow.tableId`. In this mode, `topic` can be omitted.
+When upstream rows carry table ids, the sink can route each row to the topic
with the same table id. In this mode, `topic` can be omitted.
```hocon
sink {
Pulsar {
client.service-url = "pulsar://localhost:6650"
admin.service-url = "http://localhost:8080";
- plugin_output = "test"
+ multi_table_sink_replica = 2
+ format = json
}
}
```
diff --git a/docs/en/connectors/source/Pulsar.md
b/docs/en/connectors/source/Pulsar.md
index 23833d3595..5c50f511c5 100644
--- a/docs/en/connectors/source/Pulsar.md
+++ b/docs/en/connectors/source/Pulsar.md
@@ -40,7 +40,7 @@ Source connector for Apache Pulsar.
| cursor.stop.mode | Enum | No | NEVER | Stop
position mode. Options: `NEVER` (streaming), `LATEST` (batch), `TIMESTAMP`
(batch) |
| cursor.stop.timestamp | Long | No | - | Stop
timestamp (ms) when `cursor.stop.mode=TIMESTAMP`
|
| schema | Config | No | - | Data
structure including field names and types
|
-| format | String | No | json | Data format.
Default is json. **Multi-table mode only supports JSON and CANAL_JSON**
|
+| format | String | No | json | Data format.
Supports `json` and `canal_json`.
|
| common-options | | No | - | Source
plugin common parameters. See [Source Common
Options](../common-options/source-common-options.md) for details |
### topic [String]
@@ -85,7 +85,9 @@ The interval (in ms) for the Pulsar source to discover the
new topic partitions.
### subscription.name [String]
-Specify the subscription name for this consumer. This is required for each
effective table configuration, but in multi-table mode it can be defined
globally or overridden per `tables_configs` item.
+Specify the subscription name for this consumer.
+
+For a single-table source, `subscription.name` is required. For a multi-table
source, it can be defined globally or inside each `tables_configs` item. If it
is configured in both places, the item-level value takes effect for that item.
### client.service-url [String]
@@ -121,7 +123,7 @@ The interval time (in ms) when fetching records. A shorter
time increases throug
### poll.batch.size [Integer]
-The maximum number of records to fetch to wait when polling. A longer time
increases throughput but also latency.
+The maximum number of records to fetch in one poll.
### cursor.startup.mode [Enum]
@@ -156,9 +158,9 @@ Stop from the specified epoch timestamp (in milliseconds).
The structure of the data, including field names and field types.
reference to [Schema-Feature](../../introduction/concepts/schema-feature.md)
-## format [String]
+### format [String]
-Data format. The default format is json, reference [formats](../formats).
+Data format. The default value is `json`. Pulsar source currently supports
`json` and `canal_json`.
### common options
@@ -166,50 +168,94 @@ Source plugin common parameters, please refer to [Source
Common Options](../comm
## Example
+### Single-Topic Batch Read
+
```hocon
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
source {
Pulsar {
- topic = "example"
+ topic = "topic-it"
subscription.name = "seatunnel"
client.service-url = "pulsar://localhost:6650"
- admin.service-url = "http://my-broker.example.com:8080";
- plugin_output = "test"
+ admin.service-url = "http://localhost:8080";
+ cursor.startup.mode = "EARLIEST"
+ cursor.stop.mode = "LATEST"
+ format = json
+ schema = {
+ fields {
+ c_string = string
+ c_boolean = boolean
+ c_int = int
+ c_bigint = bigint
+ c_double = double
+ c_timestamp = timestamp
+ }
+ }
+ }
+}
+```
+
+### Read Canal JSON Messages
+
+Use `format = canal_json` when the Pulsar topic stores Canal JSON change
events.
+
+```hocon
+source {
+ Pulsar {
+ topic = "test-cdc_mds"
+ subscription.name = "seatunnel-cdc-sub"
+ client.service-url = "pulsar://localhost:6650"
+ admin.service-url = "http://localhost:8080";
+ cursor.startup.mode = "EARLIEST"
+ cursor.stop.mode = "LATEST"
+ format = canal_json
+ schema = {
+ fields {
+ id = int
+ name = string
+ description = string
+ weight = string
+ }
+ }
}
}
```
-## Multi-table Example
+### Multi-Table Read
```hocon
source {
Pulsar {
- subscription.name = "seatunnel-sub"
client.service-url = "pulsar://localhost:6650"
admin.service-url = "http://localhost:8080";
cursor.startup.mode = "EARLIEST"
- cursor.stop.mode = "NEVER"
+ cursor.stop.mode = "LATEST"
format = "json"
tables_configs = [
{
- table_path = "default.orders"
+ table_path = "db.orders"
topic = "persistent://public/default/orders"
+ subscription.name = "sub-orders"
schema = {
fields {
- order_id = "bigint"
- user_id = "int"
+ order_id = int
+ amount = double
}
}
},
{
- table_path = "default.users"
+ table_path = "db.users"
topic-pattern = "persistent://public/default/users-.*"
- subscription.name = "users-sub"
- format = "canal_json"
+ subscription.name = "sub-users"
schema = {
fields {
- user_id = "int"
- name = "string"
+ user_id = int
+ name = string
}
}
}
@@ -218,7 +264,7 @@ source {
}
```
-In batch mode, replace `cursor.stop.mode = "NEVER"` with a bounded mode such
as `LATEST` or `TIMESTAMP`.
+For batch jobs, use a bounded stop mode such as `LATEST` or `TIMESTAMP`. Use
`cursor.stop.mode = "NEVER"` for streaming jobs.
## Changelog
diff --git a/docs/zh/connectors/sink/Pulsar.md
b/docs/zh/connectors/sink/Pulsar.md
index ef0b820c03..4e53fab6ce 100644
--- a/docs/zh/connectors/sink/Pulsar.md
+++ b/docs/zh/connectors/sink/Pulsar.md
@@ -2,149 +2,147 @@ import ChangeLog from '../changelog/connector-pulsar.md';
# Pulsar
-> Pulsar 数据连接器
+> Pulsar Sink 连接器
## 引擎支持
> Spark<br/>
> Flink<br/>
-> Seatunnel Zeta<br/>
+> SeaTunnel Zeta<br/>
-## 核心特性
+## 主要特性
-- [x] [精准一次](../../introduction/concepts/connector-v2-features.md)
-- [ ] [timer flush](../../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)
## 描述
-Apache Pulsar 的接收连接器。
+Pulsar Sink 用于将 SeaTunnel 数据写入 Apache Pulsar topic。它既可以写入一个固定
topic，也可以在多表写入时根据每行数据携带的表标识路由到不同 topic。
## 支持的数据源信息
-| 数据源 | 支持的版本 |
-|--------|-----------|
-| Pulsar | Universal |
+| 数据源 | 支持的版本 |
+|--------|------------|
+| Pulsar | Universal |
## 输出选项
-| 名称 | 类型 | 是否必须 | 默认值 |
描述 |
-|----------------------|--------|------|---------------------|-------------------------------------------------|
-| topic | String | Yes | - | 输出到Pulsar主题名称.
|
-| client.service-url | String | Yes | - | Pulsar 服务的服务
URL 提供者. |
-| admin.service-url | String | Yes | - | 管理端点的 Pulsar 服务
HTTP URL. |
-| auth.plugin-class | String | No | - | 身份验证插件的名称.
|
-| auth.params | String | No | - | 身份验证插件的参数.
|
-| format | String | No | json | 数据格式。默认格式为
json。可选的文本格式. |
-| field_delimiter | String | No | , | 自定义数据格式的字段分隔符.
|
-| semantics | Enum | No | AT_LEAST_ONCE | 写入 pulsar
的一致性语义. |
-| transaction_timeout | Int | No | 600 | 默认情况下，事务超时指定为
10 分钟. |
-| pulsar.config | Map | No | - | 除了上述必须由 Pulsar
生产者客户端指定的参数外. |
-| message.routing.mode | Enum | No | RoundRobinPartition | 要分区的消息的默认路由模式.
|
-| partition_key_fields | array | No | - | 配置哪些字段用作 pulsar
消息的键. |
-| common-options | config | no | - | Sink插件常用参数，详见
[Sink通用选项](../common-options/sink-common-options.md). |
+| 名称 | 类型 | 是否必须 | 默认值 | 描述
|
+|--------------------------|--------|----------|---------------------|----------------------------------------------------------------------|
+| topic | String | 否 | - | 目标
Pulsar topic。普通单表写入必须配置；多表数据带有表标识时可以省略。 |
+| client.service-url | String | 是 | - | Pulsar
客户端服务地址，例如 `pulsar://localhost:6650`。 |
+| admin.service-url | String | 是 | - | Pulsar
管理端 HTTP 地址，例如 `http://localhost:8080`。 |
+| auth.plugin-class | String | 否 | - | Pulsar
认证插件类名。 |
+| auth.params | String | 否 | - |
认证插件参数。需要和 `auth.plugin-class` 一起配置。 |
+| format | String | 否 | json | 数据格式。支持
`json` 和 `text`。 |
+| field_delimiter | String | 否 | , | 当
`format = "text"` 时使用的字段分隔符。 |
+| semantics | Enum | 否 | AT_LEAST_ONCE |
写入一致性语义。可选值：`NON`、`AT_LEAST_ONCE`、`EXACTLY_ONCE`。 |
+| transaction_timeout | Int | 否 | 600 | Pulsar
事务超时时间，单位为秒。用于 `EXACTLY_ONCE`。 |
+| pulsar.config | Map | 否 | - | 传给
Pulsar 生产者客户端的额外参数。 |
+| message.routing.mode | Enum | 否 | RoundRobinPartition | 分区 topic
的消息路由模式。可选值：`SinglePartition`、`RoundRobinPartition`。 |
+| partition_key_fields | Array | 否 | - | 用于生成
Pulsar 消息 key 的字段。 |
+| multi_table_sink_replica | Int | 否 | 1 |
多表写入时的写入器副本数。 |
+| common-options | Config | 否 | - | Sink
插件通用参数，详见 [Sink 通用选项](../common-options/sink-common-options.md)。 |
## 参数解释
-### client.service-url [String]
+### topic [String]
+
+Sink 默认写入的 Pulsar topic。
+
+普通单表任务需要配置 `topic`。多表任务中，如果每行数据带有表标识，Sink 会把这个表标识作为目标 topic；只有当数据没有表标识时才会回退使用
`topic`。
-Pulsar 服务的 Service URL 提供程序。要使用客户端库连接到 Pulsar，
-您需要指定一个 Pulsar 协议 URL。您可以将 Pulsar 协议 URL 分配给特定集群并使用 Pulsar 方案。
+如果数据没有表标识，并且也没有配置 `topic`，任务会直接报配置错误。
+
+### client.service-url [String]
-例如, `localhost`: `pulsar://localhost:6650,localhost:6651`.
+Pulsar 客户端服务地址。请使用 Pulsar 协议，例如 `pulsar://localhost:6650`。
### admin.service-url [String]
-管理端点的 Pulsar 服务 HTTP URL.
+Pulsar 管理端 HTTP 地址。
-例如, `http://my-broker.example.com:8080`, or
`https://my-broker.example.com:8443` for TLS.
+例如：`http://my-broker.example.com:8080`；如果启用了 TLS，可以使用
`https://my-broker.example.com:8443`。
### auth.plugin-class [String]
-身份验证插件的名称。
+认证插件类名。
### auth.params [String]
-身份验证插件的参数。
+认证插件参数。
-例如, `key1:val1,key2:val2`
+例如：`key1:val1,key2:val2`。
### format [String]
-数据格式。默认格式为 json。可选的文本格式。默认字段分隔符为","。如果自定义分隔符，请添加"field_delimiter"选项。
+数据格式。默认值为 `json`，也可以使用 `text`。使用 `text` 时，如果默认逗号分隔符不合适，请配置 `field_delimiter`。
### field_delimiter [String]
-自定义数据格式的字段分隔符。默认field_delimiter为','。
+`text` 格式使用的字段分隔符。默认值为 `,`。
### semantics [Enum]
-写入 pulsar 的一致性语义。可用选项包括 EXACTLY_ONCE、NON、AT_LEAST_ONCE、默认AT_LEAST_ONCE。
-如果语义被指定为 EXACTLY_ONCE，我们将使用 2pc 来保证消息被准确地发送到 pulsar 一次。
-如果语义指定为 NON，我们将直接将消息发送到 pulsar，如果作业重启/重试或网络错误，数据可能会重复/丢失。
+写入一致性语义。
+
+- `AT_LEAST_ONCE`：默认值。作业重启或重试后，消息可能重复。
+- `EXACTLY_ONCE`：通过 Pulsar 事务写入。Pulsar 集群必须开启事务能力，并且 `transaction_timeout` 应大于
checkpoint 间隔。
+- `NON`：不和 checkpoint 协调，直接发送消息。作业重启、重试或网络异常后，数据可能重复或丢失。
### transaction_timeout [Int]
-默认情况下，事务超时指定为 10 分钟。如果事务未在指定的超时时间内提交，则事务将自动中止。因此，您需要确保超时大于检查点间隔。
+Pulsar 事务超时时间，单位为秒。默认值为 `600`。
+
+该参数只在 `semantics = "EXACTLY_ONCE"` 时生效。如果事务没有在超时时间内提交，Pulsar 会自动中止该事务。
### pulsar.config [Map]
-除了上述 Pulsar 生产者客户端必须指定的参数外，用户还可以为生产者客户端指定多个非强制性参数，
-涵盖 Pulsar 官方文档中指定的所有生产者参数。
+Pulsar 生产者客户端的额外参数。这些参数会传给 Pulsar producer。
### message.routing.mode [Enum]
-分区消息的默认路由模式。可用选项包括 `SinglePartition` 和 `RoundRobinPartition`。
-选择 `SinglePartition` 且未提供 key 时，分区生产者会随机选择一个分区，并将所有消息发布到该分区。提供 key 时，生产者会对 key
做哈希，并将消息发送到对应分区。
-选择 `RoundRobinPartition` 且未提供 key
时，生产者会以轮询方式将消息发布到所有分区，以获得最大吞吐量。请注意，轮询不是逐条消息执行，而是在批处理延迟边界上执行，以确保批处理仍然有效。
-
-### partition_key_fields [String]
-
-配置哪些字段用作 pulsar 消息的键。
-
-例如，如果要使用上游数据中的字段值作为键，则可以为此属性分配字段名称。
+分区 topic 的消息路由模式。
-上游数据如下：
+- `SinglePartition`：没有消息 key 时，选择一个分区并把所有消息写入该分区；有 key 时，Pulsar 会根据 key 哈希选择分区。
+- `RoundRobinPartition`：没有消息 key 时，按轮询方式写入不同分区。Pulsar 的轮询发生在批处理延迟边界上，不是逐条消息轮询。
-| name | age | data |
-|------|-----|---------------|
-| Jack | 16 | data-example1 |
-| Mary | 23 | data-example2 |
+### partition_key_fields [Array]
-如果将 name 设置为键，则 name 列的哈希值将确定消息发送到哪个分区。
+用于生成 Pulsar 消息 key 的字段。
-如果未设置分区键字段，则将向 null 消息键发送至。
+例如上游数据包含 `name` 和 `age`，配置 `partition_key_fields = ["name"]` 后，会生成类似
`{"name":"Jack"}` 的 JSON key。所选字段必须存在于上游数据结构中。
-消息键的格式为 json，如果 name 设置为键，例如 '{“name”：“Jack”}'。
+### multi_table_sink_replica [Int]
-所选字段必须是上游的现有字段。
+多表写入时的写入器副本数。该配置会对多表 Sink 中的所有目标 topic 生效。
-### 常见选项
+### common options
-Sink插件常用参数，详见[Sink通用选项](../common-options/sink-common-options.md).
+Sink 插件通用参数，请参考 [Sink 通用选项](../common-options/sink-common-options.md)。
## 任务示例
-### 简单
-
-> 该示例定义了一个 SeaTunnel 同步任务，该任务通过 FakeSource 自动生成数据并将其发送到 Pulsar Sink。FakeSource
总共生成 16 行数据 （row.num=16），每行有两个字段，name（字符串类型）和 age（int
类型）。最终目标主题是test_topic主题中还将有 16 行数据。 如果您尚未安装和部署
SeaTunnel，则需要按照[安装Seatunnel](../../getting-started/locally/deployment.md)
SeaTunnel 中的说明安装和部署 SeaTunnel。然后按照 [SeaTunnel
引擎快速入门](../../getting-started/locally/quick-start-seatunnel-engine.md)中的说明运行此作业。
+### 写入 FakeSource 数据到 Pulsar
```hocon
-# Defining the runtime environment
env {
- # You can set flink configuration here
- execution.parallelism = 1
+ parallelism = 1
job.mode = "BATCH"
}
source {
FakeSource {
- parallelism = 1
plugin_output = "fake"
- row.num = 16
+ row.num = 10
schema = {
fields {
- name = "string"
- age = "int"
+ c_string = string
+ c_int = int
+ c_bigint = bigint
+ c_double = double
+ c_timestamp = timestamp
}
}
}
@@ -152,17 +150,68 @@ source {
sink {
Pulsar {
- topic = "example"
- client.service-url = "localhost:pulsar://localhost:6650"
- admin.service-url = "http://my-broker.example.com:8080";
- plugin_output = "test"
+ topic = "topic_test"
+ client.service-url = "pulsar://localhost:6650"
+ admin.service-url = "http://localhost:8080";
+ format = json
pulsar.config = {
- sendTimeoutMs = 30000
+ sendTimeoutMs = 30000
}
}
}
```
+### 使用消息 Key 写入
+
+```hocon
+sink {
+ Pulsar {
+ topic = "orders"
+ client.service-url = "pulsar://localhost:6650"
+ admin.service-url = "http://localhost:8080";
+ partition_key_fields = ["order_id"]
+ message.routing.mode = "SinglePartition"
+ format = json
+ }
+}
+```
+
+### 精确一次写入
+
+```hocon
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+ checkpoint.interval = 10000
+}
+
+sink {
+ Pulsar {
+ topic = "orders"
+ client.service-url = "pulsar://localhost:6650"
+ admin.service-url = "http://localhost:8080";
+ semantics = "EXACTLY_ONCE"
+ transaction_timeout = 600
+ format = json
+ }
+}
+```
+
+### 多表写入
+
+当上游数据带有表标识时，Sink 可以把每行数据路由到同名 topic。此时可以不配置 `topic`。
+
+```hocon
+sink {
+ Pulsar {
+ client.service-url = "pulsar://localhost:6650"
+ admin.service-url = "http://localhost:8080";
+ multi_table_sink_replica = 2
+ format = json
+ }
+}
+```
+
## 变更日志
<ChangeLog />
diff --git a/docs/zh/connectors/source/Pulsar.md
b/docs/zh/connectors/source/Pulsar.md
index b84dbef738..da41f38690 100644
--- a/docs/zh/connectors/source/Pulsar.md
+++ b/docs/zh/connectors/source/Pulsar.md
@@ -40,7 +40,7 @@ Apache Pulsar 的源连接器。
| cursor.stop.mode | Enum | 否 | NEVER |
停止位置模式。可选值:`NEVER`(流式)、`LATEST`(批式)、`TIMESTAMP`(批式)
|
| cursor.stop.timestamp | Long | 否 | - | 当
`cursor.stop.mode=TIMESTAMP` 时的停止时间戳(毫秒)
|
| schema | Config | 否 | - | 数据结构,包括字段名称和字段类型
|
-| format | String | 否 | json | 数据格式。默认为 json。**多表模式仅支持
JSON 和 CANAL_JSON** |
+| format | String | 否 | json | 数据格式。支持 `json` 和
`canal_json`。 |
| common-options | | 否 | - | Source 插件通用参数,请参考
[Source Common Options](../common-options/source-common-options.md) 了解详情
|
### topic [String]
@@ -86,7 +86,9 @@ Pulsar 源发现新主题分区的间隔（毫秒）。非正值禁用主题分
### subscription.name [String]
-消费者订阅名。对每个最终生效的表配置都是必需的；在多表模式下，可以定义在全局，也可以在 `tables_configs` 的 item 中单独覆盖。
+消费者订阅名。
+
+单表读取时必须配置 `subscription.name`。多表读取时，可以配置在全局，也可以配置在每个 `tables_configs` item
内；如果两处都配置，item 内的值对该 item 生效。
### client.service-url [String]
@@ -120,7 +122,7 @@ Pulsar 服务管理端点的 HTTP URL。
### poll.batch.size [Integer]
-轮询时要获取的最大记录数。更长的时间会增加吞吐量但也会增加延迟。
+单次轮询最多获取的记录数。
### cursor.startup.mode [Enum]
@@ -148,9 +150,9 @@ Pulsar 消费者的启动模式，有效值为 `'EARLIEST'`、`'LATEST'`、`'SUB
数据结构定义，包括字段名和字段类型。参考 [Schema
Feature](../../introduction/concepts/schema-feature.md)。
-## format [String]
+### format [String]
-数据格式。默认值为 `json`。更多格式说明参考 [formats](../formats)。
+数据格式。默认值为 `json`。Pulsar Source 当前支持 `json` 和 `canal_json`。
### 通用参数
@@ -158,50 +160,94 @@ Source 插件通用参数请参考 [Source Common
Options](../common-options/sou
## 示例
+### 单 Topic 批式读取
+
```hocon
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
source {
Pulsar {
- topic = "example"
+ topic = "topic-it"
subscription.name = "seatunnel"
client.service-url = "pulsar://localhost:6650"
- admin.service-url = "http://my-broker.example.com:8080";
- plugin_output = "test"
+ admin.service-url = "http://localhost:8080";
+ cursor.startup.mode = "EARLIEST"
+ cursor.stop.mode = "LATEST"
+ format = json
+ schema = {
+ fields {
+ c_string = string
+ c_boolean = boolean
+ c_int = int
+ c_bigint = bigint
+ c_double = double
+ c_timestamp = timestamp
+ }
+ }
+ }
+}
+```
+
+### 读取 Canal JSON 消息
+
+当 Pulsar topic 中保存的是 Canal JSON 变更事件时，使用 `format = canal_json`。
+
+```hocon
+source {
+ Pulsar {
+ topic = "test-cdc_mds"
+ subscription.name = "seatunnel-cdc-sub"
+ client.service-url = "pulsar://localhost:6650"
+ admin.service-url = "http://localhost:8080";
+ cursor.startup.mode = "EARLIEST"
+ cursor.stop.mode = "LATEST"
+ format = canal_json
+ schema = {
+ fields {
+ id = int
+ name = string
+ description = string
+ weight = string
+ }
+ }
}
}
```
-## 多表示例
+### 多表读取
```hocon
source {
Pulsar {
- subscription.name = "seatunnel-sub"
client.service-url = "pulsar://localhost:6650"
admin.service-url = "http://localhost:8080";
cursor.startup.mode = "EARLIEST"
- cursor.stop.mode = "NEVER"
+ cursor.stop.mode = "LATEST"
format = "json"
tables_configs = [
{
- table_path = "default.orders"
+ table_path = "db.orders"
topic = "persistent://public/default/orders"
+ subscription.name = "sub-orders"
schema = {
fields {
- order_id = "bigint"
- user_id = "int"
+ order_id = int
+ amount = double
}
}
},
{
- table_path = "default.users"
+ table_path = "db.users"
topic-pattern = "persistent://public/default/users-.*"
- subscription.name = "users-sub"
- format = "canal_json"
+ subscription.name = "sub-users"
schema = {
fields {
- user_id = "int"
- name = "string"
+ user_id = int
+ name = string
}
}
}
@@ -210,7 +256,7 @@ source {
}
```
-如果用于 batch 作业，请将 `cursor.stop.mode = "NEVER"` 改为有界模式，例如 `LATEST` 或 `TIMESTAMP`。
+用于 batch 作业时，请使用 `LATEST` 或 `TIMESTAMP` 这类有界停止模式；`cursor.stop.mode = "NEVER"`
适合流式作业。
## 变更日志
