This is an automated email from the ASF dual-hosted git repository.
JNSimba pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris-website.git
The following commit(s) were added to refs/heads/master by this push:
new b2724cb04f5 [doc] Add TVF, TLS, and Binlog connector documentation
(#4105)
b2724cb04f5 is described below
commit b2724cb04f5e4eb6c310fe0a629d0121a62aff39
Author: wudi <[email protected]>
AuthorDate: Tue Sep 1 17:48:05 2026 +0800
[doc] Add TVF, TLS, and Binlog connector documentation (#4105)
## Versions
- [x] dev
- [ ] 4.x
- [ ] 3.x
- [ ] 2.1 or older (not covered by version/language sync gate)
## Languages
- [x] Chinese
- [x] English
## Docs Checklist
- [x] Checked by AI
- [ ] Test Cases Built
- [x] Updated required version and language counterparts, or explained
why not
- [x] If only one language changed, confirmed whether source/translation
counterparts need sync
## Proposed changes
- Document Doris Binlog incremental reading for the Flink Connector.
- Document TLS and S3 TVF configuration and usage for the Flink, Spark,
and Kafka Connectors.
- Keep the current English and Chinese documentation aligned.
This change targets the current dev documentation. Versioned
documentation remains unchanged.
---
.../data-integration/doris-kafka-connector.md | 60 ++++++-
.../data-integration/flink-doris-connector.md | 183 ++++++++++++++++++++-
.../data-integration/spark-doris-connector.md | 81 +++++++--
.../data-integration/doris-kafka-connector.md | 60 ++++++-
.../data-integration/flink-doris-connector.md | 183 ++++++++++++++++++++-
.../data-integration/spark-doris-connector.md | 81 +++++++--
6 files changed, 616 insertions(+), 32 deletions(-)
diff --git
a/docs/connection-integration/data-integration/doris-kafka-connector.md
b/docs/connection-integration/data-integration/doris-kafka-connector.md
index 2471d09d380..3fbd8ee022c 100644
--- a/docs/connection-integration/data-integration/doris-kafka-connector.md
+++ b/docs/connection-integration/data-integration/doris-kafka-connector.md
@@ -42,6 +42,7 @@ To continuously write data from Kafka into Doris, choose the
configuration that
| 24.0.0 | 2.4+ | 2.0+ | 8 |
| 25.0.0 | 2.4+ | 2.0+ | 8 |
| 26.0.0 | 2.4+ | 2.0+ | 8 |
+| 26.1.0 | 2.4+ | 2.0+ | 8 |
## Preparing the Doris Kafka Connector
@@ -59,7 +60,7 @@ In a Maven project, add the following dependency:
<dependency>
<groupId>org.apache.doris</groupId>
<artifactId>doris-kafka-connector</artifactId>
- <version>25.0.0</version>
+ <version>26.1.0</version>
</dependency>
```
@@ -251,7 +252,12 @@ The following configurations are used to create a Doris
Sink Connector. For othe
| `doris.user` | - | - | Y | Doris username. |
| `doris.password` | - | - | Y | Doris password. |
| `doris.database` | - | - | Y | The database to write to. It can be left
empty when writing to multiple databases, in which case the database name must
be specified in `topic2table.map`. |
+| `doris.enable.tls` | `true`,<br />`false` | false | N | Whether to enable
TLS for Doris HTTP and MySQL connections. |
+| `doris.tls.ca-certificate-path` | - | Empty string | N | Worker-local PEM CA
certificate chain path. When empty, the Connector does not load a custom CA and
uses the corresponding client's default trust store. |
+| `doris.tls.skip-hostname-verification` | `true`,<br />`false` | false | N |
Whether to skip hostname verification while retaining CA verification. |
+| `doris.tls.excluded-protocols` | `http`,<br />`mysql` | Empty string | N |
Comma-separated Doris protocols that do not use TLS. |
| `doris.topic2table.map` | - | - | Y | The mapping between topics and tables,
for example `topic1:tb1,topic2:tb2`. If left empty, the topic name is used as
the target table name by default. The format for multiple databases is
`topic1:db1.tbl1,topic2:db2.tbl2`. |
+| `load.model` | `stream_load`,<br />`tvf` | stream_load | N | Doris write
mode. Set this option to `tvf` to write through S3 TVF. |
| `buffer.count.records` | - | 50000 | N | The number of records written per
Stream Load. |
| `buffer.flush.time` | - | 120 | N | The buffer flush interval in seconds.
The default value is 120 seconds. |
| `buffer.size.bytes` | - | 104857600(100MB) | N | The data size written per
Stream Load. |
@@ -260,6 +266,13 @@ The following configurations are used to create a Doris
Sink Connector. For othe
| `label.prefix` | - | `${name}` | N | The label prefix for Stream Load when
importing data. The default value is the Connector application name. |
| `auto.redirect` | - | true | N | Whether to redirect Stream Load requests.
When enabled, Stream Load is redirected through the FE to the BE that needs to
write the data, and BE information is no longer displayed. |
| `sink.properties.*` | - | `'sink.properties.format':'json'`,<br
/>`'sink.properties.read_json_by_line':'true'` | N | Stream Load import
parameters. For example, define the column separator with
`'sink.properties.column_separator':','`. For detailed parameters, see [Stream
Load manual](../../data-operate/import/import-way/stream-load-manual.md).<br
/><br />To enable Group Commit, for example to enable `sync_mode`:
`"sink.properties.group_commit":"sync_mode"`. Group Commit supports three [...]
+| `sink.s3.endpoint` | - | - | TVF write mode only | Absolute HTTP or HTTPS
endpoint of the S3-compatible object storage. |
+| `sink.s3.region` | - | - | TVF write mode only | Object storage region. |
+| `sink.s3.bucket` | - | - | TVF write mode only | Bucket used to stage data. |
+| `sink.s3.prefix` | - | - | TVF write mode only | Object key prefix used to
stage data. |
+| `sink.s3.access-key` | - | - | TVF write mode only | Object storage access
key. |
+| `sink.s3.secret-key` | - | - | TVF write mode only | Object storage secret
key. |
+| `sink.s3.path-style-access` | `true`,<br />`false` | false | N | Whether to
use path-style object storage access in TVF write mode. |
| `delivery.guarantee` | `at_least_once`,<br />`exactly_once` | at_least_once
| N | The data consistency guarantee when consuming Kafka data and importing it
into Doris. Supports `at_least_once` and `exactly_once`. The default value is
`at_least_once`. Doris must be upgraded to 2.1.0 or later to guarantee
`exactly_once`. |
| `converter.mode` | `normal`,<br />`debezium_ingestion` | normal | N | The
upstream data type conversion mode used when the Connector consumes Kafka data.
`normal` means consuming Kafka data normally without special type conversion.
`debezium_ingestion` means special type conversion is required when the
upstream Kafka data is collected through CDC (Change Data Capture) tools such
as Debezium. |
| `debezium.schema.evolution` | `none`,<br />`basic` | none | N | When
collecting from upstream database systems (such as MySQL) through Debezium, if
a schema change occurs, added fields can be synchronized to Doris. `none` means
schema changes in the upstream database system are not synchronized to Doris.
`basic` means data change operations in the upstream database are synchronized.
Because column schema changes are dangerous operations and may accidentally
drop columns from the Doris [...]
@@ -558,6 +571,51 @@ After SMT processing, the sample data becomes:
Here, `repo` is the static field added by `InsertField`, and `registertime` is
the time string converted by `TimestampConverter`. For more Kafka Connect
Single Message Transforms (SMT) examples, see the [SMT
documentation](https://docs.confluent.io/cloud/current/connectors/transforms/overview.html).
+## Best practices
+
+### Access a TLS-enabled Doris environment from Kafka Connect
+
+Add the following options to the Connector configuration when writing to a
TLS-enabled Doris cluster:
+
+```json
+{
+"doris.urls":"doris-fe.example.com",
+"doris.http.port":"8040",
+"doris.query.port":"9030",
+"doris.enable.tls":"true",
+"doris.tls.ca-certificate-path":"/etc/kafka-connect/certs/doris-ca.pem"
+}
+```
+
+Keep only host names in `doris.urls`; do not add a scheme or port. If the CA
path is not configured, the Connector does not load a custom CA and uses the
corresponding client's default trust store.
+
+Distribute the CA file according to the Kafka Connect deployment mode:
+
+- **Standalone**: Place the CA file on the Worker host and configure its local
path.
+- **Distributed**: Place the CA file at the same local path on every Worker
because a task may be assigned to any Worker.
+- **Kubernetes**: Mount the same Secret or volume into every Kafka Connect
Worker Pod and configure the mounted path.
+
+### Use S3 TVF write mode
+
+TVF write mode stores data as JSON in S3-compatible object storage and then
writes it to Doris through S3 TVF. Ensure that Kafka Connect and Doris can
access the object storage and that the target Doris table already exists. Add
the following options to the Connector configuration:
+
+```json
+{
+"load.model":"tvf",
+"enable.combine.flush":"true",
+"delivery.guarantee":"at_least_once",
+"sink.properties.columns":"id,name,age",
+"sink.s3.endpoint":"$YOUR_S3_ENDPOINT",
+"sink.s3.region":"$YOUR_S3_REGION",
+"sink.s3.bucket":"$YOUR_S3_BUCKET",
+"sink.s3.prefix":"$YOUR_S3_PREFIX",
+"sink.s3.access-key":"$YOUR_S3_ACCESS_KEY",
+"sink.s3.secret-key":"$YOUR_S3_SECRET_KEY"
+}
+```
+
+`sink.properties.columns` must list the target table columns in write order.
The Connector does not automatically delete staged objects; configure an object
storage lifecycle policy as needed.
+
## FAQ
### Reading JSON data reports `JsonConverter with schemas.enable requires
"schema" and "payload" fields`
diff --git
a/docs/connection-integration/data-integration/flink-doris-connector.md
b/docs/connection-integration/data-integration/flink-doris-connector.md
index 1e3bcff05c6..078ab96ffaa 100644
--- a/docs/connection-integration/data-integration/flink-doris-connector.md
+++ b/docs/connection-integration/data-integration/flink-doris-connector.md
@@ -34,6 +34,7 @@ Main capabilities include:
| 25.1.0 | 1.15 - 1.20 | 1.0+ | 8 |
- |
| 26.0.0 | 1.15 - 1.20,2.0 - 2.2 | 1.0+ | 8(1.x),17(2.x) |
- |
| 26.1.1 | 1.15 - 1.20,2.0 - 2.2 | 1.0+ | 8(1.x),17(2.x) |
- |
+| 26.2.0 | 1.15 - 1.20,2.0 - 2.2 | 1.0+ | 8(1.x),17(2.x) |
- |
## Installation
@@ -64,7 +65,7 @@ For example:
<dependency>
<groupId>org.apache.doris</groupId>
<artifactId>flink-doris-connector-1.16</artifactId>
- <version>25.1.0</version>
+ <version>26.2.0</version>
</dependency>
```
@@ -198,7 +199,7 @@ mysql> select * from test.student_trans;
### Case 1: Reading Doris Data
-When Flink reads Doris data, the Doris Source is a bounded stream and does not
support continuous reading via CDC. The following two read protocols are
supported:
+By default, the Doris Source reads a bounded snapshot. It can also
continuously read row-level changes from Doris Binlog by using an incremental
scan mode. The following two read protocols are supported:
| Protocol | Description
| Recommended Version |
| -------------- |
------------------------------------------------------------ |
------------------- |
@@ -248,6 +249,98 @@ WITH (
SELECT * FROM student;
```
+#### Incremental Reading with Doris Binlog
+
+For a Doris table with ROW-format Binlog enabled, Flink Doris Connector can
continuously read row-level changes. In `initial` mode, the Connector first
reads the current table snapshot and then seamlessly switches to incremental
reading.
+
+First, enable row-format Binlog on the Doris source table.
`binlog.need_historical_value` is required when the consumer needs the
before-image of updated rows:
+
+```sql
+CREATE DATABASE IF NOT EXISTS test;
+
+CREATE TABLE test.student_binlog_source (
+ id INT,
+ name VARCHAR(50),
+ age INT
+)
+UNIQUE KEY(id)
+DISTRIBUTED BY HASH(id) BUCKETS 1
+PROPERTIES (
+ "replication_num" = "1",
+ "binlog.enable" = "true",
+ "binlog.format" = "ROW",
+ "binlog.need_historical_value" = "true",
+ "binlog.ttl_seconds" = "86400"
+);
+
+INSERT INTO test.student_binlog_source VALUES (1, 'Alice', 18);
+```
+
+Then enable Flink Checkpoint and create the Doris source table:
+
+```sql
+SET 'execution.checkpointing.interval' = '10s';
+
+CREATE TABLE student_binlog (
+ id INT,
+ name STRING,
+ age INT
+) WITH (
+ 'connector' = 'doris',
+ 'fenodes' = '127.0.0.1:8030',
+ 'table.identifier' = 'test.student_binlog_source',
+ 'username' = 'root',
+ 'password' = '',
+ 'source.scan.mode' = 'initial'
+);
+
+SELECT * FROM student_binlog;
+```
+
+After the job starts, changes to `test.student_binlog_source` are continuously
emitted as Flink changelog records. Select the startup mode with
`source.scan.mode`:
+
+| Mode | Behavior |
+| ---- | -------- |
+| `snapshot` | Reads the current snapshot and stops. This is the default mode.
|
+| `initial` | Reads the current snapshot and switches to continuous Binlog
reading when the snapshot is complete. |
+| `latest` | Skips the snapshot and reads changes generated after the job
starts. |
+| `from-timestamp` | Skips the snapshot and reads changes after the exclusive
start time specified by `source.scan.timestamp` in `yyyy-MM-dd HH:mm:ss`
format. |
+
+By default, the Connector emits full row changes in `detail` mode. Set
`source.binlog.increment-type` to `min_delta` for the minimal change set or
`append_only` for append events only.
+
+Note the following:
+
+- Incremental reading uses Arrow Flight SQL. The Connector enables it by
default and automatically discovers its port.
+- Enable Flink Checkpoint.
+- Configure Doris Binlog retention to cover the maximum expected job downtime.
If the required Binlog data has expired, restart from a new snapshot or specify
a new start time.
+
+##### Publishing Consumption Progress to Doris (Optional)
+
+Consumption progress is stored in Flink Checkpoints by default. To also query
the progress in Doris, create the following offset table:
+
+```sql
+CREATE DATABASE IF NOT EXISTS ops;
+
+CREATE TABLE ops.flink_source_offsets (
+ consumer_id VARCHAR(256) NOT NULL,
+ offset_timestamp DATETIME NOT NULL,
+ update_time DATETIMEV2(3) NOT NULL
+)
+UNIQUE KEY(consumer_id)
+DISTRIBUTED BY HASH(consumer_id) BUCKETS 1
+PROPERTIES (
+ "replication_num" = "1"
+);
+```
+
+```sql
+'jdbc-url' = 'jdbc:mysql://127.0.0.1:9030',
+'source.binlog.offset-table' = 'ops.flink_source_offsets',
+'source.binlog.consumer-id' = 'student-sync'
+```
+
+`source.binlog.consumer-id` identifies the consumer job and should remain
unchanged when the same job restarts.
+
#### Reading via DataStream API
When reading data via the DataStream API, you must add the dependency to the
project's POM file in advance. See the [Installation](#installation) section.
@@ -793,6 +886,10 @@ After the Flink cluster is started, you can run the
corresponding command accord
| doris.request.retries | 3 | N | Number of retries
for sending requests to Doris |
| doris.request.connect.timeout | 30s | N | Connection
timeout for sending requests to Doris |
| doris.request.read.timeout | 30s | N | Read timeout for
sending requests to Doris |
+| doris.enable.tls | FALSE | N | Whether to enable
TLS for Doris HTTP, MySQL/JDBC, BE Thrift, and Arrow Flight SQL connections. |
+| doris.tls.ca-certificate-path | -- | N | Local path to a
PEM CA certificate chain. When empty, the Connector does not load a custom CA
and uses the corresponding client's default trust store. |
+| doris.tls.skip-hostname-verification | FALSE | N | Whether to skip
server hostname verification while retaining CA validation. |
+| doris.tls.excluded-protocols | -- | N | Comma-separated
protocols that remain plaintext while TLS is enabled. Supported values: `http`,
`mysql`, `thrift`, and `arrowflight`. |
### Source Configuration
@@ -804,6 +901,12 @@ After the Flink cluster is started, you can run the
corresponding command accord
| doris.exec.mem.limit | 8192mb | N | Memory limit for a
single query. The default is 8GB, in bytes.
|
| source.use-flight-sql | TRUE | N | Whether to use
Arrow Flight SQL for reading
|
| source.flight-sql-port | - | N | When using Arrow
Flight SQL for reading, the FE's `arrow_flight_sql_port`
|
+| source.scan.mode | snapshot | N | Source startup
mode. Supported values: `snapshot`, `initial`, `latest`, and `from-timestamp`.
|
+| source.scan.timestamp | -- | N | Exclusive start
time in `yyyy-MM-dd HH:mm:ss` format. Required only for `from-timestamp`.
|
+| source.binlog.increment-type | detail | N | Binlog change type:
`detail`, `min_delta`, or `append_only`.
|
+| source.binlog.poll-interval | 10s | N | Interval for
polling new Binlog data. The minimum value is 1 second.
|
+| source.binlog.offset-table | -- | N | Doris table in
`database.table` format used to publish offsets covered by completed
Checkpoints. Configure with `source.binlog.consumer-id` and `jdbc-url`. |
+| source.binlog.consumer-id | -- | N | Stable consumer
identifier written to `source.binlog.offset-table`.
|
**DataStream-Specific Configuration**
@@ -816,10 +919,18 @@ After the Flink cluster is started, you can run the
corresponding command accord
| Key | Default Value | Required | Comment
[...]
| --------------------------- | ------------- | -------- |
-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
[...]
-| sink.label-prefix | -- | Y | The label prefix
used for Stream Load imports. In 2pc scenarios, it must be globally unique to
guarantee the EOS semantics of Flink.
[...]
+| sink.label-prefix | -- | Y | The label prefix
used for imports. In 2PC scenarios, it must be globally unique to guarantee
Flink EOS semantics.
[...]
| sink.properties.* | -- | N | Stream Load import
parameters. For example: `'sink.properties.column_separator' = ', '` defines
the column separator; `'sink.properties.escape_delimiters' = 'true'` indicates
that special characters are used as separators, and `\x01` will be converted to
the binary `0x01`; for JSON-format imports: `'sink.properties.format' =
'json'`, `'sink.properties.read_json_by_line' = 'true'`. For detailed
parameters, see [Stream Load](../../ [...]
| sink.enable-delete | TRUE | N | Whether to enable
deletion. This option requires the Doris table to have batch deletion enabled
(enabled by default in Doris 0.15+) and only supports the Unique model.
[...]
| sink.enable-2pc | TRUE | N | Whether to enable
two-phase commit (2pc). The default is true, which guarantees Exactly-Once
semantics. For information on two-phase commit, see [Stream Load
2PC](../../data-operate/transaction.md#streamload-2pc).
[...]
+| sink.write-mode | STREAM_LOAD | N | Write mode.
Supported values: `STREAM_LOAD`, `STREAM_LOAD_BATCH`, and `TVF`.
[...]
+| sink.s3.endpoint | -- | TVF write mode only |
S3-compatible object-storage endpoint.
[...]
+| sink.s3.region | -- | TVF write mode only |
Object-storage region.
[...]
+| sink.s3.bucket | -- | TVF write mode only | Bucket
used to stage files.
[...]
+| sink.s3.prefix | -- | TVF write mode only | Object
key prefix for staged files. The prefix cannot contain glob characters.
[...]
+| sink.s3.access-key | -- | TVF write mode only |
Object-storage access key.
[...]
+| sink.s3.secret-key | -- | TVF write mode only |
Object-storage secret key.
[...]
+| sink.s3.path-style-access | FALSE | N | Whether TVF mode
uses path-style object-storage access.
[...]
| sink.buffer-size | 1MB | N | Buffer size for the
write data cache, in bytes. Modifying this is not recommended; the default
configuration is sufficient.
[...]
| sink.buffer-count | 3 | N | Number of write
data cache buffers. Modifying this is not recommended; the default
configuration is sufficient.
[...]
| sink.max-retries | 3 | N | The maximum number
of retries after a Commit failure. The default is 3.
[...]
@@ -1101,6 +1212,72 @@ For this scenario, you typically need to write a
DataStream API program and use
In the full-database synchronization tool provided by Connector, no additional
configuration is required; upstream DDL is automatically synchronized and
Schema Change operations are performed in Doris.
+### Accessing a TLS-Enabled Doris Cluster from Flink
+
+The Connector can enable TLS for Doris HTTP and Stream Load, MySQL/JDBC, BE
Thrift, and Arrow Flight SQL connections. TLS is disabled by default, and the
Connector verifies the Doris server certificate.
+
+Add the following options to a Doris Source, Sink, or Catalog configuration:
+
+```sql
+'doris.enable.tls' = 'true',
+'doris.tls.ca-certificate-path' = '/etc/doris-tls/ca-chain.pem'
+```
+
+When specifying a CA file with `doris.tls.ca-certificate-path`, use a PEM
certificate chain and ensure that every Flink process connecting to Doris can
read it from the local filesystem. When this path is not configured, the
Connector does not load a custom CA and uses the corresponding client's default
trust store. Keep hostname verification enabled in production.
+
+If a Doris protocol intentionally remains plaintext, exclude only that
protocol. Supported values are `http`, `mysql`, `thrift`, and `arrowflight`:
+
+```sql
+'doris.tls.excluded-protocols' = 'arrowflight'
+```
+
+The Connector does not probe protocols or fall back to plaintext after a TLS
failure.
+
+When `doris.enable.tls` is enabled, the Connector passes TLS settings through
JDBC connection properties without modifying `jdbc-url`. Therefore, specify
only the connection address in `jdbc-url`; do not add TLS parameters such as
`sslMode`, `useSSL`, or trust store settings.
+
+Arrow Flight SQL supports TLS but does not support skipping hostname
verification only. If `doris.tls.skip-hostname-verification` is set to `true`,
exclude `arrowflight` through `doris.tls.excluded-protocols`.
+
+Distribute the CA file according to the Flink deployment mode:
+
+- **Standalone**: place the file at the same path on all JobManager,
TaskManager, and SQL Gateway hosts that connect to Doris.
+- **YARN**: localize the file with `yarn.ship-files: /local/path/ca.pem`, and
set `doris.tls.ca-certificate-path` to the container-localized file name, such
as `ca.pem`.
+- **Kubernetes**: mount a ConfigMap or Secret at the same path in the relevant
JobManager and TaskManager pods.
+
+### Writing with S3 TVF
+
+TVF write mode first stages data in JSON format in S3 object storage and then
imports it into Doris through the S3 table-valued function. Use this mode when
object storage is the preferred data transfer path or when the Stream Load
network path is unavailable.
+
+Before use, ensure that both Flink and Doris can access S3 and that the target
table already exists in Doris.
+
+```sql
+SET 'execution.checkpointing.interval' = '30s';
+
+CREATE TABLE student_tvf_sink (
+ id INT,
+ name STRING,
+ age INT
+) WITH (
+ 'connector' = 'doris',
+ 'fenodes' = '127.0.0.1:8030',
+ 'jdbc-url' = 'jdbc:mysql://127.0.0.1:9030',
+ 'table.identifier' = 'test.student_tvf',
+ 'username' = 'root',
+ 'password' = '',
+ 'sink.write-mode' = 'TVF',
+ 'sink.label-prefix' = 'student_tvf',
+ 'sink.s3.endpoint' = 'https://s3.example.com',
+ 'sink.s3.region' = 'us-east-1',
+ 'sink.s3.bucket' = 'staging-bucket',
+ 'sink.s3.prefix' = 'doris/student',
+ 'sink.s3.access-key' = 'access-key',
+ 'sink.s3.secret-key' = 'secret-key'
+);
+
+INSERT INTO student_tvf_sink VALUES (1, 'Alice', 18);
+```
+
+The Connector does not automatically delete staged objects from S3. Configure
an object-storage lifecycle policy as needed.
+
## FAQ
**1. errCode = 2, detailMessage = Label [label_0_1] has already been used,
relate to txn [19650]**
diff --git
a/docs/connection-integration/data-integration/spark-doris-connector.md
b/docs/connection-integration/data-integration/spark-doris-connector.md
index 18d36a37aa2..fb00ace2d42 100644
--- a/docs/connection-integration/data-integration/spark-doris-connector.md
+++ b/docs/connection-integration/data-integration/spark-doris-connector.md
@@ -42,6 +42,7 @@ First select the corresponding Connector version based on
your Spark, Doris, Jav
| Connector | Spark | Doris | Java | Scala |
| --- | --- | --- | --- | --- |
+| 26.1.0 | 4.1, 3.5 - 3.1, 2.4 | 1.0 + | 17 (4.1), 8 (3.x, 2.x) | 2.13 (4.1),
2.12 (3.x), 2.11 (2.x) |
| 26.0.0 | 3.5 - 3.1, 2.4 | 1.0 + | 8 | 2.12, 2.11 |
| 25.2.0 | 3.5 - 3.1, 2.4 | 1.0 + | 8 | 2.12, 2.11 |
| 25.1.0 | 3.5 - 3.1, 2.4 | 1.0 + | 8 | 2.12, 2.11 |
@@ -62,7 +63,7 @@ Add the Spark Doris Connector dependency in your project
`pom.xml`, and replace
<dependency>
<groupId>org.apache.doris</groupId>
<artifactId>spark-doris-connector-spark-3.5</artifactId>
- <version>25.2.0</version>
+ <version>26.1.0</version>
</dependency>
```
@@ -82,22 +83,22 @@ You can also download the Jar file of the corresponding
version from the [Maven
If you need to compile the source code yourself, run `sh build.sh` in the
source directory and enter the required Scala and Spark versions when prompted.
-After successful compilation, the target Jar file is generated in the `dist`
directory, for example `spark-doris-connector-spark-3.5-25.2.0.jar`. Copy this
file into the Spark `classpath` to start using the Spark Doris Connector:
+After successful compilation, the target Jar file is generated in the `dist`
directory, for example `spark-doris-connector-spark-3.5-26.1.0.jar`. Copy this
file into the Spark `classpath` to start using the Spark Doris Connector:
| Spark run mode | How to place the Jar file |
| --- | --- |
| Local mode | Place the Jar file in the `jars/` directory. |
| Yarn cluster mode | Place the Jar file in the pre-deployed package. |
-For example, upload `spark-doris-connector-spark-3.5-25.2.0.jar` to HDFS and
add the dependency through `spark.yarn.jars`:
+For example, upload `spark-doris-connector-spark-3.5-26.1.0.jar` to HDFS and
add the dependency through `spark.yarn.jars`:
```shell
-# 1. Upload spark-doris-connector-spark-3.5-25.2.0.jar to HDFS
+# 1. Upload spark-doris-connector-spark-3.5-26.1.0.jar to HDFS
hdfs dfs -mkdir /spark-jars/
-hdfs dfs -put /your_local_path/spark-doris-connector-spark-3.5-25.2.0.jar
/spark-jars/
+hdfs dfs -put /your_local_path/spark-doris-connector-spark-3.5-26.1.0.jar
/spark-jars/
-# 2. Add the spark-doris-connector-spark-3.5-25.2.0.jar dependency in the
cluster
-spark.yarn.jars=hdfs:///spark-jars/spark-doris-connector-spark-3.5-25.2.0.jar
+# 2. Add the spark-doris-connector-spark-3.5-26.1.0.jar dependency in the
cluster
+spark.yarn.jars=hdfs:///spark-jars/spark-doris-connector-spark-3.5-26.1.0.jar
```
## Scenario 1: Batch read Doris data
@@ -458,13 +459,21 @@ A Java version of the example is available under
`samples/doris-demo/spark-demo/
| `doris.sink.batch.interval.ms` | 0 | Interval between sink batches, in ms. |
| `doris.sink.enable-2pc` | false | Whether to enable two-phase commit. When
enabled, transactions are committed at the end of the job, and if some tasks
fail, all transactions in pre-commit state are rolled back. |
| `doris.sink.auto-redirect` | true | Whether to redirect Stream Load
requests. When enabled, Stream Load is written through the FE, and BE
information is no longer fetched explicitly. |
-| `doris.enable.https` | false | Whether to enable FE HTTPS requests. |
-| `doris.https.key-store-path` | - | HTTPS key store path. |
-| `doris.https.key-store-type` | JKS | HTTPS key store type. |
-| `doris.https.key-store-password` | - | HTTPS key store password. |
+| `doris.enable.tls` | false | Whether to enable TLS for Doris connections. |
+| `doris.tls.ca-certificate-path` | Empty string | Path to a PEM CA
certificate chain. When empty, the Connector does not load a custom CA and uses
the corresponding client's default trust store. |
+| `doris.tls.skip-hostname-verification` | false | Whether to skip hostname
verification while retaining CA verification. Arrow Flight SQL does not support
this option. |
+| `doris.tls.excluded-protocols` | Empty string | Comma-separated protocols
that do not use TLS. Available values are `http`, `mysql`, `thrift`, and
`arrowflight`. |
| `doris.read.mode` | thrift | Doris read mode. Available options are `thrift`
and `arrow`. |
| `doris.read.arrow-flight-sql.port` | - | Arrow Flight SQL port of the Doris
FE. When `doris.read.mode` is `arrow`, this is used to read data through Arrow
Flight SQL. For server-side configuration, refer to [High-speed data
transmission link based on Arrow Flight SQL](../arrow-flight-sql.md). |
-| `doris.sink.label.prefix` | spark-doris | Import label prefix when writing
through Stream Load. |
+| `doris.sink.mode` | stream_load | Doris write mode. Supported values are
`stream_load` and `tvf`. |
+| `doris.sink.label.prefix` | spark-doris | Label prefix for Doris writes. |
+| `doris.sink.s3.endpoint` | - | S3-compatible object storage endpoint.
Required in TVF write mode. |
+| `doris.sink.s3.region` | - | Object storage region. Required in TVF write
mode. |
+| `doris.sink.s3.bucket` | - | Bucket used to stage data. Required in TVF
write mode. |
+| `doris.sink.s3.prefix` | - | Object key prefix used to stage data. Required
in TVF write mode. |
+| `doris.sink.s3.access-key` | - | Object storage access key. Required in TVF
write mode. |
+| `doris.sink.s3.secret-key` | - | Object storage secret key. Required in TVF
write mode. |
+| `doris.sink.s3.path-style-access` | false | Whether to use path-style object
storage access in TVF write mode. |
| `doris.thrift.max.message.size` | 2147483647 | Maximum message size when
reading data through Thrift. |
| `doris.fe.auto.fetch` | false | Whether to automatically fetch FE
information. When set to `true`, all FE node information is fetched based on
the nodes configured in `doris.fenodes`, so there is no need to configure
multiple nodes or to configure `doris.read.arrow-flight-sql.port` and
`doris.query.port` separately. |
| `doris.read.bitmap-to-string` | false | Whether to convert the Bitmap type
to a string composed of array indexes when reading. For the result format,
refer to the function definition
[BITMAP_TO_STRING](../../sql-manual/sql-functions/scalar-functions/bitmap-functions/bitmap-to-string.md).
|
@@ -545,6 +554,54 @@ Starting from version 24.0.0, the read return type for the
Bitmap type is string
:::
+## Best practices
+
+### Access a TLS-enabled Doris environment from Spark
+
+Set `doris.enable.tls` to `true` when Spark reads from or writes to a
TLS-enabled Doris cluster:
+
+```scala
+mockDataDF.write.format("doris")
+ .option("doris.table.identifier",
"$YOUR_DORIS_DATABASE_NAME.$YOUR_DORIS_TABLE_NAME")
+ .option("doris.fenodes",
"$YOUR_DORIS_FE_HOSTNAME:$YOUR_DORIS_FE_HTTPS_PORT")
+ .option("user", "$YOUR_DORIS_USERNAME")
+ .option("password", "$YOUR_DORIS_PASSWORD")
+ .option("doris.enable.tls", "true")
+ .option("doris.tls.ca-certificate-path", "/etc/doris-tls/ca.pem")
+ .save()
+```
+
+The CA file must be available to the Spark driver and every executor. If
`doris.tls.ca-certificate-path` is not configured, the Connector does not load
a custom CA and uses the corresponding client's default trust store.
+
+Distribute the CA file according to the Spark deployment mode:
+
+- **YARN**: Add `--files /local/path/ca.pem` to `spark-submit`, and set
`doris.tls.ca-certificate-path` to `ca.pem`.
+- **Kubernetes**: Mount the same Secret or volume into both the driver and
executor Pods. For example, add `--conf
spark.kubernetes.driver.secrets.doris-tls=/etc/doris-tls` and `--conf
spark.kubernetes.executor.secrets.doris-tls=/etc/doris-tls` to `spark-submit`,
and set the CA path to `/etc/doris-tls/ca.pem`.
+
+### Use S3 TVF write mode
+
+TVF write mode stores data as JSON in S3-compatible object storage and then
writes it to Doris through S3 TVF. It supports batch writes only. Ensure that
both Spark and Doris can access the object storage and that the target Doris
table already exists.
+
+```scala
+mockDataDF.write.format("doris")
+ .option("doris.table.identifier",
"$YOUR_DORIS_DATABASE_NAME.$YOUR_DORIS_TABLE_NAME")
+ .option("doris.fenodes",
"$YOUR_DORIS_FE_HOSTNAME:$YOUR_DORIS_FE_HTTP_PORT")
+ .option("doris.query.port", "$YOUR_DORIS_FE_QUERY_PORT")
+ .option("user", "$YOUR_DORIS_USERNAME")
+ .option("password", "$YOUR_DORIS_PASSWORD")
+ .option("doris.sink.mode", "tvf")
+ .option("doris.sink.s3.endpoint", "$YOUR_S3_ENDPOINT")
+ .option("doris.sink.s3.region", "$YOUR_S3_REGION")
+ .option("doris.sink.s3.bucket", "$YOUR_S3_BUCKET")
+ .option("doris.sink.s3.prefix", "$YOUR_S3_PREFIX")
+ .option("doris.sink.s3.access-key", "$YOUR_S3_ACCESS_KEY")
+ .option("doris.sink.s3.secret-key", "$YOUR_S3_SECRET_KEY")
+ .mode(SaveMode.Append)
+ .save()
+```
+
+The Connector does not automatically delete staged objects. Configure an
object storage lifecycle policy as needed.
+
## FAQ and troubleshooting
<!-- Knowledge type: FAQ + Troubleshooting -->
diff --git
a/i18n/zh-CN/docusaurus-plugin-content-docs/current/connection-integration/data-integration/doris-kafka-connector.md
b/i18n/zh-CN/docusaurus-plugin-content-docs/current/connection-integration/data-integration/doris-kafka-connector.md
index 753c87d3824..7c15cb9ce32 100644
---
a/i18n/zh-CN/docusaurus-plugin-content-docs/current/connection-integration/data-integration/doris-kafka-connector.md
+++
b/i18n/zh-CN/docusaurus-plugin-content-docs/current/connection-integration/data-integration/doris-kafka-connector.md
@@ -42,6 +42,7 @@
| 24.0.0 | 2.4+ | 2.0+ | 8 |
| 25.0.0 | 2.4+ | 2.0+ | 8 |
| 26.0.0 | 2.4+ | 2.0+ | 8 |
+| 26.1.0 | 2.4+ | 2.0+ | 8 |
## 准备 Doris Kafka Connector
@@ -59,7 +60,7 @@
<dependency>
<groupId>org.apache.doris</groupId>
<artifactId>doris-kafka-connector</artifactId>
- <version>25.0.0</version>
+ <version>26.1.0</version>
</dependency>
```
@@ -251,7 +252,12 @@ errors.deadletterqueue.topic.replication.factor=1
| `doris.user` | - | - | Y | Doris 用户名。 |
| `doris.password` | - | - | Y | Doris 密码。 |
| `doris.database` | - | - | Y | 要写入的数据库。多个库时可以为空，同时需要在 `topic2table.map`
中配置具体库名。 |
+| `doris.enable.tls` | `true`,<br />`false` | false | N | 是否为 Doris HTTP 和
MySQL 连接启用 TLS。 |
+| `doris.tls.ca-certificate-path` | - | 空字符串 | N | Kafka Connect Worker 本地 PEM
格式的 CA 证书链路径。未配置时 Connector 不加载自定义 CA，使用对应客户端的默认信任库。 |
+| `doris.tls.skip-hostname-verification` | `true`,<br />`false` | false | N |
是否在保留 CA 验证的同时跳过主机名验证。 |
+| `doris.tls.excluded-protocols` | `http`,<br />`mysql` | 空字符串 | N | 不使用 TLS 的
Doris 协议，多个协议使用逗号分隔。 |
| `doris.topic2table.map` | - | - | Y | Topic 和表的对应关系，例如
`topic1:tb1,topic2:tb2`。如果留空，默认将 Topic 名称作为写入的表名。多个库的格式为
`topic1:db1.tbl1,topic2:db2.tbl2`。 |
+| `load.model` | `stream_load`,<br />`tvf` | stream_load | N | Doris 写入模式。设置为
`tvf` 时通过 S3 TVF 写入。 |
| `buffer.count.records` | - | 50000 | N | 单次 Stream Load 写入的条数。 |
| `buffer.flush.time` | - | 120 | N | Buffer 刷新间隔，单位为秒，默认值为 120 秒。 |
| `buffer.size.bytes` | - | 104857600(100MB) | N | 单次 Stream Load 写入的数据大小。 |
@@ -260,6 +266,13 @@ errors.deadletterqueue.topic.replication.factor=1
| `label.prefix` | - | `${name}` | N | Stream Load 导入数据时的 Label 前缀。默认值为
Connector 应用名称。 |
| `auto.redirect` | - | true | N | 是否重定向 Stream Load 请求。开启后，Stream Load 会通过 FE
重定向到需要写入数据的 BE，并且不再显示获取 BE 信息。 |
| `sink.properties.*` | - | `'sink.properties.format':'json'`,<br
/>`'sink.properties.read_json_by_line':'true'` | N | Stream Load 的导入参数。例如，通过
`'sink.properties.column_separator':','` 定义列分隔符。详细参数请参考 [Stream Load
手册](../../data-operate/import/import-way/stream-load-manual.md)。<br /><br />开启
Group Commit 时，例如开启 `sync_mode`
模式：`"sink.properties.group_commit":"sync_mode"`。Group Commit 可以配置
`off_mode`、`sync_mode`、`async_mode` 三种模式，具体使用请参考 [Group
Commit](https://doris.apache.org/docs/data-oper [...]
+| `sink.s3.endpoint` | - | - | 仅 TVF 写入模式 | 兼容 S3 的对象存储 Endpoint，必须为完整的 HTTP 或
HTTPS 地址。 |
+| `sink.s3.region` | - | - | 仅 TVF 写入模式 | 对象存储 Region。 |
+| `sink.s3.bucket` | - | - | 仅 TVF 写入模式 | 暂存数据的 Bucket。 |
+| `sink.s3.prefix` | - | - | 仅 TVF 写入模式 | 暂存数据的对象路径前缀。 |
+| `sink.s3.access-key` | - | - | 仅 TVF 写入模式 | 对象存储 Access Key。 |
+| `sink.s3.secret-key` | - | - | 仅 TVF 写入模式 | 对象存储 Secret Key。 |
+| `sink.s3.path-style-access` | `true`,<br />`false` | false | N | TVF
写入模式下是否使用路径风格访问对象存储。 |
| `delivery.guarantee` | `at_least_once`,<br />`exactly_once` | at_least_once
| N | 消费 Kafka 数据并导入 Doris 时的数据一致性保障方式。支持 `at_least_once` 和 `exactly_once`，默认值为
`at_least_once`。Doris 需要升级至 2.1.0 以上，才能保障数据的 `exactly_once`。 |
| `converter.mode` | `normal`,<br />`debezium_ingestion` | normal | N |
Connector 消费 Kafka 数据时的上游数据类型转换模式。`normal` 表示正常消费 Kafka
中的数据，不进行特殊类型转换；`debezium_ingestion` 表示当 Kafka 上游数据通过 Debezium 等 CDC（Change Data
Capture，变更数据捕获）工具采集时，需要进行特殊类型转换。 |
| `debezium.schema.evolution` | `none`,<br />`basic` | none | N | 通过 Debezium
采集上游数据库系统（如 MySQL）时，如果发生结构变更，可以将增加的字段同步到 Doris 中。`none`
表示上游数据库系统发生结构变更时，不同步变更后的结构到 Doris 中。`basic`
表示同步上游数据库的数据变更操作。由于列结构变更是危险操作，可能导致误删 Doris
表结构中的列，目前仅支持同步上游增加列。当列被重命名后，旧列保持原样，Connector 会在目标表中新增一列，并将重命名后的新增数据 Sink 到新列中。 |
@@ -558,6 +571,51 @@ curl -i http://127.0.0.1:8083/connectors -H "Content-Type:
application/json" -X
其中，`repo` 为 `InsertField` 增加的静态字段，`registertime` 为 `TimestampConverter`
转换后的时间字符串。更多 Kafka Connect Single Message Transforms (SMT) 使用案例，请参考 [SMT
documentation](https://docs.confluent.io/cloud/current/connectors/transforms/overview.html)。
+## 最佳实践
+
+### Kafka Connect 访问启用 TLS 的 Doris 环境
+
+向启用 TLS 的 Doris 集群写入数据时，在 Connector 配置中增加以下参数：
+
+```json
+{
+"doris.urls":"doris-fe.example.com",
+"doris.http.port":"8040",
+"doris.query.port":"9030",
+"doris.enable.tls":"true",
+"doris.tls.ca-certificate-path":"/etc/kafka-connect/certs/doris-ca.pem"
+}
+```
+
+`doris.urls` 中只填写主机名，不要添加协议或端口。未配置 CA 路径时，Connector 不加载自定义 CA，使用对应客户端的默认信任库。
+
+根据 Kafka Connect 部署模式分发 CA 文件：
+
+- **Standalone**：将 CA 文件放置在 Worker 主机上，并配置其本地路径。
+- **Distributed**：Task 可能被分配到任意 Worker，需要将 CA 文件放置在每个 Worker 的相同本地路径。
+- **Kubernetes**：通过 Secret 或 Volume 将 CA 挂载到所有 Kafka Connect Worker Pod
的相同路径，并配置挂载后的文件路径。
+
+### 使用 S3 TVF 写入
+
+TVF 写入模式先将数据以 JSON 形式存到兼容 S3 的对象存储，再通过 S3 TVF 写入 Doris。请确保 Kafka Connect 和
Doris 均可访问对象存储，并已创建 Doris 目标表。在 Connector 配置中增加以下参数：
+
+```json
+{
+"load.model":"tvf",
+"enable.combine.flush":"true",
+"delivery.guarantee":"at_least_once",
+"sink.properties.columns":"id,name,age",
+"sink.s3.endpoint":"$YOUR_S3_ENDPOINT",
+"sink.s3.region":"$YOUR_S3_REGION",
+"sink.s3.bucket":"$YOUR_S3_BUCKET",
+"sink.s3.prefix":"$YOUR_S3_PREFIX",
+"sink.s3.access-key":"$YOUR_S3_ACCESS_KEY",
+"sink.s3.secret-key":"$YOUR_S3_SECRET_KEY"
+}
+```
+
+`sink.properties.columns` 需要按照写入顺序列出 Doris 目标表字段。Connector
不会自动删除暂存对象，请按需配置对象存储生命周期策略。
+
## 常见问题
### 读取 JSON 类型数据时报 `JsonConverter with schemas.enable requires "schema" and
"payload" fields`
diff --git
a/i18n/zh-CN/docusaurus-plugin-content-docs/current/connection-integration/data-integration/flink-doris-connector.md
b/i18n/zh-CN/docusaurus-plugin-content-docs/current/connection-integration/data-integration/flink-doris-connector.md
index cc7d0deb5f9..92b7f63493f 100644
---
a/i18n/zh-CN/docusaurus-plugin-content-docs/current/connection-integration/data-integration/flink-doris-connector.md
+++
b/i18n/zh-CN/docusaurus-plugin-content-docs/current/connection-integration/data-integration/flink-doris-connector.md
@@ -34,6 +34,7 @@
| 25.1.0 | 1.15 - 1.20 | 1.0+ | 8 |
- |
| 26.0.0 | 1.15 - 1.20,2.0 - 2.2 | 1.0+ | 8(1.x),17(2.x) |
- |
| 26.1.1 | 1.15 - 1.20,2.0 - 2.2 | 1.0+ | 8(1.x),17(2.x) |
- |
+| 26.2.0 | 1.15 - 1.20,2.0 - 2.2 | 1.0+ | 8(1.x),17(2.x) |
- |
## 安装方式
@@ -64,7 +65,7 @@
<dependency>
<groupId>org.apache.doris</groupId>
<artifactId>flink-doris-connector-1.16</artifactId>
- <version>25.1.0</version>
+ <version>26.2.0</version>
</dependency>
```
@@ -198,7 +199,7 @@ mysql> select * from test.student_trans;
### 场景一：读取 Doris 数据
-Flink 读取 Doris 数据时，Doris Source 是有界流，不支持以 CDC 的方式持续读取。支持以下两种读取协议：
+默认情况下，Doris Source 读取有界快照。通过增量扫描模式，还可以持续读取 Doris Binlog 中的行级变更。支持以下两种读取协议：
| 协议 | 说明 |
推荐版本 |
| -------------- |
------------------------------------------------------------ |
------------------ |
@@ -248,6 +249,98 @@ WITH (
SELECT * FROM student;
```
+#### 使用 Doris Binlog 增量读取
+
+对于使用 Doris 开启 ROW 格式 Binlog 的表，Flink Doris Connector 可以持续读取行级数据变更。使用 `initial`
模式时，Connector 会先读取当前表快照，再无缝切换到增量变更读取。
+
+首先，在 Doris 源表上启用 ROW 格式的 Binlog。如果消费端需要更新前的行数据，还需要启用
`binlog.need_historical_value`：
+
+```sql
+CREATE DATABASE IF NOT EXISTS test;
+
+CREATE TABLE test.student_binlog_source (
+ id INT,
+ name VARCHAR(50),
+ age INT
+)
+UNIQUE KEY(id)
+DISTRIBUTED BY HASH(id) BUCKETS 1
+PROPERTIES (
+ "replication_num" = "1",
+ "binlog.enable" = "true",
+ "binlog.format" = "ROW",
+ "binlog.need_historical_value" = "true",
+ "binlog.ttl_seconds" = "86400"
+);
+
+INSERT INTO test.student_binlog_source VALUES (1, 'Alice', 18);
+```
+
+然后启用 Flink Checkpoint 并创建 Doris Source 表：
+
+```sql
+SET 'execution.checkpointing.interval' = '10s';
+
+CREATE TABLE student_binlog (
+ id INT,
+ name STRING,
+ age INT
+) WITH (
+ 'connector' = 'doris',
+ 'fenodes' = '127.0.0.1:8030',
+ 'table.identifier' = 'test.student_binlog_source',
+ 'username' = 'root',
+ 'password' = '',
+ 'source.scan.mode' = 'initial'
+);
+
+SELECT * FROM student_binlog;
+```
+
+任务启动后，对 `test.student_binlog_source` 的修改会以 Flink Changelog 形式持续输出。通过
`source.scan.mode` 选择启动模式：
+
+| 模式 | 行为 |
+| ---- | ---- |
+| `snapshot` | 读取当前快照后结束，为默认模式。 |
+| `initial` | 先读取当前快照，快照读取完成后切换到持续 Binlog 读取。 |
+| `latest` | 跳过快照，只读取任务启动后产生的变更。 |
+| `from-timestamp` | 跳过快照，读取 `source.scan.timestamp` 指定时间之后（不含该时间点）的变更，时间格式为
`yyyy-MM-dd HH:mm:ss`。 |
+
+默认以 `detail` 类型输出完整的行变更。也可以通过 `source.binlog.increment-type` 设置为
`min_delta`（最小变更集）或 `append_only`（仅追加事件）。
+
+使用时请注意：
+
+- 增量读取使用 Arrow Flight SQL，Connector 默认启用并自动获取端口。
+- 需要启用 Flink Checkpoint。
+- Doris Binlog 的保留时间应覆盖任务可能停止的最长时间。如果恢复所需的 Binlog 已过期，需要重新读取快照或指定新的起始时间。
+
+##### 将消费进度写入 Doris（可选）
+
+消费进度默认保存在 Flink Checkpoint 中。如果还需要在 Doris 中查询消费进度，可以创建以下 Offset 表：
+
+```sql
+CREATE DATABASE IF NOT EXISTS ops;
+
+CREATE TABLE ops.flink_source_offsets (
+ consumer_id VARCHAR(256) NOT NULL,
+ offset_timestamp DATETIME NOT NULL,
+ update_time DATETIMEV2(3) NOT NULL
+)
+UNIQUE KEY(consumer_id)
+DISTRIBUTED BY HASH(consumer_id) BUCKETS 1
+PROPERTIES (
+ "replication_num" = "1"
+);
+```
+
+```sql
+'jdbc-url' = 'jdbc:mysql://127.0.0.1:9030',
+'source.binlog.offset-table' = 'ops.flink_source_offsets',
+'source.binlog.consumer-id' = 'student-sync'
+```
+
+`source.binlog.consumer-id` 用于标识当前消费任务，同一任务重启时应保持不变。
+
#### DataStream API 读取
使用 DataStream API 读取数据时，需要在项目 POM 文件中提前引入依赖，参考 [安装方式](#安装方式) 章节。
@@ -793,6 +886,10 @@ Flink Doris Connector 集成了 [Flink
CDC](https://nightlies.apache.org/flink/f
| doris.request.retries | 3 | N | 向 Doris 发送请求的重试次数
|
| doris.request.connect.timeout | 30s | N | 向 Doris
发送请求的连接超时时间 |
| doris.request.read.timeout | 30s | N | 向 Doris
发送请求的读取超时时间 |
+| doris.enable.tls | FALSE | N | 是否为 Doris
HTTP、MySQL/JDBC、BE Thrift 和 Arrow Flight SQL 连接启用 TLS |
+| doris.tls.ca-certificate-path | -- | N | PEM CA
证书链的本地路径。为空时 Connector 不加载自定义 CA，使用对应客户端的默认信任库 |
+| doris.tls.skip-hostname-verification | FALSE | N | 是否在保留 CA
校验的同时跳过服务端 hostname 校验 |
+| doris.tls.excluded-protocols | -- | N | 启用 TLS
时仍使用明文的协议列表，逗号分隔。支持 `http`、`mysql`、`thrift` 和 `arrowflight` |
### Source 配置项
@@ -804,6 +901,12 @@ Flink Doris Connector 集成了 [Flink
CDC](https://nightlies.apache.org/flink/f
| doris.exec.mem.limit | 8192mb | N | 单个查询的内存限制。默认为
8GB，单位为字节
|
| source.use-flight-sql | TRUE | N | 是否使用 Arrow Flight
SQL 读取
|
| source.flight-sql-port | - | N | 使用 Arrow Flight SQL
读取时，FE 的 `arrow_flight_sql_port`
|
+| source.scan.mode | snapshot | N | Source 启动模式，支持
`snapshot`、`initial`、`latest` 和 `from-timestamp`
|
+| source.scan.timestamp | -- | N | `from-timestamp`
模式的开区间起始时间，格式为 `yyyy-MM-dd HH:mm:ss`
|
+| source.binlog.increment-type | detail | N | Binlog 变更类型，支持
`detail`、`min_delta` 和 `append_only`
|
+| source.binlog.poll-interval | 10s | N | 轮询新 Binlog
数据的时间间隔，最小值为 1 秒
|
+| source.binlog.offset-table | -- | N | 用于发布成功 Checkpoint
所覆盖 offset 的 Doris 表，格式为 `database.table`。需要同时配置 `source.binlog.consumer-id` 和
`jdbc-url` |
+| source.binlog.consumer-id | -- | N | 写入
`source.binlog.offset-table` 的稳定消费者标识
|
**DataStream 专有配置项**
@@ -816,10 +919,18 @@ Flink Doris Connector 集成了 [Flink
CDC](https://nightlies.apache.org/flink/f
| Key | Default Value | Required | Comment
[...]
| --------------------------- | ------------- | -------- |
-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
[...]
-| sink.label-prefix | -- | Y | Stream Load 导入使用的
label 前缀。2pc 场景下要求全局唯一，用来保证 Flink 的 EOS 语义。
[...]
+| sink.label-prefix | -- | Y | 导入使用的 label 前缀。2PC
场景下要求全局唯一，用来保证 Flink 的 EOS 语义。
[...]
| sink.properties.* | -- | N | Stream Load
的导入参数。例如：`'sink.properties.column_separator' = ', '`
定义列分隔符；`'sink.properties.escape_delimiters' = 'true'` 表示特殊字符作为分隔符，`\x01`
会被转换为二进制的 `0x01`；JSON 格式导入：`'sink.properties.format' =
'json'`、`'sink.properties.read_json_by_line' = 'true'`，详细参数参考 [Stream
Load](../../data-operate/import/import-way/stream-load-manual.md#导入配置参数)。Group
Commit 模式：`'sink.properties.group_commit' = 'sync_mode'` 设置 group commit
为同步模式。Flink Connec [...]
| sink.enable-delete | TRUE | N | 是否启用删除。此选项需要 Doris
表开启批量删除功能（Doris 0.15+ 版本默认开启），只支持 Unique 模型。
[...]
| sink.enable-2pc | TRUE | N | 是否开启两阶段提交（2pc），默认为
true，保证 Exactly-Once 语义。关于两阶段提交可参考 [Stream Load
2PC](../../data-operate/transaction.md#streamload-2pc)。
[...]
+| sink.write-mode | STREAM_LOAD | N | 写入模式，支持
`STREAM_LOAD`、`STREAM_LOAD_BATCH` 和 `TVF`
[...]
+| sink.s3.endpoint | -- | 仅 TVF 写入模式 | 兼容 S3 的对象存储
endpoint
[...]
+| sink.s3.region | -- | 仅 TVF 写入模式 | 对象存储 region
[...]
+| sink.s3.bucket | -- | 仅 TVF 写入模式 | 暂存文件使用的 bucket
[...]
+| sink.s3.prefix | -- | 仅 TVF 写入模式 | 暂存文件的对象 key
前缀，不能包含 glob 字符
[...]
+| sink.s3.access-key | -- | 仅 TVF 写入模式 | 对象存储 access key
[...]
+| sink.s3.secret-key | -- | 仅 TVF 写入模式 | 对象存储 secret key
[...]
+| sink.s3.path-style-access | FALSE | N | TVF 模式是否使用
path-style 对象存储访问方式
[...]
| sink.buffer-size | 1MB | N | 写数据缓存 buffer
大小，单位字节。不建议修改，默认配置即可
[...]
| sink.buffer-count | 3 | N | 写数据缓存 buffer
个数。不建议修改，默认配置即可
[...]
| sink.max-retries | 3 | N | Commit
失败后的最大重试次数，默认 3 次
[...]
@@ -1101,6 +1212,72 @@ from KAFKA_SOURCE;
在 Connector 提供的整库同步工具中，无需额外配置，会自动同步上游 DDL 并在 Doris 进行 Schema Change 操作。
+### Flink 访问启用 TLS 的 Doris 集群
+
+Connector 可以对 Doris HTTP 和 Stream Load、MySQL/JDBC、BE Thrift 以及 Arrow Flight
SQL 连接启用 TLS。TLS 默认关闭。Connector 会校验 Doris 服务端证书。
+
+在 Doris Source、Sink 或 Catalog 配置中增加以下选项：
+
+```sql
+'doris.enable.tls' = 'true',
+'doris.tls.ca-certificate-path' = '/etc/doris-tls/ca-chain.pem'
+```
+
+通过 `doris.tls.ca-certificate-path` 指定 CA 文件时，应使用 PEM 证书链，并确保所有需要连接 Doris 的
Flink 进程都能从本地文件系统读取该文件。未配置该路径时，Connector 不加载自定义 CA，使用对应客户端的默认信任库。生产环境应保持
hostname 校验开启。
+
+如果 Doris 的某个协议有意保留明文连接，只排除该协议即可。支持的值为 `http`、`mysql`、`thrift` 和 `arrowflight`：
+
+```sql
+'doris.tls.excluded-protocols' = 'arrowflight'
+```
+
+Connector 不会探测协议，也不会在 TLS 失败后回退到明文连接。
+
+启用 `doris.enable.tls` 后，Connector 会通过 JDBC 连接属性传入 TLS 配置，不会修改
`jdbc-url`。因此，`jdbc-url` 只需填写连接地址，无需添加 `sslMode`、`useSSL` 或 Trust Store 等 TLS
参数。
+
+Arrow Flight SQL 支持 TLS，但不支持仅跳过 hostname 校验；如果将
`doris.tls.skip-hostname-verification` 设置为 `true`，需要通过
`doris.tls.excluded-protocols` 排除 `arrowflight`。
+
+根据 Flink 部署模式分发 CA 文件：
+
+- **Standalone**：将文件放在所有会连接 Doris 的 JobManager、TaskManager 和 SQL Gateway
主机上的相同路径。
+- **YARN**：通过 `yarn.ship-files: /local/path/ca.pem` 分发文件，并将
`doris.tls.ca-certificate-path` 设置为容器内文件名，例如 `ca.pem`。
+- **Kubernetes**：通过 ConfigMap 或 Secret 将 CA 挂载到相关 JobManager 和 TaskManager Pod
的相同路径。
+
+### 使用 S3 TVF 写入
+
+TVF 写入模式先将数据以 JSON 格式暂存至 S3 对象存储，再通过 S3 表值函数导入 Doris。适合优先使用对象存储作为数据传输通道，或无法使用
Stream Load 网络链路的场景。
+
+使用前，确保 Flink 和 Doris 均可访问 S3，且 Doris 中已创建目标表。
+
+```sql
+SET 'execution.checkpointing.interval' = '30s';
+
+CREATE TABLE student_tvf_sink (
+ id INT,
+ name STRING,
+ age INT
+) WITH (
+ 'connector' = 'doris',
+ 'fenodes' = '127.0.0.1:8030',
+ 'jdbc-url' = 'jdbc:mysql://127.0.0.1:9030',
+ 'table.identifier' = 'test.student_tvf',
+ 'username' = 'root',
+ 'password' = '',
+ 'sink.write-mode' = 'TVF',
+ 'sink.label-prefix' = 'student_tvf',
+ 'sink.s3.endpoint' = 'https://s3.example.com',
+ 'sink.s3.region' = 'us-east-1',
+ 'sink.s3.bucket' = 'staging-bucket',
+ 'sink.s3.prefix' = 'doris/student',
+ 'sink.s3.access-key' = 'access-key',
+ 'sink.s3.secret-key' = 'secret-key'
+);
+
+INSERT INTO student_tvf_sink VALUES (1, 'Alice', 18);
+```
+
+Connector 不会自动删除 S3 中的暂存对象，请按需配置对象存储生命周期策略。
+
## 常见问题
**1. errCode = 2, detailMessage = Label [label_0_1] has already been used,
relate to txn [19650]**
diff --git
a/i18n/zh-CN/docusaurus-plugin-content-docs/current/connection-integration/data-integration/spark-doris-connector.md
b/i18n/zh-CN/docusaurus-plugin-content-docs/current/connection-integration/data-integration/spark-doris-connector.md
index 9416ef676d0..bfd9f4ed2a3 100644
---
a/i18n/zh-CN/docusaurus-plugin-content-docs/current/connection-integration/data-integration/spark-doris-connector.md
+++
b/i18n/zh-CN/docusaurus-plugin-content-docs/current/connection-integration/data-integration/spark-doris-connector.md
@@ -42,6 +42,7 @@ Spark Doris Connector 是 Apache Doris 与 Apache Spark 的连接器，支持通
| Connector | Spark | Doris | Java | Scala |
| --- | --- | --- | --- | --- |
+| 26.1.0 | 4.1, 3.5 - 3.1, 2.4 | 1.0 + | 17 (4.1), 8 (3.x, 2.x) | 2.13 (4.1),
2.12 (3.x), 2.11 (2.x) |
| 26.0.0 | 3.5 - 3.1, 2.4 | 1.0 + | 8 | 2.12, 2.11 |
| 25.2.0 | 3.5 - 3.1, 2.4 | 1.0 + | 8 | 2.12, 2.11 |
| 25.1.0 | 3.5 - 3.1, 2.4 | 1.0 + | 8 | 2.12, 2.11 |
@@ -62,7 +63,7 @@ Spark Doris Connector 是 Apache Doris 与 Apache Spark 的连接器，支持通
<dependency>
<groupId>org.apache.doris</groupId>
<artifactId>spark-doris-connector-spark-3.5</artifactId>
- <version>25.2.0</version>
+ <version>26.1.0</version>
</dependency>
```
@@ -82,22 +83,22 @@ Spark Doris Connector 是 Apache Doris 与 Apache Spark 的连接器，支持通
如需自行编译，在源码目录下执行 `sh build.sh`，并根据提示输入需要的 Scala 与 Spark 版本。
-编译成功后，目标 Jar 包会生成在 `dist` 目录下，例如
`spark-doris-connector-spark-3.5-25.2.0.jar`。将该文件复制到 Spark 的 `classpath` 中即可使用
Spark Doris Connector：
+编译成功后，目标 Jar 包会生成在 `dist` 目录下，例如
`spark-doris-connector-spark-3.5-26.1.0.jar`。将该文件复制到 Spark 的 `classpath` 中即可使用
Spark Doris Connector：
| Spark 运行模式 | Jar 包放置方式 |
| --- | --- |
| Local 模式 | 将 Jar 包放入 `jars/` 目录。 |
| Yarn 集群模式 | 将 Jar 包放入预部署包中。 |
-例如，将 `spark-doris-connector-spark-3.5-25.2.0.jar` 上传到 HDFS，并通过
`spark.yarn.jars` 添加依赖：
+例如，将 `spark-doris-connector-spark-3.5-26.1.0.jar` 上传到 HDFS，并通过
`spark.yarn.jars` 添加依赖：
```shell
-# 1. 上传 spark-doris-connector-spark-3.5-25.2.0.jar 到 HDFS
+# 1. 上传 spark-doris-connector-spark-3.5-26.1.0.jar 到 HDFS
hdfs dfs -mkdir /spark-jars/
-hdfs dfs -put /your_local_path/spark-doris-connector-spark-3.5-25.2.0.jar
/spark-jars/
+hdfs dfs -put /your_local_path/spark-doris-connector-spark-3.5-26.1.0.jar
/spark-jars/
-# 2. 在集群中添加 spark-doris-connector-spark-3.5-25.2.0.jar 依赖
-spark.yarn.jars=hdfs:///spark-jars/spark-doris-connector-spark-3.5-25.2.0.jar
+# 2. 在集群中添加 spark-doris-connector-spark-3.5-26.1.0.jar 依赖
+spark.yarn.jars=hdfs:///spark-jars/spark-doris-connector-spark-3.5-26.1.0.jar
```
## 场景一：批量读取 Doris 数据
@@ -458,13 +459,21 @@ insert into
your_catalog_name.your_doris_db.your_doris_table select * from your_
| `doris.sink.batch.interval.ms` | 0 | 每个批次 Sink 的间隔时间，单位为 ms。 |
| `doris.sink.enable-2pc` | false |
是否开启两阶段提交。开启后将在作业结束时提交事务，而部分任务失败时会将所有预提交状态的事务回滚。 |
| `doris.sink.auto-redirect` | true | 是否重定向 Stream Load 请求。开启后 Stream Load 将通过
FE 写入，不再显式获取 BE 信息。 |
-| `doris.enable.https` | false | 是否开启 FE HTTPS 请求。 |
-| `doris.https.key-store-path` | - | HTTPS key store 路径。 |
-| `doris.https.key-store-type` | JKS | HTTPS key store 类型。 |
-| `doris.https.key-store-password` | - | HTTPS key store 密码。 |
+| `doris.enable.tls` | false | 是否为 Doris 连接启用 TLS。 |
+| `doris.tls.ca-certificate-path` | 空字符串 | PEM 格式的 CA 证书链路径。未配置时 Connector
不加载自定义 CA，使用对应客户端的默认信任库。 |
+| `doris.tls.skip-hostname-verification` | false | 是否在保留 CA 验证的同时跳过主机名验证。Arrow
Flight SQL 不支持该配置。 |
+| `doris.tls.excluded-protocols` | 空字符串 | 不使用 TLS 的协议，多个协议使用逗号分隔。可选值为
`http`、`mysql`、`thrift` 和 `arrowflight`。 |
| `doris.read.mode` | thrift | Doris 读取模式，可选项为 `thrift` 和 `arrow`。 |
| `doris.read.arrow-flight-sql.port` | - | Doris FE 的 Arrow Flight SQL 端口。当
`doris.read.mode` 为 `arrow` 时，用于通过 Arrow Flight SQL 方式读取数据。服务端配置方式请参考 [基于 Arrow
Flight SQL 的高速数据传输链路](../arrow-flight-sql.md)。 |
-| `doris.sink.label.prefix` | spark-doris | Stream Load 方式写入时的导入标签前缀。 |
+| `doris.sink.mode` | stream_load | Doris 写入模式，支持 `stream_load` 和 `tvf`。 |
+| `doris.sink.label.prefix` | spark-doris | Doris 写入的 Label 前缀。 |
+| `doris.sink.s3.endpoint` | - | 兼容 S3 的对象存储 Endpoint，仅 TVF 写入模式下必填。 |
+| `doris.sink.s3.region` | - | 对象存储 Region，仅 TVF 写入模式下必填。 |
+| `doris.sink.s3.bucket` | - | 暂存数据的 Bucket，仅 TVF 写入模式下必填。 |
+| `doris.sink.s3.prefix` | - | 暂存数据的对象路径前缀，仅 TVF 写入模式下必填。 |
+| `doris.sink.s3.access-key` | - | 对象存储 Access Key，仅 TVF 写入模式下必填。 |
+| `doris.sink.s3.secret-key` | - | 对象存储 Secret Key，仅 TVF 写入模式下必填。 |
+| `doris.sink.s3.path-style-access` | false | TVF 写入模式下是否使用路径风格访问对象存储。 |
| `doris.thrift.max.message.size` | 2147483647 | 通过 Thrift 方式读取数据时，消息的最大尺寸。 |
| `doris.fe.auto.fetch` | false | 是否自动获取 FE 信息。当设置为 `true` 时，会根据
`doris.fenodes` 配置的节点请求所有 FE 节点信息，无需额外配置多个节点以及单独配置
`doris.read.arrow-flight-sql.port` 和 `doris.query.port`。 |
| `doris.read.bitmap-to-string` | false | 是否将 Bitmap
类型转换为数组索引组成的字符串读取。具体结果形式参考函数定义
[BITMAP_TO_STRING](../../sql-manual/sql-functions/scalar-functions/bitmap-functions/bitmap-to-string.md)。
|
@@ -545,6 +554,54 @@ insert into
your_catalog_name.your_doris_db.your_doris_table select * from your_
:::
+## 最佳实践
+
+### Spark 访问启用 TLS 的 Doris 环境
+
+Spark 读写启用 TLS 的 Doris 集群时，将 `doris.enable.tls` 设置为 `true`：
+
+```scala
+mockDataDF.write.format("doris")
+ .option("doris.table.identifier",
"$YOUR_DORIS_DATABASE_NAME.$YOUR_DORIS_TABLE_NAME")
+ .option("doris.fenodes",
"$YOUR_DORIS_FE_HOSTNAME:$YOUR_DORIS_FE_HTTPS_PORT")
+ .option("user", "$YOUR_DORIS_USERNAME")
+ .option("password", "$YOUR_DORIS_PASSWORD")
+ .option("doris.enable.tls", "true")
+ .option("doris.tls.ca-certificate-path", "/etc/doris-tls/ca.pem")
+ .save()
+```
+
+CA 文件需要在 Spark Driver 和所有 Executor 上可访问。未配置 `doris.tls.ca-certificate-path`
时，Connector 不加载自定义 CA，使用对应客户端的默认信任库。
+
+根据 Spark 部署模式分发 CA 文件：
+
+- **YARN**：在 `spark-submit` 中增加 `--files /local/path/ca.pem`，并将
`doris.tls.ca-certificate-path` 设置为 `ca.pem`。
+- **Kubernetes**：通过 Secret 或 Volume 将 CA 挂载到 Driver 和 Executor Pod 的相同路径。例如，在
`spark-submit` 中增加 `--conf
spark.kubernetes.driver.secrets.doris-tls=/etc/doris-tls` 和 `--conf
spark.kubernetes.executor.secrets.doris-tls=/etc/doris-tls`，并将 CA 路径设置为
`/etc/doris-tls/ca.pem`。
+
+### 使用 S3 TVF 写入
+
+TVF 写入模式先将数据以 JSON 形式存到兼容 S3 的对象存储，再通过 S3 TVF 写入 Doris，目前仅支持批量写入。请确保 Spark 和
Doris 均可访问对象存储，并已创建 Doris 目标表。
+
+```scala
+mockDataDF.write.format("doris")
+ .option("doris.table.identifier",
"$YOUR_DORIS_DATABASE_NAME.$YOUR_DORIS_TABLE_NAME")
+ .option("doris.fenodes",
"$YOUR_DORIS_FE_HOSTNAME:$YOUR_DORIS_FE_HTTP_PORT")
+ .option("doris.query.port", "$YOUR_DORIS_FE_QUERY_PORT")
+ .option("user", "$YOUR_DORIS_USERNAME")
+ .option("password", "$YOUR_DORIS_PASSWORD")
+ .option("doris.sink.mode", "tvf")
+ .option("doris.sink.s3.endpoint", "$YOUR_S3_ENDPOINT")
+ .option("doris.sink.s3.region", "$YOUR_S3_REGION")
+ .option("doris.sink.s3.bucket", "$YOUR_S3_BUCKET")
+ .option("doris.sink.s3.prefix", "$YOUR_S3_PREFIX")
+ .option("doris.sink.s3.access-key", "$YOUR_S3_ACCESS_KEY")
+ .option("doris.sink.s3.secret-key", "$YOUR_S3_SECRET_KEY")
+ .mode(SaveMode.Append)
+ .save()
+```
+
+Connector 不会自动删除暂存对象，请按需配置对象存储生命周期策略。
+
## 常见问题与故障处理
<!-- 知识类型: FAQ + 故障排查 -->
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
