使用SeaTunnel进行一键数据同步

一.Apache SeaTunnel

      Apache SeaTunnel是一款数据集成工具，支持批处理和流处理。它允许用户通过简单的配置实现数据的抽取，转换和加载（ETL过程）。SaeTunnel提供了多种连接器，能够轻松集成不同的数据源项目和目标，包括关系型数据库，NoSQL数据库，文件系统等。其灵活性和扩展性使其成为企业数据集成的重要选择。

二.使用方式

2.1.使用SeaTunnel Zeta 引擎

• 本人此次同步涉及到Oracle，PostgresSql, Doris, Mysql

• 所需jar包有：ojdbc8.jar postgresql-42.3.3.jar mysql-connector-java-8.0.18.jar connector-doris-2.3.8.jar connector-jdbc-2.3.8.jar

三. 数据同步

DorisSink参数详解：

schema_save_mode[Enum]

在开启同步任务之前，针对现有的表结构选择不同的处理方案。 选项介绍：
RECREATE_SCHEMA ：表不存在时创建，表保存时删除并重建。

CREATE_SCHEMA_WHEN_NOT_EXIST ：表不存在时会创建，表存在时跳过。
ERROR_WHEN_SCHEMA_NOT_EXIST ：表不存在时会报错。
IGNORE ：忽略对表的处理。

data_save_mode[Enum]

在开启同步任务之前，针对目标端已有的数据选择不同的处理方案。 选项介绍：
DROP_DATA： 保留数据库结构并删除数据。
APPEND_DATA：保留数据库结构，保留数据。
CUSTOM_PROCESSING：用户自定义处理。
ERROR_WHEN_DATA_EXISTS：有数据时报错。

PS：这个参数在本人亲测时DROP_DATA 这个选项并不会生效，原因是在Sink端无权限清空表中数据，感兴趣的可以做一下测试

save_mode_create_template

使用模板自动创建Doris表， 会根据上游数据类型和schema类型创建相应的建表语句， 默认模板可以根据情况进行修改。

默认模板：

CREATE TABLE IF NOT EXISTS `${database}`.`${table_name}` (
${rowtype_primary_key},
${rowtype_fields}
) ENGINE=OLAP
 UNIQUE KEY (${rowtype_primary_key})
DISTRIBUTED BY HASH (${rowtype_primary_key})
 PROPERTIES (
"replication_allocation" = "tag.location.default: 1",
"in_memory" = "false",
"storage_format" = "V2",
"disable_auto_compaction" = "false"
)

如果模板中填写了自定义字段，例如添加 id 字段

CREATE TABLE IF NOT EXISTS `${database}`.`${table_name}`
(   
    id,
    ${rowtype_fields}
) ENGINE = OLAP UNIQUE KEY (${rowtype_primary_key})
    DISTRIBUTED BY HASH (${rowtype_primary_key})
    PROPERTIES
(
    "replication_num" = "1"
);

连接器会自动从上游获取对应类型完成填充， 并从“rowtype_fields”中删除 id 字段。 该方法可用于自定义字段类型和属性的修改。

可以使用以下占位符：

• database：用于获取上游schema中的数据库。
• table_name：用于获取上游schema中的表名。
• rowtype_fields：用于获取上游schema中的所有字段，自动映射到Doris的字段描述。
• rowtype_primary_key：用于获取上游模式中的主键（可能是列表）
• rowtype_unique_key：用于获取上游模式中的唯一键（可能是列表）

数据类型映射

支持的导入数据格式

支持的格式包括 CSV 和 JSON。

调优指南

适当增加sink.buffer-size和doris.batch.size的值可以提高写性能。

在流模式下，如果doris.batch.size和checkpoint.interval都配置为较大的值，最后到达的数据可能会有较大的延迟(延迟的时间就是检查点间隔的时间)。

这是因为最后到达的数据总量可能不会超过doris.batch.size指定的阈值。因此，在接收到数据的数据量没有超过该阈值之前只有检查点才会触发提交操作。因此，需要选择一个合适的检查点间隔。

此外，如果你通过sink.enable-2pc=true属性启用2pc。sink.buffer-size将会失去作用，只有检查点才能触发提交。

OracleSource参数详解：

上述参数与数据库版本相关，使用时请严格对应版本，避免因jar包与数据库版本冲突产生相关问题。

Source Options

 Task Example

# Defining the runtime environment
env {
  parallelism = 4
  job.mode = "BATCH"
}
source{
    Jdbc {
        url = "jdbc:oracle:thin:@datasource01:1523:xe"
        driver = "oracle.jdbc.OracleDriver"
        user = "root"
        password = "123456"
        query = "SELECT * FROM TEST_TABLE"
    }
}

transform {
    # If you would like to get more information about how to configure seatunnel and see full list of transform plugins,
    # please go to https://seatunnel.apache.org/docs/transform-v2/sql
}

sink {
    Console {}
}