一次 SeaTunnel 任务提交，看起来只是一次 submitJob 请求；但在 Server 内部，它会经过 Master 判断、任务协调、JobMaster 初始化、物理执行计划构建、Pipeline 资源申请、TaskGroup 部署等多个阶段。

这篇文章基于我整理的 submitJob 时序，先聚焦一条主线：任务从提交请求进入 SeaTunnel Server，到最终调用 TaskExecutionService.deployTask() 部署 TaskGroup，中间都经过了什么。

本文暂不展开 TaskExecutionService 内部的线程模型、Task 运行细节和数据流转，重点放在任务提交与调度部署链路上。

核心角色​

在进入流程之前，先看 submitJob 主链路里几个关键对象分别负责什么。

总体流程​

先用一张简化流程图看全局链路。

这条链路可以压缩成一句话：

SubmitJobServlet
  -> JobInfoService
  -> MasterNode / CoordinatorService
  -> JobMaster
  -> PhysicalPlan
  -> SubPlan
  -> PhysicalVertex
  -> TaskExecutionService

下面按阶段拆开看。

第一阶段：请求进入 JobInfoService​

任务提交入口首先进入 SubmitJobServlet，随后交给 JobInfoService 处理。

这里的关键点不是马上启动任务，而是先判断：当前接收请求的节点是不是 Master。

如果当前节点就是 Master，JobInfoService 可以继续本地提交；如果当前节点是 Worker，则需要通过 MasterNode.submitJob() 转发给 Master。

这个设计保证了任务提交最终由 Master 统一协调，避免多个节点各自独立创建 Job 调度上下文。

第二阶段：CoordinatorService 接管任务​

请求进入 Master 后，会继续调用 CoordinatorService.submitJob()。

CoordinatorService 在这里主要做两件事：

1. 判断这个 Job 是否已经存在或正在运行。
2. 如果是新任务，则创建并初始化对应的 JobMaster。

如果任务已经运行，SeaTunnel 不需要重复创建调度上下文，可以直接返回提交成功。如果是一个新任务，就会进入 JobMaster 初始化流程。

也就是说，submitJob 到这里已经从“接口请求处理”进入了“调度系统处理”。

第三阶段：JobMaster 初始化​

JobMaster 可以理解为一个 Job 的运行控制中心。

创建 JobMaster 后，会完成一些运行前准备工作，例如：

• 构建任务运行所需的 classloader。
• 初始化 checkpoint 相关配置。
• 准备从逻辑 DAG 构建物理执行计划所需的上下文。

这一阶段还没有真正部署 Task，它更像是在为后续调度准备运行环境。

第四阶段：从逻辑 DAG 到 PhysicalPlan​

JobMaster 初始化后，会基于逻辑 DAG 构建 PhysicalPlan。

这里有一个重要概念：SeaTunnel 不是一次性把任务全部启动，而是通过状态机逐步推进。

在 Job 级别，核心状态推进可以简化理解为：

CREATED -> SCHEDULED -> startSubPlanStateProcess

PhysicalPlan 负责 Job 级别状态流转，而真正的 Pipeline 调度会继续下沉到 SubPlan。

第五阶段：SubPlan 申请资源并进入部署​

到了 SubPlan 这一层，SeaTunnel 关注的粒度已经从整个 Job 下沉到 Pipeline。

SubPlan.stateProcess() 会根据当前 Pipeline 状态执行不同逻辑：

这一层的重点是：

• CREATED 状态下，Pipeline 会先推进到 SCHEDULED。
• SCHEDULED 状态下，开始通过 ResourceUtils.applyResourceForPipeline() 申请资源。
• 资源申请成功后，Pipeline 进入 DEPLOYING。
• 如果资源申请失败，则进入 makePipelineFailing(e)。

因此，Pipeline 不是被立即部署的；它必须先具备运行资源。

第六阶段：PhysicalVertex 部署 TaskGroup​

当 Pipeline 进入 DEPLOYING 后，SubPlan 会开始启动内部的 PhysicalVertex。

PhysicalVertex 会先把 Task 状态更新为 DEPLOYING，然后根据分配到的 slotProfile 执行部署。

部署时有一个关键分支：目标 Worker 是本地还是远端。

如果目标 Worker 就是当前节点，可以直接调用本地 TaskExecutionService.deployTask(taskGroupInfo)。

如果目标 Worker 是远端节点，则需要通过 DeployTaskOperation 把部署请求发送过去，最终仍然会在目标 Worker 上进入 TaskExecutionService.deployTask(taskGroupInfo)。

部署成功后，PhysicalVertex 会把 Task 状态更新为 RUNNING；如果失败，则进入 makeTaskGroupFailing()。

当 Pipeline 内部的 TaskGroup 部署完成并进入运行态后，SubPlan 也会推进到 RUNNING。

失败、取消与恢复分支​

除了正常提交和部署，SubPlan 状态机里还需要处理失败、取消和恢复。

可以简化成下面这张图：

这也是为什么前面的状态机设计很重要：

• 正常路径可以推进部署和运行。
• 失败路径可以进入 failing / failed。
• 取消路径可以进入 canceling / canceled。
• 如果满足恢复条件，还可以释放资源后重新申请并恢复 Pipeline。

换句话说，状态机不是为了让流程复杂，而是为了让任务生命周期可控。

完整时序图​

最后，把主流程用时序图串起来，方便对照整体调用顺序。

总结​

SeaTunnel 提交任务后的核心逻辑，并不是“收到请求后直接启动任务”。

它大致会经过下面这条主链路：

SubmitJobServlet
  -> JobInfoService
  -> MasterNode / CoordinatorService
  -> JobMaster
  -> PhysicalPlan
  -> SubPlan
  -> PhysicalVertex
  -> TaskExecutionService

其中：

• JobInfoService 负责处理提交入口，并判断是否需要转发到 Master。
• CoordinatorService 负责接管任务协调，避免重复提交，并创建 JobMaster。
• JobMaster 负责初始化 Job 运行上下文。
• PhysicalPlan 负责 Job 级别状态推进。
• SubPlan 负责 Pipeline 级别的资源申请和调度。
• PhysicalVertex 负责 TaskGroup 部署。
• TaskExecutionService 是最终执行 TaskGroup 部署的入口。

理解这条链路之后，再去看 SeaTunnel 的 Task 执行线程模型、数据流转和 checkpoint 机制，就会更容易把各个模块放到正确的位置上。

在企业级数据集成中，数据一致性 是技术决策者最关心的核心问题之一。但在这个看似简单的诉求背后，实际隐藏着复杂的技术挑战和架构设计。

当企业用户使用 SeaTunnel 进行 批流数据同步 时，通常会关注这些问题：

🔍 “如何确保源库和目标库之间的数据完整性？”
🔄 “任务中断或恢复后，能否避免数据重复或丢失？”
⚙️ “全量同步与增量同步过程中，如何保证一致性？”

本文以 Apache SeaTunnel 2.3.13 为配置基线，说明 读取一致性、写入一致性和状态一致性 如何协同工作。“零丢失”和“零重复”并非默认能力，而是有前提的结果：Source 和 Sink 语义必须兼容，Checkpoint 必须成功完成，还需要正确的主键或唯一键以及文档要求的 exactly-once 配置。

下文示例和术语以 2.3.13 版本的 MySQL-CDC Source、JDBC Source、JDBC Sink 和 Job Env 配置 文档为准。

一、理解数据一致性的三个维度​

在数据集成领域，“一致性”并不是一个单一概念，而是一组覆盖多个维度的保障。为了便于分析，本文将 SeaTunnel 的相关机制归纳为三个维度：

读取一致性​

读取一致性 确保从源端系统获取的数据，在某个时间点或事件序列上保持逻辑完整性。这个维度解决的是“应该采集哪些数据”的问题：

• 全量读取：在特定时间点获取完整的数据快照
• 增量采集：准确记录所有数据变更事件（CDC 模式）
• 无锁快照一致性：设置 exactly_once = true 时，通过 low watermark 和 high watermark 补齐快照期间发生的变更

写入一致性​

写入一致性 确保数据能够可靠、正确地写入目标端系统，解决的是“如何安全写入”的问题：

• 幂等写入：存在稳定主键/唯一键且启用 Upsert 语义时，同一个键被重放只会更新一条目标记录
• 事务完整性：当 Sink 支持时，将单个 Sink Writer 处理的数据放入与 Checkpoint 对齐的事务中提交
• 错误处理：从已完成的 Checkpoint 恢复，重放行为由 Source 和 Sink 语义共同决定

状态一致性​

状态一致性 是连接读取端和写入端的桥梁，确保整个数据同步过程中的状态跟踪与恢复：

• 位点管理：记录读取进度，用于精确增量同步
• Checkpoint 机制：周期性保存任务状态
• Checkpoint 恢复：从已完成的 Checkpoint 恢复；该 Checkpoint 之后的数据可能被重放，除非 Sink 具备幂等或事务型 exactly-once 能力

二、MySQL 同步架构：CDC 与 JDBC 模式对比​

SeaTunnel 提供两种主流的 MySQL 数据同步模式：JDBC 批模式 和 CDC 实时采集模式。这两种模式适用于不同业务场景，并且在一致性保障上各有特点。

CDC 模式：基于 Binlog 的低延迟变更采集​

MySQL-CDC Connector 基于嵌入式 Debezium 框架实现，直接读取并解析 MySQL 的 binlog 变更流：

核心优势：

• 低延迟：持续读取 binlog 变更，实际延迟取决于源端负载、网络和任务资源
• 减少轮询：避免反复轮询业务表，但初始化快照仍会消耗源端资源
• 完整性：完整捕获 INSERT/UPDATE/DELETE 事件
• 变更元数据：输出携带 binlog 位点元数据的行级变更事件

恢复与顺序特征：

• 在 Checkpoint 中保存 binlog 文件名和位点，用于失败恢复
• 支持多种启动模式（初始化快照 + 增量 / 仅增量）
• 保持单个 Source Reader 观察到的事件顺序；端到端顺序仍受表路由、并行度和下游处理影响

MySQL-CDC 不会把一个源端事务转换成一个下游原子事务。Connector 输出的是独立的行级变更事件，最终交付语义由 Checkpoint 和 Sink 能力决定。

JDBC 模式：基于 SQL 的批量同步方案​

JDBC Connector 通过 SQL 查询从 MySQL 读取数据，适用于周期性全量同步或低频变更场景：

核心优势：

• 开发简单：基于标准 SQL，配置灵活
• 全量同步：适合大规模数据初始化
• 过滤能力：支持复杂 WHERE 条件过滤
• 并行加载：可基于主键或范围进行多分片并行读取

恢复特征：

• 跟踪 JDBC Split，而不是 Split 内部的行级 offset
• 失败后重新分配或重放未完成的 Split
• 支持表级并行处理

因此，JDBC Source 是 Split 级恢复。失败时，正在处理的 Split 可能从其边界重新读取，重复数据必须由幂等 Sink 或事务型 exactly-once Sink 处理。

三、读取一致性：如何确保源端数据完整采集​

CDC 模式：基于 Binlog 的精确增量读取​

MySQL-CDC Connector 的读取一致性基于两个核心机制：初始化快照 和 Binlog 位点跟踪。

启动模式与一致性保障：

SeaTunnel 的 MySQL-CDC 提供多种启动模式，用于满足不同场景下的一致性要求：

1. Initial Mode：先创建全量快照，再继续读取增量 binlog。如果快照阶段必须补齐 low watermark 与 high watermark 之间的变更，需要设置 exactly_once = true。

   MySQL-CDC {
     startup.mode = "initial"
     exactly_once = true
   }
   

2. Latest Mode：只采集 Connector 启动之后的最新变更

   MySQL-CDC {
     startup.mode = "latest"
   }
   

3. Specific Mode：从指定 binlog 位点开始同步

   MySQL-CDC {
     startup.mode = "specific"
     startup.specific-offset.file = "mysql-bin.000003"
     startup.specific-offset.pos = 4571
   }
   

此外还有 earliest 启动模式：从能够找到的最早 offset 开始；以及 timestamp 启动模式（startup.timestamp）：从用户指定的毫秒级时间戳开始。

JDBC 模式：基于分片的高效批量读取​

JDBC Connector 通过智能分片策略实现高效并行读取：

分片策略与一致性：

• 主键/唯一键分片：存在受支持的键时，按该键拆分表
• 指定分片列：自动发现的键不合适时，通过 partition_column 指定分片列
• 均匀或采样分片：根据键分布和配置的阈值选择分片策略

SeaTunnel JDBC 读取分片示例配置：

Jdbc {
  url = "jdbc:mysql://source_mysql:3306/test"
  driver = "com.mysql.cj.jdbc.Driver"
  user = "root"
  password = "password"
  table_path = "test.users"
  split.size = 10000
  split.even-distribution.factor.upper-bound = 100
  split.even-distribution.factor.lower-bound = 0.05
  split.sample-sharding.threshold = 1000
}

通过这种方式，SeaTunnel 可以实现：

• 并行处理相互独立的 Split
• 通过 Checkpoint 跟踪待处理的 Split 状态
• 恢复时从 Split 边界重放未完成的 Split

这不是行级断点续传。如果重放可能让同一行多次到达目标端，应使用具备稳定主键/唯一键的幂等 Upsert，或启用受支持的 exactly-once Sink。

四、写入一致性：如何确保目标端数据准确​

在数据写入阶段，SeaTunnel 提供可配置机制，用于控制目标端 MySQL 的重放与事务行为。

幂等写入：确保数据不重复​

SeaTunnel 的 JDBC Sink Connector 通过多种策略实现幂等写入：

Upsert 模式：

幂等写入示例配置：

Jdbc {
  url = "jdbc:mysql://target_mysql:3306/test"
  driver = "com.mysql.cj.jdbc.Driver"
  user = "root"
  password = "password"
  generate_sink_sql = true
  database = "test"
  table = "users"
  primary_keys = ["id"]
  enable_upsert = true
}

批量提交与优化：

JDBC Sink 的批处理和重试行为由固定配置明确控制：

• 固定批大小：batch_size 控制缓冲记录达到多少条时触发 Flush
• Checkpoint 对齐 Flush：Checkpoint 处理过程中也会刷新缓冲数据
• 显式重试次数：max_retries 控制批执行重试，默认值为 0；启用 XA exactly-once 时必须保持为 0

分布式事务：XA 保障与两阶段提交​

对于受支持的 Connector 路径，JDBC Sink 会将每个 Writer 的 XA 事务与 SeaTunnel Checkpoint 协调起来：

启用 XA 分布式事务的示例配置：

Jdbc {
  url = "jdbc:mysql://target_mysql:3306/test"
  driver = "com.mysql.cj.jdbc.Driver"
  user = "root"
  password = "password"
  generate_sink_sql = true
  database = "test"
  table = "users"
  primary_keys = ["id"]
  enable_upsert = true
  max_retries = 0
  is_exactly_once = true
  xa_data_source_class_name = "com.mysql.cj.jdbc.MysqlXADataSource"
  max_commit_attempts = 3
}

XA 事务作用范围：

• 每个 Sink Writer 为一个 Checkpoint 准备自己的 XA 事务
• Checkpoint 完成后提交已 Prepare 的事务
• 恢复时，Connector 按协议处理该 Writer 未完成或已 Prepare 的事务

这为每个受支持的 JDBC Sink Writer 提供与 Checkpoint 对齐的 exactly-once 交付，但它不会将源端事务原样保留为一个下游事务，也不是跨多个表、Writer 或数据库的单一全局原子事务。跨系统业务原子性需要单独的事务设计。

五、状态一致性：断点续传与失败恢复​

基于 Checkpoint 的状态管理为受支持的 Source 和 Sink Connector 提供可恢复的边界。

分布式 Checkpoint 机制​

SeaTunnel 在分布式环境中实现状态一致性 checkpoint：

核心实现原则：

1. 位点记录：CDC Source 记录 Split offset；JDBC Source 记录 Split 状态，但不记录正在处理的 Split 内部行级 offset
2. Checkpoint 触发：按照 checkpoint.interval 周期性调度 Checkpoint
3. 状态持久化：将状态信息持久化到存储系统
4. 失败恢复：恢复最近一次已完成的 Checkpoint；该 Checkpoint 之后的工作可能被重放

有前提的端到端交付语义​

SeaTunnel 通过 Checkpoint 协调 Source 和 Sink 状态，最终交付语义取决于两端 Connector 及其配置：

使用 at-least-once Sink 时，重放可能产生重复写入。存在稳定主键/唯一键时，幂等 Upsert 可以吸收重复。JDBC XA exactly-once 还要求 is_exactly_once = true、兼容的 XA DataSource、max_retries = 0、已启用 Checkpoint 以及数据库端支持。

Checkpoint 配置示例：

env {
  checkpoint.interval = 5000
  checkpoint.timeout = 60000
}

六、实战配置：MySQL CDC 到 MySQL 全量 + 增量同步​

下面通过一个实战示例，展示如何配置 SeaTunnel 来实现可靠的 MySQL 到 MySQL 数据同步。

经典 CDC 模式配置​

下面的 SeaTunnel 2.3.13 示例启用了 MySQL-CDC 快照一致性和与 Checkpoint 对齐的 JDBC XA 交付。其保障成立的前提包括：源端和目标端具有稳定主键、MySQL 驱动及服务端支持 XA、Checkpoint 存储可靠且 Checkpoint 成功完成。它不是跨两个目标表的全局事务。

env {
  job.mode = "STREAMING"
  parallelism = 3
  checkpoint.interval = 60000
  checkpoint.timeout = 120000
}

source {
  MySQL-CDC {
    url = "jdbc:mysql://source_mysql:3306/test_db"
    username = "root"
    password = "password"
    database-names = [
      "test_db"
    ]
    table-names = [
      "test_db.mysqlcdc_to_mysql_table1",
      "test_db.mysqlcdc_to_mysql_table2"
    ]
    server-id = "5400-5408"

    # Initialization mode (full + incremental)
    startup.mode = "initial"
    exactly_once = true

    # Enable DDL changes
    schema-changes.enabled = true

    # Parallel read configuration
    snapshot.split.size = 8096
    snapshot.fetch.size = 1024
  }
}

transform {
  # Optional data transformation processing
}

sink {
  Jdbc {
    url = "jdbc:mysql://mysql_target:3306/test_db?useUnicode=true&characterEncoding=UTF-8&rewriteBatchedStatements=true"
    driver = "com.mysql.cj.jdbc.Driver"
    user = "root"
    password = "password"
    generate_sink_sql = true
    database = "${database_name}"
    table = "${table_name}"
    primary_keys = ["${primary_key}"]
    schema_save_mode = "CREATE_SCHEMA_WHEN_NOT_EXIST"
    data_save_mode = "APPEND_DATA"
    enable_upsert = true
    max_retries = 0
    is_exactly_once = true
    xa_data_source_class_name = "com.mysql.cj.jdbc.MysqlXADataSource"
  }
}

生产使用前，应确认每张路由表都能正确解析 ${primary_key}，且目标表具有匹配的主键或唯一键。如果不具备这些前提，应将任务描述为 at-least-once，而不是零重复。

七、一致性校验与监控​

任务上线后，必须使用独立方法验证一致性。应记录源端 binlog 位点等逻辑切面，等待目标端追平后比较固定快照，或者在停止写入的窗口内比较；直接比较持续变化的源端与存在延迟的目标端，既不能证明一致，也不能证明不一致。

数据一致性校验方法​

1. 行数对比：在同一一致性窗口内，按相同主键范围比较记录数

   -- Source database
   SELECT COUNT(*) FROM source_db.users;
   
   -- Target database
   SELECT COUNT(*) FROM target_db.users;
   

2. 确定性分段摘要：按主键顺序读取有界范围内的规范化记录，由校验程序计算 SHA-256 等强摘要

   SELECT id, name, updated_at
   FROM users
   WHERE id >= ? AND id < ?
   ORDER BY id;
   

   哈希前必须为每个字段定义明确的 NULL 标记和无歧义的长度或转义规则，再逐段比较行数与摘要。不要使用 SUM(CRC32(CONCAT_WS(...)))：CRC32 碰撞和 NULL 处理都可能掩盖差异。

3. 按主键下钻：某个范围存在差异时，按主键逐行比较。随机抽样适合定位问题，但不能证明全量一致。

一致性监控指标​

SeaTunnel 任务执行过程中，应监控真实的 Connector 指标和 Checkpoint 信号：

• CDCRecordFetchDelay：CDC 记录抓取阶段观测到的延迟
• CDCRecordEmitDelay：CDC 记录输出阶段观测到的延迟
• Checkpoint 状态：引擎报告的完成、超时和失败信号
• 外部校验结果：由独立校验任务或数据质量平台产生的行数、摘要和逐行差异

“写入成功率”和“数据偏差率”不是 SeaTunnel 内置的一致性证明。如果在外部监控系统中定义这些指标，必须明确时间窗口和分母。

八、最佳实践与性能优化​

以下建议遵循 SeaTunnel 2.3.13 的 Connector 契约。上线前仍需使用具有代表性的数据量和故障场景完成基准与恢复验证。

一致性场景配置建议​

1. 高可靠场景（例如核心业务数据）：

   • 启用 MySQL-CDC exactly_once 和周期性 Checkpoint
   • 仅在驱动与数据库兼容时使用 JDBC XA，并保持 max_retries = 0
   • 配置稳定的目标端主键/唯一键与幂等 Upsert
   • 使用可靠的 Checkpoint 存储，并验证重启、超时和已 Prepare 事务恢复

2. 高性能场景（例如分析类应用）：

   • 使用 CDC 模式 + 批量写入
   • 仅在可以接受 at-least-once 或幂等重放时关闭 XA
   • 增大 batch size
   • 优化并行度设置

3. 大规模初始化场景：

   • 一个任务需要同时覆盖快照与增量时，优先使用 MySQL-CDC initial 模式
   • 仅在具备协调切换方案并能记录对应 binlog 位点时使用 JDBC 初始化
   • 配置合适的分片大小
   • 根据服务器资源调整并行度
   • 不要直接从 JDBC 切换到 CDC；未协调的切换可能产生数据缺口或重叠

常见问题与解决方案​

1. 网络环境不稳定：

   • 调整连接超时和任务级恢复配置
   • 启用 XA exactly-once 时保持 JDBC Sink max_retries = 0
   • 依赖已完成的 Checkpoint，并验证实际重放行为
   • 考虑使用更小的 batch size

2. 高并发写入场景：

   • 根据目标数据库的连接和写入能力调整任务并行度
   • 测量锁与事务压力后，再考虑表分区或增大批次

3. 资源受限环境：

   • 降低并行度
   • 仅在接受更大恢复/重放窗口后增加 Checkpoint 间隔
   • 优化 JVM 内存配置

九、结语：SeaTunnel 的一致性保障之路​

SeaTunnel 为可靠的批流同步提供了必要机制，但最终保障属于完整任务配置和外部系统共同形成的结果。必须同时评估 Source 位点、已完成的 Checkpoint、幂等键以及 Sink 事务。

SeaTunnel 的一致性保障理念可以总结为：

1. Source 恢复状态：CDC 位点或 JDBC Split 状态决定从哪里恢复
2. Checkpoint 协调：已完成的 Checkpoint 对齐可恢复的 Source 与 Sink 状态
3. 明确的 Sink 语义：幂等 Upsert 或受支持的 XA 决定如何处理重放
4. 独立校验：基于一致性窗口的对账用于验证最终结果

满足这些前提时，SeaTunnel 可以在受支持的 Connector 路径上提供零丢失、零重复的交付语义。它不会自动提供跨表或跨数据库原子性，可达到的数据规模和延迟也必须通过具体工作负载验证。

如果你对 SeaTunnel 的数据一致性机制还有更多问题，欢迎加入社区交流。

问一位数据工程师他的管道是 ETL 还是 ELT，几乎总能得到秒答——老派工程师说 ETL，dbt 用户说 ELT。

但这两个答案都不完整。有第三种模式更准确地描述了现代数据管道实际在做的事——它一直存在，只是没有被命名：EtLT。

三种范式​

ETL：落库前先转换​

原始数据从源端提取后，经过专用转换层（Spark、DataStage、Informatica）处理，再写入目标数仓。

源端 ──► [转换层] ──► 目标端

优点： 目标端始终存放干净、业务就绪的数据。合规管控在数据落库前就已强制执行。
缺点： 转换层容易成为瓶颈。Schema 变更需要跨多层同步修改。专用计算集群的运维成本高。

ELT：先落库，原地转换​

由 dbt 带火的模式。原始数据直接落入数仓（BigQuery、Snowflake、ClickHouse），再用 SQL 原地做转换。转换层不再是独立集群——就是数仓本身。

源端 ──► 目标端（原始层）──► [数仓内 SQL] ──► 业务层

优点： 原始数据保留完整审计价值。复用数仓算力。迭代速度快。
缺点： 手机号、身份证、邮箱等敏感字段以明文落库，存在合规风险。数据质量问题要等 Load 之后才可见，此时下游模型可能已经被污染。

EtLT：传输途中做轻量变换​

源端 ──► [tiny t] ──► 目标端（净化原始层）──► [目标端内的 T]

tiny t 是数据在传输过程中完成的一组行级变换：

大 T（Load 后的 Transform）才是真正的业务逻辑所在：多表 JOIN、指标计算、聚合、ML 特征工程。

为什么工程师长期忽略 EtLT​

根本原因是工具，不是概念本身。

传统 ETL 工具（Informatica、DataStage）把转换做得极重，工程师被迫把大量业务逻辑塞进管道，结果管道变得脆弱、难以维护。

现代 ELT 工具（Airbyte、Fivetran）走向另一个极端：源端到目标端几乎没有在途处理，全部交给 dbt。

两类工具都没能填好的空白： 当你同时需要在途操作（脱敏、过滤）和目标端的复杂分析 SQL 时，两个工具族都会让你陷入尴尬的临时方案。

EtLT 天然填补这个空白。

SeaTunnel 天然契合 EtLT​

Apache SeaTunnel 在官方文档中把自己定位为 "EL(T) 数据集成工具"——括号里的 T 是刻意为之。Transform 步骤是轻量的、可选的。这不是营销话术，而是一个架构约束。

三层模型与 EtLT 直接对应​

SeaTunnel 的执行模型分三层：Source → Transform → Sink。

Source（E）
  └──► Transform（tiny t）  ← 可选，轻量处理
        └──► Sink（L）
              └──► [数仓 SQL / dbt]（T）  ← 大 T 在这里

SeaTunnel 对 Transform 能做的事划定了清晰边界。官方文档写道：

"Transform can only be used for some simple transformations of data, such as converting a column to uppercase/lowercase, modifying column names, or splitting one column into multiple columns."

这描述的是内置 Transform 的目标边界。当前 Zeta SQL Transform 不支持 JOIN 和 GROUP BY，因此跨表 JOIN 与聚合应放在目标系统或其他专用处理层中完成。

内置 Transform 覆盖典型 tiny t 需求​

SeaTunnel 同时提供 map 和 flat-map Transform 接口。当前内置 Transform 主要处理单条记录和 Schema，而不是跨行聚合，因此很适合作为 tiny t 层。

EtLT 最关键的战场：CDC 管道​

实时 CDC 同步是 SeaTunnel 的核心使用场景之一——也是纯 ELT 在合规上失守的地方。

根本问题在于：binlog 事件里的字段值一旦落库，就没有第二次机会打码了。 写入 ClickHouse 的手机号或身份证，不可能被下游 dbt 模型"反写"回来。原始明文已经持久化。

EtLT 解决了 ELT 无法解决的问题：在数据到达目标端之前的唯一时间窗口内，完成合规变换。

下面给出一个带 tiny t 的 SeaTunnel CDC 作业最小配置形态。运行前请替换示例中的账号、密码和服务地址：

env {
  job.mode = "STREAMING"
}

source {
  MySQL-CDC {
    plugin_output = "raw_user_info"
    url = "jdbc:mysql://localhost:3306/orders"
    username = "seatunnel"
    password = "change-me"
    server-id = 5601-5604
    table-names = ["orders.user_info"]
  }
}

transform {
  # 在传输途中打码手机号——原始值永远不会到达数仓
  Replace {
    plugin_input = "raw_user_info"
    plugin_output = "masked_user_info"
    replace_field = "phone"
    pattern = "(\\d{3})\\d{4}(\\d{4})"
    replacement = "$1****$2"
    is_regex = true
  }
}

sink {
  # L：落入 ClickHouse 原始层
  Clickhouse {
    plugin_input = "masked_user_info"
    host = "clickhouse-host:8123"
    database = "raw"
    table = "user_info"
    username = "default"
    password = "change-me"
    primary_key = "id"
    support_upsert = true
    allow_experimental_lightweight_delete = true
  }
}

数据进入 ClickHouse 后，再用 SQL 构建宽表、计算留存指标、做分析查询——这就是大 T。dbt 模型在这里是天然的搭档。

SeaTunnel + dbt：全栈 EtLT​

dbt 负责 Load 之后的大 T；SeaTunnel 负责从源到 Load（含 tiny t）。两者互补，而非竞争。

源端
  └── SeaTunnel（E + tiny t + L）
        └── dbt（T）
              └── BI / ML

SeaTunnel 负责数据到达和在途卫生；dbt 负责数据建模和业务逻辑。每个工具只做它擅长的事。

常见陷阱：把大 T 塞进 SeaTunnel​

因为 SeaTunnel 支持 SQL Transform，有些工程师会尝试在里面写 GROUP BY 聚合。当前 Zeta SQL Transform 会明确拒绝 GROUP BY 和 JOIN 查询。

如果确实需要 Load 前的聚合，有两条路：

1. 在目标系统里做——这本来就是 EtLT 和 ELT 中大 T 的用途。
2. 在 SeaTunnel Transform 链路之外使用专用处理作业——此时构建的是完整流式 ETL 管道，而不是把转换留在 tiny t 层。

边界清晰，失败才容易定位。把 tiny t 和大 T 混在同一层，出问题时几乎无法推断根源在哪里。

总结​

SeaTunnel 天然适合 EtLT——不是因为它声称这个标签，而是因为 Source/Transform/Sink 三层分离、当前内置 Transform 的能力边界和明确的"EL(T)"定位都指向同一件事：快速移动数据，在传输途中轻量净化，把重活留给目标系统。

延伸阅读​

• Apache SeaTunnel — About — 官方 EL(T) 定位说明（在页面 "SeaTunnel Work Flowchart" 小节）
• SeaTunnel Transform Connectors — 内置轻量 Transform 算子一览
• dbt — EtLT 中大 T 的标准搭档
• SeaTunnel API Deep Dive — Transform map 接口单行约束详解

同程旅行的数据通道经过多年演进，逐步形成了离线搬运、实时集成、Sqoop 和 SeaTunnel 四套系统并存的格局。每套系统都解决了特定阶段的问题，但功能重叠、执行引擎割裂、运维体系分散，也逐渐成为平台统一治理的障碍。

在 Apache SeaTunnel Meetup 上，负责同程旅行数据平台相关工作的周晓晨，分享了同程旅行以 Apache SeaTunnel Zeta 引擎为统一底座，将四套系统收敛为批流一体数据通道的实践。整个项目有三个不可妥协的目标：迁移过程对业务透明、切换前证明数据一致、同时提升执行效率和运行稳定性。

本文以该场公开分享的事实为依据，独立梳理其统一架构、迁移保障、AI 辅助任务生成、数据校验设计以及后续规划。

过去十年，Modern Data Stack 改变了企业建设数据平台的方式。数据摄取、Lakehouse 或云数仓、dbt、编排系统、指标层、BI 与治理工具，被重新拆分成一组可以灵活组合的模块化能力。它让数据平台从过去那种重型、封闭、强耦合的体系，逐步演化成一个更开放、更云原生、也更容易扩展的现代工程系统。

但当企业真正把这套体系跑到生产环境，另一个事实会越来越明显：整个平台的输入质量、变化速度和可恢复性，往往还是取决于最前面的那一层，也就是数据集成。

也正因为如此，下一代数据工程体系的关键问题，已经不再只是“把数据放到哪里”，而是“数据如何持续、可靠、可治理地进入平台”。从这个角度看，Apache SeaTunnel 的价值，不应该只被理解成一个 connector 工具，而更应该被理解成面向现代数据平台和未来 Agent 系统的 Data Integration Runtime。

一、Modern Data Stack 让平台更模块化了，但也低估了“数据进入平台”这件事​

Modern Data Stack 的贡献非常大。它把传统数据平台中混在一起的一大团能力，拆成了更清晰的层次：数据摄取负责 E，Lakehouse / Warehouse 负责统一存储与计算，dbt 负责大部分建模与转换，编排系统负责任务生命周期，治理、质量、指标与 BI 负责把数据价值推向业务侧。

这种分层让企业更容易选型，也更容易把不同技术组合成适合自己的平台。但在真实生产环境里，最容易被低估的恰恰是最前面的数据摄取层。因为所谓“把数据同步进来”从来都不只是一个连接器问题，它很快会变成一整套持续运行的问题：要不要做 CDC、如何做增量捕获、如何处理 schema drift、如何控制源端压力、如何在失败后恢复、如何避免重复写入、如何让下游接收到更健康的数据。

换句话说，Modern Data Stack 让企业重新认识了 Lakehouse 和建模层的重要性，但今天我们还需要继续往前走一步：重新认识数据集成层本身的工程价值。它决定了后面所有层看到的事实是否完整、及时、一致、可恢复。

二、从 ETL 到 ELT，再到 EtLT：真正变化的是责任边界，不只是术语​

传统 ETL 的逻辑是先 Transform 再 Load。它适合规则明确、边界稳定、变化不快的场景。优势是进入目标系统之前就已经把数据处理得比较规整，但代价是前置链路重、灵活性弱，一旦业务变化频繁，维护成本就会持续上升。

ELT 则把更多事情推到目标端去做：先 Extract，再 Load，最后在 Lakehouse 或云数仓里 Transform。它充分利用了现代底座更强的存储和计算能力，让企业能保留更多原始数据，也让分析建模更灵活。

但 Pure ELT 并不总是最优解。如果什么都先加载进去，再统一处理，那么 schema drift、脏字段、CDC 事件差异、敏感数据、重复记录、格式不一致等问题，也会被原样推进平台。后面的 Lakehouse、dbt、语义层和分析团队，就不得不为前面没有处理好的工程问题买单。

这也是 EtLT 变得重要的原因。EtLT 并不是重新回到重型 ETL，而是在 Load 之前增加一层必要的 lightweight transform：先做轻量标准化、过滤、映射、路由、CDC 事件整理和结构适配，再进入统一底座，最后再由 dbt、SQL 模型和语义层完成真正面向业务的大 Transform。它重新划分的不是字母顺序，而是工程责任边界。

三、SeaTunnel 在 EtLT 里的位置：它更像运行时，而不只是连接器集合​

当我们把视角切到 EtLT，就会发现 SeaTunnel 的定位非常自然。它最适合承担的是 E + 小写 t + L：连接异构数据源、执行 CDC / 增量采集、完成轻量 transform，并把数据稳定送入 Lakehouse、分析引擎或下游存储系统。

这也是 SeaTunnel 和单纯 connector catalog 的区别。企业真正需要的不是“能不能连上”，而是“连上之后能不能持续跑、稳定跑、出错后恢复、处理状态，并把数据准备成下游可消费的样子”。

从工程本质看，更准确的表达不是“SeaTunnel 是一个数据同步工具”，而是“SeaTunnel 是面向现代数据平台的数据集成运行时”。结合不同 connector、执行引擎和配置方式，它可以承接 checkpoint 恢复、schema-aware 处理、可靠交付等模式，这恰恰是 EtLT 最需要的一层。

四、为什么 Pure ELT 会开始失灵：好分析的前提，是先有一次“好落地”​

很多团队刚开始做现代数据平台时，会天然倾向于 Pure ELT：既然 Lakehouse 足够强，那就先全量装进去，后面再慢慢整理。这种思路在早期、简单或探索性场景里没有问题，但只要进入高频变更、CDC、多源异构和持续生产运行阶段，问题就会迅速冒出来。

schema drift 会直接带进平台，让表结构越来越难治理；脏字段和敏感字段未经处理就落地，会抬高治理与合规成本；不同来源的 CDC 事件差异会把复杂度推给下游模型；重复和噪声记录增多后，分析结果会越来越依赖补救性清洗。

EtLT 的意义就在这里。它不是为了把所有业务逻辑都搬回前置层，而是为了先把 landing 这一步做好：把输入标准化、把 schema-aware 处理做好、把 CDC 事件整理清楚、把必要的过滤和路由完成，再把数据送进 Lakehouse。只有这样，下游的 dbt、SQL 模型、语义层和指标层，才不会一开始就建立在混乱输入上。

五、到了 Agent 时代，SeaTunnel 会更像“可被调用的数据执行层”​

如果说 Modern Data Stack 解决的是工具模块化，那么 Agentic Data Stack 解决的就是工具可调用。未来的数据平台不会只是让人手动点界面、配任务、看日志，而会越来越多地让 Agent 围绕目标去理解系统、分析上下文、规划动作、验证结果，并根据反馈不断优化。

在这个闭环里，Agent 负责的是理解与决策：识别目标属于什么业务语境，分析 lineage、依赖和影响范围，判断该走 ETL、ELT 还是 EtLT，决定调用哪些工具、承担哪些风险。

SeaTunnel 负责的则是把这些决策变成真正可执行的数据动作。更准确地说，它可以通过执行 API、编排集成或更高层控制层，承接同步、CDC 执行、受控重跑、选择性重载以及可恢复交付等动作。Agent 只有在能够调用可靠的数据执行系统时，才真正具备“行动力”，而这正是运行时层的重要性所在。

六、最终形态：SeaTunnel 不只是 Modern Data Stack 的一个组件，而是 Agentic Data Stack 的运行时基础设施之一​

如果用分层视角来看下一代数据工程平台，我们大致会看到这样一套结构：最底层是 Data Sources；其上是 Data Integration Runtime；再往上是 Unified Data Foundation；然后是 Transformation & Semantics；再上面是 Orchestration & Governance；最上层才是 Agent Layer。

在这个结构里，SeaTunnel 所在的位置非常关键。它处于“数据源”和“统一数据底座”之间，负责把原始世界里的异构变化，转化成平台能够持续承接的标准化输入。没有这一层，Lakehouse 更像一个被动接收数据的容器；有了这一层，Lakehouse 才更有机会成为稳定的统一事实底座。

也正因为如此，SeaTunnel 不应该只被理解成某个时代里的“同步工具”。在 Modern Data Stack 里，它承担的是更真实、更生产级的数据集成职责；到了 Agentic Data Stack 里，它进一步成为让数据工程系统从“能分析”走向“能行动”的运行时基础设施之一。

结语​

从 Modern Data Stack 走向 Agentic Data Stack，真正变化的并不是多了一个大模型界面，而是数据工程的操作范式正在改变：过去是人定义流程、系统执行流程；未来会越来越变成人定义目标、系统生成动作、工具执行动作、反馈再回流到系统。

在这个变化里，Lakehouse 很重要，dbt 很重要，编排和治理也都很重要。但如果要问“谁负责把真实世界的数据变化，持续、可靠、可恢复地带入平台”，答案仍然会指向数据集成运行时。

这正是 Apache SeaTunnel 的独特价值所在。它不是简单的连接器列表，而是让整个数据平台真正运转起来的一层关键基础设施。

过去二十年，企业数据工程体系一直建立在一个默认前提之上：

人负责理解系统，系统负责执行流程。

工程师理解业务，拆解链路，编写 SQL、Spark、Shell、同步脚本或调度任务，然后交给系统执行。调度系统不需要理解业务，只需要按照 DAG 把任务跑完；数据同步系统不需要理解指标，只需要把数据从源端搬到目标端。

这套模式支撑了很长时间的数据仓库、数据湖、BI、报表和批量调度体系。

但今天，这个前提正在开始失效。

企业数据系统变得越来越复杂：数据源更多，链路更长，实时性更强，业务变化更快，指标口径更容易冲突，AI 应用又不断带来新的交互数据、模型反馈数据、向量索引和非结构化数据。

在这样的环境下，企业真正缺少的，已经不只是更多 Pipeline，也不只是一个更会写 SQL 的 Copilot。

企业真正需要的是：

能够理解系统、规划任务、调用工具、验证结果，并持续积累经验的 Data Engineering Agent。

而在这个变化中，Apache SeaTunnel 的角色会变得非常关键。

因为 Agent 时代的数据工程，不只是“会思考”就够了。它还必须能稳定连接数据源、捕获变化、执行同步、处理增量、保障一致性，并把数据安全、可靠、低成本地移动到目标系统。

换句话说：

Agent 负责理解目标和规划动作，SeaTunnel 负责把这些动作变成真实、可靠、可恢复的数据工程执行。

这也是为什么从 ETL、ELT、EtLT 到 Agent 的演进里，SeaTunnel 会成为下一代数据工程体系的重要执行底座。

一、从 ETL 到 ELT：第一次范式迁移​

传统 ETL 的逻辑很清楚：

1. 先 Extract，把数据从源端抽出来。
2. 再 Transform，在中间层完成清洗、转换、聚合。
3. 最后 Load，把处理后的结果加载到目标系统。

这套模式适合早期数据仓库时代。

因为当时数据源相对有限，链路相对清楚，计算资源也比较集中。企业希望在进入数仓之前，把脏数据清洗掉，把口径处理好，把结构整理成可以直接使用的结果。

ETL 本质上是一种确定性 Pipeline。

它的核心假设是：

人提前定义好流程，系统按流程执行。

后来，随着云数仓、数据湖、Lakehouse 和弹性计算的发展，ELT 开始流行。

ELT 的思路是：

1. 先 Extract。
2. 再 Load。
3. 最后在目标系统里 Transform。

也就是说，先把原始数据或近原始数据装进统一存储底座，再利用目标端强大的计算能力做转换。

ELT 解决了 ETL 的一些问题。比如，它降低了前置处理的复杂度，保留了更多原始数据，也让后续建模和分析更加灵活。

但 ELT 也带来了新的问题。

如果所有转换都推迟到目标端，源端数据的脏问题、类型问题、字段变化、CDC 事件、隐私字段、格式差异、结构漂移，都会被直接带进目标系统。

在简单批量场景里，这还可以接受；但在今天的实时同步、CDC、多表同步、湖仓入湖、SaaS API 接入和 AI 数据工程场景里，完全“先 Load 再说”会让下游治理成本急剧上升。

于是，企业数据工程开始进入第三种形态：

EtLT。

二、EtLT：从 Pipeline 到 Agent 的关键过渡形态​

这里的 EtLT，不是简单把 ETL 和 ELT 混在一起。

我更愿意把它理解为：

Extract -> lightweight transform -> Load -> semantic Transform

也就是：

• 抽取数据。
• 先做一层必要的轻量 transform。
• 再把数据加载到统一数据底座。
• 最后做面向业务语义、指标口径和数据产品的大 Transform。

这里最关键的是小写 t 和大写 T 的区别。

小写 t 不是复杂业务建模，而是数据进入平台之前必须完成的工程化处理，例如：

• 字段裁剪
• 类型映射
• 格式标准化
• 主键或分区字段处理
• 敏感字段脱敏
• CDC 事件格式转换
• 多表路由
• Schema Evolution
• 数据质量前置校验
• 一读多写
• 限速和并行控制

这些处理不应该全部推迟到目标端。否则，数据湖、数仓或 Lakehouse 里会堆积大量结构不一致、语义不清楚、质量不可控的原始数据。

但小写 t 也不应该承载过重的业务逻辑。

真正复杂的业务口径、指标定义、主题建模、语义映射和跨域聚合，应该放到大写 T 里，在 Lakehouse、数仓、指标层、语义层和治理体系中完成。

所以 EtLT 的核心价值在于：

在数据进入统一底座之前，先完成必要的工程标准化；在数据进入统一底座之后，再完成业务语义化。

这正是 SeaTunnel 非常适合承担的部分。

Apache SeaTunnel 的 Source、Transform、Sink 架构，本质上天然适合 EtLT 里的小写 t：它既能连接多种异构数据源，也能在数据流动过程中完成轻量转换、结构适配、CDC 处理、多表同步和写入目标系统。

因此，在 EtLT 架构里，SeaTunnel 不只是一个“数据搬运工具”，而是企业数据进入统一底座之前的 Data Integration Runtime。

三、为什么传统 ETL 开始吃力​

传统 ETL 面对的是相对确定的流程。

把规则写清楚，DAG 画出来，任务按顺序执行，失败之后工程师排查修复。

但今天的数据工程环境已经不是当年的样子。

企业里同时存在 OLTP 数据库、Kafka 消息流、CDC 链路、SaaS API、日志、对象存储、Lakehouse、实时 OLAP、向量数据库、AI 交互日志和模型结果数据。

数据不只是变多了，也变得更碎、更异构、更实时。

更麻烦的是链路长度。

一个看上去很普通的经营指标，背后可能经过几十张表、多层宽表、多个业务域拼接，还有一堆复杂口径转换。到了这个阶段，很多企业面临的真实问题已经不是“流程有没有搭起来”，而是几乎没人能把整条链路从头到尾讲明白。

传统 ETL 在这里暴露出来的是结构性问题。

• 一个字段改名，可能带崩上百个任务。
• 一个枚举值变化，可能让多个核心指标悄悄漂移。
• 一段增量逻辑写错，影响的可能不只是单张表，而是一串下游分析结果。

调度系统能告诉你任务失败了，但它不一定知道为什么失败。 同步工具能把数据写过去，但它不一定知道这份数据影响了哪个业务指标。 工程师能修复脚本，但前提是他先能找到完整上下文。

所以传统 ETL 真正吃力的地方，不只是性能或稳定性，而是它天然缺少系统级认知能力。

它能执行流程，但不能理解系统。

四、为什么 Copilot 没有真正解决问题​

很多团队第一次把 AI 引入数据工程，往往是从 Copilot 开始：写 SQL、补 Spark 代码、生成 YAML、写测试样例。

这些能力当然有价值，尤其能提高局部开发效率。

但它们没有触达企业数据工程最深层的问题。

因为数据工程真正难的，从来不只是代码生成，而是系统理解。

Copilot 可以帮你生成一段 SQL，却不知道这个字段的业务含义是什么；可以帮你写一个同步任务，却不知道上游 schema change 会影响哪些下游指标；可以帮你生成调度配置，却不知道这次变更是否破坏了历史一致性。

企业数据工程真正复杂的部分，是：

• lineage reasoning
• dependency analysis
• semantic understanding
• metric governance
• risk estimation
• impact analysis
• incremental recovery

这些不是单靠代码补全就能解决的。

所以企业真正需要的，不只是一个 AI IDE，而是一套能理解数据系统、围绕目标拆解任务、调用工程工具并验证结果的 Agentic Data Engineering 系统。

五、从 Pipeline 到 Agent，真正变化是什么​

如果只保留一句核心判断，我会这样说：

传统 ETL 的核心是：人定义流程，系统执行流程。Agent 数据工程的核心是：人定义目标，系统生成流程。

这不是一句包装口号，而是两套系统组织方式的根本差异。

在传统模式里，工程师先设计好任务链路，配置 Source、Transform、Sink，再交给调度系统执行。

系统面对的是一个固定流程。

而在 Agent 模式里，人输入的可能只是一个目标。

比如：

新增一个订单利润指标，并保证和财务口径一致。

传统做法是工程师自己去找数据源、查表结构、看血缘、写转换逻辑、配置同步任务、补质量校验、发起回归测试。

Agent 理想中的工作方式，则是围绕这个目标自动拆解一组动作：

• 识别涉及哪些业务对象
• 查找候选数据源
• 分析上游 lineage
• 判断应该走 ETL、ELT 还是 EtLT
• 生成 SeaTunnel 同步任务
• 配置 CDC 或全量同步
• 完成轻量 transform
• 写入 Lakehouse 或目标系统
• 触发质量校验
• 评估下游影响
• 把结果反馈给工程师确认

这里真正变化的，不是“AI 帮你写了一段 SQL”，而是系统开始围绕目标生成工程动作。

但这也带来一个关键问题：

Agent 规划出来的数据动作，谁来可靠执行？

这正是 SeaTunnel 的价值所在。

六、SeaTunnel 在 Agent 时代的位置：Agent 的数据集成执行层​

Agent 不能只停留在分析和建议阶段。

如果一个 Data Engineering Agent 发现某张表需要同步、某条链路需要回放、某个 CDC 任务需要调整、某批数据需要重新写入目标系统，它必须能调用一个稳定、可控、可观测的数据集成执行层。

这个执行层需要具备几类能力。

1. 能连接足够多的数据源​

企业数据系统天然异构。Agent 不可能只面对一种数据库或一种文件系统。它需要连接 MySQL、Oracle、PostgreSQL、SQL Server、Kafka、Hive、Iceberg、Doris、ClickHouse、StarRocks、Elasticsearch、S3、HDFS、MongoDB、SaaS API 等不同系统。

SeaTunnel 的 Connector API 和插件化架构，正好解决了这个问题。它把 Source、Transform、Sink 抽象成统一接口，让不同数据源以相对一致的方式被接入和调用。

2. 能同时处理批、流、CDC 和整库同步​

Agent 时代的数据工程不是单一范式。它既需要一次性全量迁移，也需要持续 CDC；既需要离线批处理，也需要实时同步；既需要单表同步，也需要多表或整库级别的数据移动。

SeaTunnel 的价值就在于，它不是单纯的 ETL 脚本工具，而是面向数据同步和数据集成场景构建的运行体系，可以承载全量、增量、实时、CDC、多表同步等多种数据移动任务。

3. 能承载 EtLT 里的轻量 transform​

EtLT 不是把所有业务逻辑都塞进同步链路，而是在数据进入统一底座之前，先完成必要的工程标准化。

SeaTunnel 的 Transform 能力，适合处理字段映射、类型转换、数据过滤、字段裁剪、格式调整、脱敏、路由等轻量处理。

这些能力让 SeaTunnel 可以承担 EtLT 中的小写 t：

不做过重的业务建模，但把数据整理到可治理、可加载、可继续加工的状态。

4. 能提供一致性、容错和恢复能力​

Agent 可以判断“应该重跑这一段链路”，但真正执行重跑时，底层系统必须具备 Checkpoint、失败恢复、状态管理和一致性保障能力。

一个只会规划、不可靠执行的系统，最终只是“会想但不会做”。

所以在 Agent 时代，执行质量仍然和智能水平同样重要。

七、未来的数据工程栈会越来越像“操作系统”​

如果再往前看一步，企业数据工程体系会越来越像一套分层的数据操作系统，而不再只是很多零散 Pipeline 的集合。

在这套体系里：

• Semantic Layer 负责定义业务世界模型
• Metadata 负责提供结构化上下文
• Memory 负责沉淀经验
• Planning Layer 负责把目标拆成动作
• Execution Layer 负责真正执行同步、CDC、数据移动、回放和恢复

SeaTunnel 就处在这个 Execution Layer 里。

这个位置非常关键。

未来不是“在 ETL 上面套一个大模型”就结束了。

未来是一个清晰分层、各司其职的系统：

• Agent 决定应该做什么
• SeaTunnel 保障这件事真的被可靠执行出来

八、用一句话总结这个演进​

ETL 时代，企业构建的是数据流水线。

ELT 时代，企业开始把数据沉淀到统一底座。

EtLT 时代，企业开始重新平衡数据接入、轻量治理和业务转换之间的关系。

Agent 时代，企业要构建的更像是一套数据操作系统。

在这套系统里，Semantic Layer 提供业务世界模型，Metadata 提供上下文，Memory 沉淀经验，Planning Layer 负责决策，SeaTunnel 这样的数据集成引擎负责把数据移动、CDC、轻量 transform、增量同步、回放和恢复真正执行出来。

所以，Agent 不是在 ETL 外面简单挂一个大模型，也不是给调度系统加一层聊天入口。

它真正改变的是企业数据工程的底层操作范式。

而 SeaTunnel 的价值，也会在这个过程中被重新定义。

它不只是一个数据同步工具，而是下一代 Agentic Data Engineering 体系里的数据集成执行底座。

Agent 让数据系统开始理解目标。

EtLT 让数据进入平台的过程更加可控。

SeaTunnel 让这些目标真正变成可靠的数据工程动作。

这就是从 ETL、ELT、EtLT 到 Agent 时代，企业数据工程正在发生的根本变化。

在数据集成与同步领域，Apache SeaTunnel 无疑是当下最炙手可热的工具之一。本系列将深挖其高级用法。

首篇从 SeaTunnel 核心概念“数据流”切入，剖析底层原理，如数据流动与转换机制，结合实例讲解在复杂场景中的应用，助你掌握这一工具。

一句话总结（先给结论）​

SeaTunnel 不是“source → sink”的线性工具

• 它是一个 “数据流（DataStream / DataFlow）驱动的 DAG 执行引擎”

两个 source 可以流入一个 sink，正是这个模型的直接体现。

一、SeaTunnel 的核心概念：数据流​

在 SeaTunnel 内部，一切围绕“数据流”展开。

数据流是什么？​

数据流 = 一组结构一致的 Record 流（带 Schema）

它不是表、不是文件、不是 SQL 结果 而是：

Record1 → Record2 → Record3 → ...

每个插件都在“操作数据流”​

二、plugin_output / plugin_input 的真实含义（非常重要）​

你之前一直在“用”，但现在该“理解”它了。

1️⃣ plugin_output​

plugin_output = "source_data_output_1"

含义不是“名字”，而是：

给当前插件产生的数据流起一个唯一 ID

可以理解为：

DataStream<ID = source_data_output_1>

2️⃣ plugin_input​

plugin_input = "source_data_output_1"

含义是：

我这个插件，要消费哪个数据流

用一句话说透​

plugin_output / plugin_input = 数据流的“连线端口”

三、SeaTunnel 的 DAG 模型（你现在已经用到了）​

你这个成功的实验，本质上是：

SourceA ─┐
         ├──► Sink
SourceB ─┘

SeaTunnel 内部会构建这样的 DAG：​

DataStream A ─┐
              ├──► Sink Operator
DataStream B ─┘

关键点：为什么能合并？​

因为：

Sink 不是“绑定一个 source”，而是“订阅一个或多个数据流”

当你写：

sink {
  jdbc {
    plugin_input = "a,b"
  }
}

或你用了多个 source 但最终都连到同一个 sink，SeaTunnel 内部会：

• 把多个输入流
• 合并成一个 Logical Input
• 按 Record 顺序写入

⚠️ 注意：

• 不是 join
• 不是 union SQL
• 是流级别的合并（append）

四、这和你理解的 “SQL / ETL” 有什么本质区别？​

这是很多人会迷糊的地方。

SQL 的世界​

SELECT * FROM A
UNION ALL
SELECT * FROM B

• 是 “结果集语义”

SeaTunnel 的世界​

A 的 Record 流
B 的 Record 流
↓
Sink 持续消费

• 是 “流语义”

只要 Schema 一致，就能进同一个 sink。

五、Schema 在数据流里的地位（你一定要记住）​

数据流 = Record + Schema

SeaTunnel 合流的前提：​

• 字段数一致
• 字段类型兼容
• 字段名对齐（或可映射）

否则：

• 运行期直接报错
• 或 Sink 写入失败

你刚才说“目标字段肯定是对齐的”，这就是为什么你的实验能成功。

六、SeaTunnel 的“数据流转模型”正式定义（给你一个标准版）​

你以后在设计 / 讲方案 / 写文档，可以直接用这段话：

SeaTunnel 以 数据流（DataStream） 为核心抽象。 Source 插件负责生成数据流，Transform 插件对数据流进行处理并输出新的数据流，Sink 插件消费一个或多个数据流并将数据写入外部系统。 多个数据流可以在 Sink 处汇聚，只要其 Schema 兼容，SeaTunnel 会以流合并（append）的方式进行写入，而不是关系型 join。

七、这对你 Builder / Strategy 设计的直接影响（重点）​

你现在可以非常确定三件事：

1️⃣ Builder 必须支持 N Source → M Sink​

不是 1→1，而是 图模型。

2️⃣ plugin_output 是“一等公民”​

你现在 Builder 里如果有人不设置 plugin_output：

• 你应该帮他自动生成一个

这是平台级能力。

3️⃣ Sink 允许多个输入流（逻辑）​

即使 DSL 上写的是一个：

plugin_input = "s1"

你 Builder 的语义应该是：

Set<DataStream>

而不是 String。

八、你现在已经“踩过并验证”的几个关键事实​

我帮你总结一下你已经实证的结论：

✅ SeaTunnel 是 DAG，不是线性 ETL ✅ 多 Source 可以汇入一个 Sink ✅ 合并是流合并，不是 SQL join ✅ Schema 对齐是前提 ✅ DSL 是数据流描述，不是 SQL

九、总结​

SeaTunnel 的核心只有 3 个角色​

Source     →   Transform   →   Sink
（产生流）     （改流）        （吃流）

数据流怎么“连”？​

你只要记住这张“万能规则表”

靠的就是这两个东西：

• plugin_output ：我产生的这条数据流叫什么
• plugin_input ：我要吃哪条（或哪几条）数据流

比如，两个 source → 一个 sink

┌──────────┐
│ Source A │──┐
└──────────┘  │
               ├──▶  Sink
┌──────────┐  │
│ Source B │──┘
└──────────┘

一个 Source → 两个 Sink

        ┌──────▶ Sink A
Source ─┤
        └──────▶ Sink B

一个 conf 里放两组“互不干扰”的流

Source A ───▶ Sink A
 
Source B ───▶ Sink B

在大规模数据集成场景中，吞吐瓶颈往往不在数据通道本身，而在“元数据路径”上：启动时的 Connector/Jar 加载、运行中的状态管理与恢复、以及初始化阶段对外部系统（如数据库、Hive Metastore）的 Schema/分区查询。任务量一旦上到千级、万级，这些“看似轻量”的动作会被放大成集群级别的压力。

Apache SeaTunnel Engine（Zeta）把一部分高频、可复用且昂贵的元数据下沉到引擎侧进行缓存，并配合分布式存储与自动清理策略，让海量同步任务可以更稳定地并行运行。

为什么“元数据”会成为瓶颈​

以“万级小作业”并行启动为例，常见的元数据瓶颈主要来自三类：

• 类加载与依赖隔离：每个作业独立创建 ClassLoader，会反复加载同一批 Connector 依赖，快速抬升 JVM Metaspace 压力。
• 状态与恢复信息：Checkpoint、任务状态、历史作业信息等若缺少分层存储与自动清理，会带来内存与 IO 的双重负担。
• 外部目录/Schema 查询：作业初始化阶段对源端数据库或 Metastore 的频繁请求，容易造成连接拥塞与元数据服务抖动。

下面从三条主线拆解 SeaTunnel 的“元数据缓存”思路与可落地的配置建议。

一：ClassLoader 缓存，降低 Metaspace 压力​

当大量作业复用相同的 Source/Sink Connector 时，持续创建和销毁类加载器会带来明显的 Metaspace 抖动，甚至触发溢出。SeaTunnel Engine 提供 classloader-cache-mode，用于复用作业之间的 ClassLoader，减少重复加载和频繁回收的开销。

在 seatunnel.yaml 中开启（该配置默认开启，若你曾手动关闭可重新启用）：

seatunnel:
  engine:
    classloader-cache-mode: true

适用场景：

• 作业规模大、启动频繁，且 Connector 类型相对有限（复用率高）。
• JVM Metaspace 频繁增长，或出现与类加载相关的内存告警。

注意点：

• 如果集群长期运行且 Connector 类型非常分散，缓存会增加常驻的类元数据占用；建议结合监控观察 Metaspace 曲线，再决定是否开启或调整作业结构。

二：分布式状态与持久化，保证可恢复与可运营​

SeaTunnel Engine 的容错语义基于 Chandy–Lamport Checkpoint 思想。为了兼顾性能与可靠性，它在引擎内部使用 Hazelcast 的分布式数据结构（如 IMap）承载一部分运行态信息，并通过外部存储（共享/分布式存储）完成故障恢复所需的数据落盘。

你通常需要关心三组配置：

1) Checkpoint 触发参数​

seatunnel:
  engine:
    checkpoint:
      interval: 300000
      timeout: 10000

说明：如果在作业配置文件 env 中配置了 checkpoint.interval/checkpoint.timeout，会优先以作业配置为准。

2) IMap 备份与持久化（建议用于生产集群）​

当集群节点数大于 1 时，建议至少配置 backup-count，以降低单点故障导致的内存态信息丢失风险；对于需要“全停全启后自动恢复”的场景，可进一步配置 IMap 外部持久化。

相关细节可参考文档：

• /docs/seatunnel-engine/deployment
• /docs/seatunnel-engine/checkpoint-storage

3) 历史作业信息的自动清理​

SeaTunnel 将已完成作业的状态、计数器、错误日志等信息存放在 IMap 中。作业越多，累积越快。建议按需配置 history-job-expire-minutes，让过期信息自动淘汰，避免内存长期膨胀（默认 1440 分钟，即 1 天）。

seatunnel:
  engine:
    history-job-expire-minutes: 1440

三：Catalog/Schema 元数据缓存，减少源端压力​

大量作业并行启动时，对外部系统的元数据请求（表结构、分区信息、约束信息等）很容易成为“隐形风暴”。SeaTunnel 在 Connector/Catalog 侧引入缓存与复用思路，尽量把高频查询前置到引擎侧，减少重复的网络往返与服务端解析开销。

• JDBC 场景：初始化阶段会读取表结构、字段类型、主键等信息，用于校验与分片规划。建议在高并发启动时避免每个作业对同一张表重复拉取全量元数据（可通过作业编排层做批次启动/预热）。
• Hive 场景：Hive Metastore 往往是共享服务且相对敏感，建议尽量复用 Catalog 实例与已加载的 Database/Table/Partition 信息，并在大规模分区表同步中关注 Metastore 的 QPS 与响应时间。

与 Flink / Spark 的差异：面向“海量小作业”的轻量化​

Flink 的设计重心是长生命周期的流作业与复杂算子状态；Spark 更偏向批处理与作业级 Context 管理。在“万级独立小任务并发”这个目标下，SeaTunnel Engine 的策略更强调把可复用的启动与运行元数据沉到引擎层：减少重复加载、减少重复查询、并对历史信息进行可控的生命周期管理，从而提升并发启动与稳定性。

生产落地建议​

• 启用合理备份：生产集群建议 backup-count >= 1，并评估是否需要 IMap 外部持久化以支持全停全启自动恢复。
• 收敛 Connector 类型：尽量在同一集群里控制 Connector 组合的离散程度，让 classloader-cache-mode 的收益最大化。
• 关注“元数据指标”：除了 JVM 指标，建议关注 Checkpoint 延迟/失败率、Hazelcast 内存使用、IMap 大小与增长速率、历史作业累积速度等。
• 配置过期策略：根据排障与审计需求设置 history-job-expire-minutes，避免“为了可观测性而撑爆内存”。

Amazon Aurora DSQL是亚马逊云科技于2024年12月推出的分布式SQL数据库，专为构建扩展性无限、高可用且免基础设施管理的应用程序设计，具有可用性高、无服务器模式架构、兼容性强、容错能力和安全级别高等特点。

由于Aurora DSQL的认证机制与IAM集成， 访问Aurora DSQL数据库需要通过IAM的身份来生成token 进行访问，而token 默认只有15分钟有效期，因此目前一些主流的数据同步工具暂不支持将其他数据库的数据迁移到Aurora DSQL。

基于这种情况，本文作者基于数据同步工具Apache SeaTunnel开发了一个专门针对Aurora DSQL的sink Connector，以满足从其他数据库迁移数据到Aurora DSQL需求。

SeaTunnel 介绍​

SeaTunnel是一个非常易用、多模态、超高性能的分布式数据集成平台，专注于数据集成和数据同步，主要旨在解决数据集成领域的常见问题。

SeaTunnel 相关特性​

• 丰富且可扩展的Connector： 目前，SeaTunnel 支持超过 190 个Connector且数量还在增加，像主流数据库MySQL 、Oracle、SQLServer、PostgreSQL等都已经提供了Connector支持。插件式设计让用户可以轻松开发自己的Connector并将其集成到SeaTunnel项目中。
• 批流集成：基于SeaTunnel Connector API开发的Connector完美兼容离线同步、实时同步、全量同步、增量同步等场景。 它们大大降低了管理数据集成任务的难度。
• 分布式快照：支持分布式快照算法，保证数据一致性。
• 多引擎支持：SeaTunnel默认使用SeaTunnel引擎（Zeta）进行数据同步。 SeaTunnel还支持使用Flink或Spark作为Connector的执行引擎，以适应企业现有的技术组件。 SeaTunnel 支持 Spark 和 Flink 的多个版本。
• JDBC复用、数据库日志多表解析：SeaTunnel支持多表或全库同步，解决了过度JDBC连接的问题； 支持多表或全库日志读取解析，解决了CDC多表同步场景下需要处理日志重复读取解析的问题。
• 高吞吐量、低延迟：SeaTunnel支持并行读写，提供稳定可靠、高吞吐量、低延迟的数据同步能力。
• 完善的实时监控：SeaTunnel支持数据同步过程中每一步的详细监控信息，让用户轻松了解同步任务读写的数据数量、数据大小、QPS等信息。

SeaTunnel 工作流程​

图一 Seatunnel工作流图

SeaTunnel的工作流程如上图所示，用户配置作业信息并选择提交作业的执行引擎。Source Connector负责并行读取源端数据并将数据发送到下游Transform或直接发送到Sink，Sink将数据写入目的地。

从源码构建SeaTunnel​

git clone https://github.com/apache/seatunnel.git
cd seatunnel
sh ./mvnw clean install -DskipTests -Dskip.spotless=true
cp seatunnel-dist/target/apache-seatunnel-${version}-bin.tar.gz /The-Path-You-Want-To-Copy
cd /The-Path-You-Want-To-Copy
tar -xzvf "apache-seatunnel-${version}-bin.tar.gz"

从源码构建成功后，所有的Connector插件和一些必要的依赖（例如：mysql驱动）都包含在二进制包中。您可以直接使用Connector插件，而无需单独安装它们。

使用Seatunnel同步MySQL数据到Aurora DSQL 配置示例​

env {
  parallelism = 1
  job.mode = "STREAMING"
  checkpoint.interval = 6000
  checkpoint.timeout = 1200000
}
source {
  MySQL-CDC {
    username = "user name"
    password = "password"
    table-names = ["db.table1"]
    url = "jdbc:mysql://dbhost:3306/db?useSSL=false&allowPublicKeyRetrieval=true&serverTimezone=UTC&connectTimeout=120000&socketTimeout=120000&autoReconnect=true&failOverReadOnly=false&maxReconnects=10"
    table-names-config = [
      {
        table = "db.table1"
        primaryKeys = ["id"]
      }
    ]
  }
}
transform {

}
sink {
    Jdbc {
        url="jdbc:postgresql://<dsql_endpoint>:5432/postgres"
        dialect="dsql"
        driver = "org.postgresql.Driver"
        username = "admin"
        access_key_id = "ACCESSKEYIDEXAMPLE"
        secret_access_key = "SECRETACCESSKEYEXAMPLE"
        region = "us-east-1"
        database = "postgres"
        generate_sink_sql = true
        primary_keys = ["id"]
        max_retries="3"
        batch_size =1000
    }
}

运行数据同步任务​

将上面的配置保存为mysql-to-dsql.conf 文件(请注意需要将示例中的值替换为真实的参数)，存放在apache-seatunnel-${version} 的config 目录下，执行以下命令:

cd "apache-seatunnel-${version}"
./bin/seatunnel.sh --config ./config/mysql-to-dsql.conf -m local

图二 数据同步日志信息

命令执行成功后，您可以通过新产生的日志观察任务执行情况，如果出现错误，也可以根据异常信息进行定位，比如数据库连接超时、表不存在情况。而正常情况下，数据会成功写入目标 Aurora DSQL，如上图所示。

总结​

Aurora DSQL是一款高度安全、易扩展、无服务器基础设施的分布式数据库，它的认证方式与IAM身份结合，因此目前缺少合适的工具可以将数据同步到Aurora DSQL中，尤其是在实时数据同步方面。SeaTunnel 是一款非常优秀数据集成和数据同步工具，目前支持多种数据源的数据同步，并且基于SeaTunnel 也可以非常灵活地实现自定义的数据同步需求，比如全量同步/增量实时同步。基于这种灵活性，本文作者开发了一种专门针对于Aurora DSQL 的Sink Connector， 以满足对于Aurora DSQL 数据同步需求。

参考文档​

• SeaTunnel 部署：https://seatunnel.apache.org/zh-CN/docs/start-v2/locally/deployment

• 开发新的SeaTunnel Connector：

• https://github.com/apache/seatunnel/blob/dev/seatunnel-connectors-v2/README.zh.md

• 在Aurora DSQL 中生成身份验证令牌：https://docs.aws.amazon.com/aurora-dsql/latest/userguide/SECTION_authentication-token.html

*前述特定亚马逊云科技生成式人工智能相关的服务目前在亚马逊云科技海外区域可用。亚马逊云科技中国区域相关云服务由西云数据和光环新网运营，具体信息以中国区域官网为准。

本篇作者​

谭志强，亚马逊云科技迁移解决方案架构师，主要负责企业级客户的上云或跨云迁移工作，具有十几年 IT 专业服务经验，历任程序设计师、项目经理、技术顾问、解决方案架构师。

Apache SeaTunnel (Incubating) 与 Apache Inlong (Incubating) 的5月联合Meetup中，第二位分享的嘉宾是来自白鲸开源的高级工程师李宗文。在使用Apache SeaTunnel (Incubating) 的过程中，他发现了 Apache SeaTunnel (Incubating) 存在的四大问题：Connector实现次数多、参数不统一、难以支持多个版本的引擎以及引擎升级难的问题。为了解决以上的难题，李宗文将目标放在将Apache SeaTunnel (Incubating)与计算引擎进行解耦，重构其中Source与Sink API，实现改良了开发体验。

本次演讲主要包含四个部分：

1. Apache SeaTunnel (Incubating)重构的背景和动机
2. Apache SeaTunnel (Incubating)重构的目标
3. Apache SeaTunnel (Incubating)重构整体的设计
4. Apache SeaTunnel (Incubating) Source API的设计
5. Apache SeaTunnel (Incubating) Sink API的设计

李宗文

白鲸开源 高级工程师

Apache SeaTunnel(Incubating)

& Flink Contributor, Flink CDC & Debezium Contributor

01 重构的背景与动机​

01 Apache SeaTunnel(Incubating)与引擎耦合​

用过Apache SeaTunnel (Incubating) 的小伙伴或者开发者应该知道，目前Apache SeaTunnel (Incubating) 与引擎完全耦合，完全基于Spark、Flink开发，其中的配置文件参数都基于Flink、Spark引擎。从贡献者和用户的角度出发，我们能发现一些问题。

从贡献者的角度：反复实现Connector，没有收获感；潜在贡献者由于引擎版本不一致无法贡献社区；

从用户的角度：目前很多公司采用Lambda架构，离线作业使用Spark，实时作业使用Flink， 使用中就会发现SeaTunnel 的Connector可能Spark有，但是Flink没有，以及两个引擎对于同一存储引擎的Connector的参数也不统一，有较高的使用成本，脱离了SeaTunnel简单易用的初衷；还有用户提问说目前支不支持Flink的1.14版本，按照目前SeaTunnel的架构，想要支持Flink的1.14就必须抛弃之前的版本，因此这也会对之前版本的用户造成很大的问题。

因此，我们不管是做引擎升级或者支持更多的版本的用户都很困难。

另外Spark和Flink都采用了Chandy-lamport算法实现的Checkpoint容错机制，也在内部进行了DataSet与DataStream的统一，以此为前提我们认为解耦是可行的。

02 Apache SeaTunnel(Incubating)与引擎解耦​

因此为了解决以上提出的问题，我们有了以下的目标：

1. Connector只实现一次：针对参数不统一、Connector多次实现的问题，我们希望实现一个统一的Source 与Sink API;
2. 支持多个版本的Spark与Flink引擎：在Source与Sink API上再加入翻译层去支持多个版本与Spark和Flink引擎，解耦后这个代价会小很多。
3. 明确Source的分片并行逻辑和Sink的提交逻辑：我们必须提供一个良好的API去支持Connector开发；
4. 支持实时场景下的数据库整库同步：这个是目前很多用户提到需要CDC支持衍生的需求。我之前参与过Flink CDC社区，当时有许多用户提出在CDC的场景中，如果直接使用Flink CDC的话会导致每一个表都持有一个链接，当遇到需要整库同步需求时，千张表就有千个链接，该情况无论是对于数据库还是DBA都是不能接受的，如果要解决这个问题，最简单的方式就是引入Canal、Debezium等组件，使用其拉取增量数据到Kafka等MQ做中间存储，再使用Flink SQL进行同步，这实际已经违背了Flink CDC最早减少链路的想法，但是Flink CDC的定位只是一个Connector，无法做全链路的需求，所以该proposal在Flink CDC社区中没有被提出，我们借着本次重构，将proposa提交到了SeaTunnel社区中。
5. 支持元信息的自动发现与存储：这一部分用户应该有所体验，如Kafka这类存储引擎，没有记录数据结构的功能，但我们在读取数据时又必须是结构化的，导致每次读取一个topic之前，用户都必须定义topic的结构化数据类型，我们希望做到用户只需要完成一次配置，减少重复的操作。

可能也有同学有疑惑为什么我们不直接使用Apache Beam，Beam的Source分为BOUNDED与UNBOUNDED，也就是需要实现两遍，并且有些Source与Sink的特性也不支持，具体所需的特性在后面会提到；

03 Apache SeaTunnel(Incubating)重构整体的设计​

Apache SeaTunnel(Incubating) API总体结构的设计如上图；

Source & Sink API：数据集成的核心API之一，明确Source的分片并行逻辑和Sink的提交逻辑，用于实现Connector；

Engine API：

Translation: 翻译层，用于将SeaTunnel的Souce与Sink API翻译成引擎内部可以运行的Connector；

Execution：执行逻辑，用于定义Source、Transform、Sink等操作在引擎内部的执行逻辑；

Table API：

Table SPI：主要用于以SPI的方式暴露Source与Sink接口，并明确Connector的必填与可选参数等；

DataType：SeaTunnel的数据结构，用于隔离引擎，声明Table Schema等；

Catalog：用于获取Table Scheme、Options等；

Catalog Storage: 用于存储用户定义Kafka等非结构化引擎的Table Scheme等；

从上图是我们现在设想的执行流程：

1. 从配置文件或UI等方式获取任务参数；
2. 通过参数从Catalog中解析得到Table Schema、Option等信息；
3. 以SPI方式拉起SeaTunnel的Connector，并注入Table信息等；
4. 将SeaTunnel的Connector翻译为引擎内部的Connector；
5. 执行引擎的作业逻辑，图中的多表分发目前只存在CDC整库同步场景下，其他Connector都是单表，不需要分发逻辑；

从以上可以看出，最难的部分是如何将Apache SeaTunnel(Incubating) 的Source和Sink翻译成引擎内部的Source和Sink。

当下许多用户不仅把Apache SeaTunnel (Incubating) 当做一个数据集成方向的工具，也当做数仓方向的工具，会使用很多Spark和Flink的SQL，我们目前希望能够保留这样的SQL能力，让用户实现无缝升级。

根据我们的调研，如上图，是对Source与Sink的理想执行逻辑，由于SeaTunnel以WaterDrop孵化，所以图上的术语偏向Spark；

理想情况下，在Driver上可以运行Source和Sink的协调器，然后Worker上运行Source的Reader和Sink的Writer。在Source协调器方面，我们希望它能支持几个能力。

一、是数据的分片逻辑，可以将分片动态添加到Reader中。

二、是可以支持Reader的协调。SourceReader用于读取数据，然后将数据发送到引擎中流转，最终流转到Source Writer中进行数据写入，同时Writer可以支持二阶段事务提交，并由Sink的协调器支持Iceberg等Connector的聚合提交需求；

04 Source API​

通过我们的调研，发现Source所需要的以下特性：

1. 统一离线和实时API：Source只实现一次，同时支持离线和实时；
2. 能够支持并行读取：比如Kafka每一个分区都生成一个的读取器，并行的执行；
3. 支持动态添加分片：比如Kafka定于一个topic正则，由于业务量的需求，需要新增一个topic，该Source API可以支持我们动态添加到作业中。
4. 支持协调读取器的工作：这个目前只发现在CDC这种Connector需要支持。CDC目前都是基于Netfilx的DBlog并行算法去支持，该情况在全量同步和增量同步两个阶段的切换时需要协调读取器的工作。
5. 支持单个读取器处理多张表：即由前面提到的支持实时场景下的数据库整库同步需求；

对应以上需求，我们做出了基础的API，如上图，目前代码以提交到Apache SeaTunnel(Incubating)的社区中api-draft分支，感兴趣的可以查看代码详细了解。

如何适配Spark和Flink引擎​

Flink与Spark都在后面统一了DataSet与DataStream API，即能够支持前两个特性，那么对于剩下的3个特性：

• 如何支持动态添加分片？
• 如何支持协调读取器？
• 如何支持单个读取器处理多张表？

带着问题，进入目前的设计。

我们发现除了CDC之外，其他Connector是不需要协调器的，针对不需要协调器的，我们会有一个支持并行的Source，并进行引擎翻译。

如上图中左边是一个分片的enumerator，可以列举source需要哪些分片，有哪些分片，实时进行分片的枚举，随后将每个分片分发到真正的数据读取模块SourceReader中。对于离线与实时作业的区分使用Boundedness标记，Connector可以在分片中标记是否有停止的Offset，如Kafka可以支持实时，同时也可以支持离线。ParallelSource可以在引擎设置任意并行度，以支持并行读取。

在需要协调器的场景，如上图，需要在Reader和Enumerator之间进行Event传输， Enumerator通过Reader发送的Event进行协调工作。Coordinated Source需要在引擎层面保证单并行度，以保证数据的一致性；当然这也不能良好的使用引擎的内存管理机制，但是取舍是必要的；

对于最后一个问题，我们如何支持单个读取器处理多张表。这会涉及到Table API层，通过Catalog读取到了所有需要的表后，有些表可能属于一个作业，可以通过一个链接去读取，有些可能需要分开，这个依赖于Source是怎么实现的。基于这是一个特殊需求，我们想要减少普通开发者的难度，在Table API这一层，我们会提供一个SupportMultipleTable接口，用于声明Source支持多表的读取。Source在实现时，要根据多张表实现对应的反序列化器。针对衍生的多表数据如何分离，Flink将采用Side Output机制，Spark预想使用Filter或Partition机制。

05 Sink API​

目前Sink所需的特性并不是很多，经过调研目前发现有三个需求：

1. 幂等写入，这个不需要写代码，主要看存储引擎是否能支持。
2. 分布式事务，主流是二阶段提交，如Kafka都是可以支持分布式事务的。
3. 聚合提交，对于Iceberg、hoodie等存储引擎而言，我们不希望有小文件问题，于是期望将这些文件聚合成一个文件，再进行提交。

基于以上三个需求，我们有对应的三个API，分别是SinkWriter、SinkCommitter、SinkAggregated Committer。SinkWriter是作为基础写入，可能是幂等写入，也可能不是。SinkCommitter支持二阶段提交。SinkAggregatedCommitter支持聚合提交。

理想状态下，AggregatedCommitter单并行的在Driver中运行，Writer与Committer运行在Worker中，可能有多个并行度，每个并行度都有自己的预提交工作，然后把自己提交的信息发送给Aggregated Committer再进行聚合。

目前Spark和Flink的高版本都支持在Driver(Job Manager)运行AggregatedCommitter，worker(Job Manager)运行writer和Committer。

但是对于Flink低版本，无法支持AggregatedCommitter在JM中运行，我们也进行翻译适配的设计。Writer与Committer会作为前置的算子，使用Flink的ProcessFunction进行包裹，支持并发的预提交与写入工作，基于Flink的Checkpoint机制实现二阶段提交，这也是目前Flink的很多Connector的2PC实现方式。这个ProcessFunction会将预提交信息发送到下游的Aggregated Committer中，Aggregated Committer可以采用SinkFunction或Process Function等算子包裹，当然，我们需要保证AggregatedCommitter只会启动一个，即单并行度，否则聚合提交的逻辑就会出现问题。

感谢各位的观看，如果大家对具体实现感兴趣，可以去 Apache SeaTunnel (Incubating) 的社区查看api-draft分支代码，谢谢大家。

作者 | Apache SeaTunnel(Incubating) Contributor 范佳

整理 | 测试工程师 冯秀兰

对于百亿级批数据的导入，传统的 JDBC 方式在一些海量数据同步场景下的表现并不尽如人意。为了提供更快的写入速度，Apache SeaTunnel(Incubating) 在刚刚发布的 2.1.1 版本中提供了 ClickhouseFile-Connector 的支持，以实现 Bulk load 数据写入。

Bulk load 指把海量数据同步到目标 DB 中，目前 SeaTunnel 已实现数据同步到 ClickHouse 中。

在 Apache SeaTunnel(Incubating) 4 月 Meetup 上，Apache SeaTunnel(Incubating) Contributor 范佳分享了《基于 SeaTunnel 的 ClickHouse bulk load 实现》，详细讲解了 ClickHouseFile 高效处理海量数据的具体实现原理和流程。

感谢本文整理志愿者 测试工程师 冯秀兰 对 Apache SeaTunnel(Incubating) 项目的支持！

本次演讲主要包含七个部分：

• ClickHouse Sink 现状
• ClickHouse Sink 弱场景
• ClickHouseFile 插件介绍
• ClickHouseFile 核心技术点
• ClickHouseFile 插件的实现解析
• 插件能力对比
• 后期优化方向

​范 佳白鲸开源 高级工程师

01 ClickHouse Sink 现状

现阶段，SeaTunnel 把数据同步到 ClickHouse 的流程是：只要是 SeaTunnel 支持的数据源，都可以把数据抽取出来，抽取出来之后，经过转换（也可以不转换），直接把源数据写入 ClickHouse sink connector 中，再通过 JDBC 写入到 ClickHouse 的服务器中。

但是，通过传统的 JDBC 写入到 ClickHouse 服务器中会存在一些问题。

首先，现阶段使用的工具是 ClickHouse 提供的驱动，实现方式是通过 HTTP，然而 HTTP 在某些场景下，实现效率不高。其次是海量数据，如果有重复数据或者一次性写入大量数据，使用传统的方式是生成对应的插入语句，通过 HTTP 发送到 ClickHouse 服务器端，在服务器端来进行逐条或分批次解析、执行，无法实现数据压缩。

最后就是我们通常会遇到的问题，数据量过大可能导致 SeaTunnel 端 OOM，或者服务器端因为写入数据量过大，频率过高，导致服务器端挂掉。

于是我们思考，是否有比 HTTP 更快的发送方式？如果可以在 SeaTunnel 端做数据预处理或数据压缩，那么网络带宽压力会降低，传输速率也会提高。

02 ClickHouse Sink 的弱场景

如果使用 HTTP 传输协议，当数据量过大，批处理以微批的形式发送请求，HTTP 可能处理不过来；

太多的 insert 请求，服务器压力大。假设带宽可以承受大量的请求，但服务器端不一定能承载。线上的服务器不仅需要数据插入，更重要的是查询数据为其他业务团队使用。若因为插入数据过多导致服务器集群宕机，是得不偿失的。

03 ClickHouse File 核心技术点

针对这些 ClickHouse 的弱场景，我们想，有没有一种方式，既能在 Spark 端就能完成数据压缩，还可以在数据写入时不增加 Server 的资源负载，并且能快速写入海量数据？于是我们开发了 ClickHouseFile 插件来满足这些需求。

ClickHouseFile 插件的关键技术是 ClickHouse -local。ClickHouse-local 模式可以让用户能够对本地文件执行快速处理，而无需部署和配置 ClickHouse 服务器。ClickHouse-local 使用与 ClickHouse Server 相同的核心，因此它支持大多数功能以及相同的格式和表引擎。

因为有这 2 个特点,这意味着用户可以直接处理本地文件，而无需在 ClickHouse 服务器端做处理。由于是相同的格式，我们在远端或者 SeaTunnel 端进行的操作所产生的数据和服务器端是无缝兼容的，可以使用 ClickHouse local 来进行数据写入。ClickHouse local 是实现 ClickHouseFile 的核心技术点，因为有了这个插件，现阶段才能够实现 ClickHouse file 连接器。

ClickHouse local 核心使用方式：

第一行：将数据通过 Linux 管道传递给 ClickHouse-local 程序的 test_table 表。

第二至五行：创建一个 result_table 表用于接收数据。

第六行：将数据从 test_table 到 result_table 表。

第七行：定义数据处理的磁盘路径。

通过调用 Clickhouse-local 组件，实现在 Apache SeaTunnel(Incubating) 端完成数据文件的生成，以及数据压缩等一系列操作。再通过和 Server 进行通信，将生成的数据直接发送到 Clickhouse 的不同节点，再将数据文件提供给节点查询。

原阶段和现阶段实现方式对比：

原来是 Spark 把数据包括 insert 语句，发送给服务器端，服务器端做 SQL 的解析，表的数据文件生成、压缩，生成对应的文件、建立对应索引。若使用 ClickHouse local 技术，则由 SeaTunnel 端做数据文件的生成、文件压缩，以及索引的创建，最终产出就是给服务器端使用的文件或文件夹，同步给服务器后，服务器就只需对数据查询，不需要做额外的操作。

04 核心技术点

以上流程可以促使数据同步更加高效，得益于我们对其中的三点优化。

第一，数据实际上师从管道传输到 ClickHouseFile，在长度和内存上会有限制。为此，我们将 ClickHouse connector，也就是 sink 端收到的数据通过 MMAP 技术写入临时文件，再由 ClickHouse local 读取临时文件的数据，生成我们的目标 local file，以达到增量读取数据的效果，解决 OM 的问题。

第二，支持分片。因为如果在集群中使用，如果只生成一个文件或文件夹，实际上文件只分发到一个节点上，会大大降低查询的性能。因此，我们进行了分片支持，用户可以在配置文件夹中设置分片的 key，算法会将数据分为多个 log file，写入到不同的集群节点中，大幅提升读取性能。

第三个重要的优化是文件传输，目前 SeaTunnel 支持两种文件传输方式，一种是 SCP，其特点是安全、通用、无需额外配置；另一种是 RSYNC，其有点事快速高效，支持断点续传，但需要额外配置，用户可以根据需要选择适合自己的方式。

05 插件实现解析

概括而言，ClickHouseFile 的总体实现流程如下：

• 缓存数据，缓存到 ClickHouse sink 端；
• 调用本地的 ClickHouse-local 生成文件；
• 将数据发送到 ClickHouse 服务端；
• 执行 ATTACH 命令

通过以上四个步骤，生成的数据达到可查询的状态。

06 插件能力对比

从数据传输角度来说，ClickHouseFile 更适用于海量数据，优势在于不需要额外的配置，通用性强，而 ClickHouseFile 配置比较复杂，目前支持的 engine 较少；

就环境复杂度来说，ClickHouse 更适合环境复杂度高的情况，不需要额外配置就能直接运行；

在通用性上，ClickHouse 由于是 SeaTunnel 官方支持的 JDBC diver，基本上支持所有的 engine 的数据写入，ClickHouseFile 支持的 engine 相对较少；从服务器压力方面来说，ClickHouseFile 的优势在海量数据传输时就体现出来了，不会对服务器造成太大的压力。

但这二者并不是竞争关系，需要根据使用场景来选择。

07 后续计划

目前虽然 SeaTunnel 支持 ClickHouseFile 插件，但是还有很多地方需要优化，主要包括：

• Rsync 支持；
• Exactly-Once 支持；
• 支持 Zero Copy 传输数据文件；
• 更多 Engine 的支持

在Apache SeaTunnel(Incubating) 4 月Meetup上，孩子王大数据专家、OLAP平台架构师 袁洪军 为我们带来了《Apache SeaTunnel (Incubating)在孩子王的应用实践》。

本次演讲主要包含五个部分：

• 孩子王引入Apache SeaTunnel (Incubating)的背景介绍

• 大数据处理主流工具对比分析

• Apache SeaTunnel (Incubating)的落地实践

• Apache SeaTunnel (Incubating)改造中的常见问题

• 对孩子王未来发展方向的预测展望

袁洪军

孩子王 大数据专家、OLAP 平台架构师。多年大数据平台研发管理经验，在数据资产、血缘图谱、数据治理、OLAP 等领域有着丰富的研究经验。

01 背景介绍​

目前孩子王的OLAP平台主要包含元数据层、任务层、存储层、SQL层、调度层、服务层以及监控层七部分，本次分享主要关注任务层中的离线任务。

其实孩子王内部有一套完整的采集推送系统，但由于一些历史遗留问题，公司现有的平台无法快速支持OLAP平台上线，因此当时公司只能选择放弃自身的平台，转而着手研发新的系统。

当时摆在OLAP面前的有三个选择：

1、给予采集推送系统做二次研发；

2、完全自研；

3、参与开源项目。

02 大数据处理主流工具对比分析​

而这三项选择却各有优劣。若采基于采集推送做二次研发，其优点是有前人的经验，能够避免重复踩坑。但缺点是代码量大，研读时间、研读周期较长，而且抽象代码较少，与业务绑定的定制化功能较多，这也导致了其二开的难度较大。

若完全自研，其优点第一是开发过程自主可控，第二是可以通过Spark等一些引擎做贴合我们自身的架构，但缺点是可能会遭遇一些未知的问题。

最后如果使用开源框架，其优点一是抽象代码较多，二是经过其他大厂或公司的验证，框架在性能和稳定方面能够得到保障。因此孩子王在OLAP数据同步初期，我们主要研究了DATAX、Sqoop和SeaTunnel这三个开源数据同步工具。

从脑图我们可以看到，Sqoop的主要功能是针对RDB的数据同步，其实现方式是基于MAP/REDUCE。Sqoop拥有丰富的参数和命令行可以去执行各种操作。Sqoop的优点在于它首先贴合Hadoop生态，并已经支持大部分RDB到HIVE任意源的转换，拥有完整的命令集和API的分布式数据同步工具。

但其缺点是Sqoop只支持RDB的数据同步，并且对于数据文件有一定的限制，以及还没有数据清洗的概念。

DataX的主要功能是任意源的数据同步，通过配置化文件+多线程的方式实现，主要分为三个流程：Reader、Framework和Writer，其中Framework主要起到通信和留空的作用。

DataX的优点是它采用了插件式的开发，拥有自己的流控和数据管控，在社区活跃度上，DataX的官网上提供了许多不同源的数据推送。但DataX的缺点在于它基于内存，对数据量可能存在限制。

Apache SeaTunnel (Incubating)做的也是任意源的数据同步，实现流程分为source、transform和sink三步，基于配置文件、Spark或Flink实现。其优点是目前官网2.1.0有非常多的插件和源的推送，基于插件式的思想也使其非常容易扩展，拥抱Spark和Flink的同时也做到了分布式的架构。要说Apache SeaTunnel (Incubating)唯一的缺点可能是目前缺少IP的调用，UI界面需要自己做管控。

综上所述，Sqoop虽然是分布式，但是仅支持RDB和HIVE、Hbase之间的数据同步且扩展能力差，不利于二开。DataX扩展性好，整体性稳定，但由于是单机版，无法分布式集群部署，且数据抽取能力和机器性能有强依赖关系。而SeaTunnel和DataX类似并弥补了DataX非分布式的问题，对于实时流也做了很好的支持，虽然是新产品，但社区活跃度高。基于是否支持分布式、是否需要单独机器部署等诸多因素的考量，最后我们选择了SeaTunnel。

03 Apache SeaTunnel (Incubating)的落地实践​

在Apache SeaTunnel (Incubating)的官网我们可以看到Apache SeaTunnel (Incubating)的基础流程包括source、transform和sink三部分。根据官网的指南，Apache SeaTunnel (Incubating)的启动需要配置脚本，但经过我们的研究发现，Apache SeaTunnel (Incubating)的最终执行是依赖config文件的spark-submit提交的一个Application应用。

这种初始化方式虽然简单，但存在必须依赖Config文件的问题，每次运行任务后都会生成再进行清除，虽然可以在调度脚本中动态生成，但也产生了两个问题。1、频繁的磁盘操作是否有意义；2、是否存在更为高效的方式支持Apache SeaTunnel (Incubating)的运行。

基于以上考量，在最终的设计方案中，我们增加了一个统一配置模板平台模块。调度时只需要发起一个提交命令，由Apache SeaTunnel (Incubating)自身去统一配置模板平台中拉取配置信息，再去装载和初始化参数。

上图展示的便是孩子王OLAP的业务流程，主要分为三块。数据从Parquet，即Hive，通过Parquet表的方式到KYLIN和CK source的整体流程。

这是我们建模的页面，主要通过拖拉拽的方式生成最终模型，每个表之间通过一些交易操作，右侧是针对Apache SeaTunnel (Incubating)的微处理。

因此我们最终提交的命令如上，其中标红的首先是【-conf customconfig/jars】，指用户可以再统一配置模板平台进行处理，或者建模时单独指定。最后标红的【421 $start_time $end_time $taskType】Unicode，属于唯一编码。

下方图左就是我们最终调度脚本提交的38个命令，下方图右是针对Apache SeaTunnel (Incubating)做的改造，可以看到一个较为特殊的名为WaterdropContext的工具类。可以首先判断Unicode是否存在，再通过Unicode_code来获取不同模板的配置信息，避免了config文件的操作。

在最后的reportMeta则是用于在任务执行完成后上报一些信息，这也会在Apache SeaTunnel (Incubating)中完成。

在最终完成的config文件如上，值得注意的是在transform方面，孩子王做了一些改造。首先是针对手机或者身份证号等做脱敏处理，如果用户指定字段，就按照字段做，如果不指定字段就扫描所有字段，然后根据模式匹配，进行脱敏加密。

第二transform还支持自定义处理，如上文说道OLAP建模的时候说到。加入了HideStr，可以保留一串字符的前十个字段，加密后方的所有字符，在数据安全上有所保障。

然后，在sink端，我们为了支持任务的幂等性，我们加入了pre_sql，这主要完成的任务是数据的删除，或分区的删除，因为任务在生产过程中不可能只运行一次，一旦出现重跑或补数等操作，就需要这一部分为数据的不同和正确性做考量。

在图右方的一个Clickhouse的Sink端，这里我们加入了一个is_senseless_mode，它组成了一个读写分离的无感模式，用户在查询和补数的时候不感知整体区域，而是用到CK的分区转换，即名为MOVE PARTITION TO TABLE的命令进行操作的。

此处特别说明KYLIN的Sink端，KYLIN是一个非常特殊的源，拥有自己一整套数据录入的逻辑，而且，他有自己的监控页面，因此我们给予KYLIN的改造只是简单地调用其API操作，在使用KYLIN时也只是简单的API调用和不断轮询的状态，所以KYLIN这块的资源在统一模板配置平台就被限制地很小。

04 Apache SeaTunnel (Incubating)改造中的常见问题​

1、OOM&Too many Parts

问题通常会出现在Hive到Hive的过程中，即使我们通过了自动资源的分配，但也存在数据突然间变大的情况，比如在举办了多次活动之后。这样的问题其实只能通过手动动态地调参，调整数据同步批量时间来避免。未来我们可能尽力去完成对于数据量的掌握，做到精细的控制。

2、字段、类型不一致问题

模型上线后，任务依赖的上游表或者字段，用户都会做一些修改，这些修改若无法感知，可能导致任务的失败。目前解决方法是依托血缘+快照的方式进行提前感知来避免错误。

3、自定义数据源&自定义分隔符

如财务部门需要单独使用的分割符，或是jar信息，现在用户可以自己在统一配置模板平台指定加载额外jar信息以及分割符信息。

4、数据倾斜问题

这可能因为用户自己设置了并行度，但无法做到尽善尽美。这一块我们暂时还没有完成处理，后续的思路可能在Source模块中添加post处理，对数据进行打散，完成倾斜。

5、KYLIN全局字典锁问题

随着业务发展，一个cube无法满足用户使用，就能需要建立多个cube，如果多个cube之间用了相同的字段，就会遇到KYLIN全局字典锁的问题。目前解决的思路是把两个或多个任务之间的调度时间进行隔开，如果无法隔开，可以做一个分布式锁的控制。KYLIN的sink端必须要拿到锁才能运行。

05 对孩子王未来发展方向的预测展望

• 多源数据同步，未来可能针对RDB源进行处理

• 基于实时Flink的实现

• 接管已有采集调度平台（主要解决分库分表的问题）

• 数据质量校验，像一些空值、整个数据的空置率、主时间的判断等

我的分享就到这里，希望以后可以和社区多多交流，共同进步，谢谢！

可管理，可调用，可计算，可变现的数据资源才能成为资产，信息系统的互联互通使得多源和多维度的数据集成需求巨大，这就对数据处理和集成的工具提出了严苛的要求。

智能化时代，在“智慧城市”、“智慧治理”、“产品智能化”等的趋势下，企业大多面临如何实现高效数据推送，提高平台质量，以及保障数据安全的挑战。选对数据集成工具和平台，数据才能发挥出做大的作用。

Apache SeaTunnel (Incubating) 作为下一代高性能、分布式、海量数据集成框架，致力于让数据同步更简单，更高效，加快分布式数据处理能力在生产环境落地。

在 Apache SeaTunnel(Incubating) Meetup（2022 年 4 月 16日），Apache SeaTunnel(Incubating) 社区将邀请了 Apache SeaTunnel(Incubating)的资深用户，分享 Apache SeaTunnel(Incubating)在智能化生产环境中落地的最佳实践。此外，还会有贡献者现场进行 Apache SeaTunnel(Incubating)的源码解析，让你对 Apache SeaTunnel(Incubating)有一个更加全面而深入的了解。

无论你是对 Apache SeaTunnel(Incubating)抱有兴趣的初学者，还是在日常的生产实践中遭遇了复杂棘手的部署问题，都可以来到这里，与我们的讲师近距离沟通，得到你想要的答案。

01 报 名 通 道​

Apache SeaTunnel (Incubating) Meetup | 4 月线上直播报名通道已开启，赶快预约吧！

时间：2022-4-16 14:00-17:00

形式：线上直播

点击链接或扫码预约报名（免费）：

扫码预约报名

扫码进直播群

02 活 动 亮 点​

• 行业案例详解
• 特色功能分析
• 一线企业踩坑心得
• 开源社区实战攻略
• 行业技术专家面对面 Q&A
• 惊喜礼品送不停

03 活 动 议 程​

活动当天，将有来自孩子王、oppo 的工程师现场分享来自厂商的一线前沿实践经验，还有来自白鲸开源的高级工程师对 Apache SeaTunnel(Incubating)的重要功能更新进行“硬核”讲解，干货满满。

袁洪军 孩子王 大数据专家、OLAP 平台架构师

多年大数据平台研发管理经验，在数据资产、血缘图谱、数据治理、OLAP 等领域有着丰富的研究经验

演讲时间：14:00-14:40

演讲题目：Apache SeaTunnel(Incubating) 在孩子王的应用实践

演讲概要： 如何实现高效数据推送？如何提高平台质量？如何保障数据安全？孩子王又对 Apache SeaTunnel(Incubating)做了哪些改造？

范佳 白鲸开源  高级工程师 Apache SeaTunnel Contributor

演讲时间： 14:40-15:20

演讲题目： 基于 Apache SeaTunnel(Incubating)的 Clickhouse Bulk Load 实现

演讲概要： 通过扩展 Apache SeaTunnel(Incubating)的 Connector实现 Clickhouse的 bulk load 数据同步功能。

王子超 oppo 高级后端工程师

演讲时间： 15:50-16:30

演讲题目： oppo智能推荐样本中心基于 Apache SeaTunnel(Incubating)的技术革新

演讲概要： 介绍 oppo 智能推荐机器学习样本流程的演进及 Apache  SeaTunnel(Incubating) 在其中发挥的作用。

除了精彩的演讲之外，现场还设置了多个抽奖环节，参与抽奖有机会获得 Apache SeaTunnel(Incubating) 精美定制礼品，敬请期待~

2021 年 12 月 9 日，Apache SeaTunnel(Incubating) 进入 Apache 孵化器，在经过社区各位贡献者近四个月的努力下，我们于2022年3月18日发布了首个Apache版本，并且保证了首个版本一次性通过检查。这意味着 2.1.0 版本，是经过 Apache SeaTunnel(Incubating) 社区和 Apache 孵化器投票检查发布的官方版本，企业和个人用户可以放心安全使用。

Note： 软件许可协议是一种具有法律性质的合同或指导，目的在于规范受著作权保护的软件的使用或散布行为。通常的许可方式会允许用户来使用单一或多份该软件的复制，因为若无许可而径予使用该软件，将违反著作权法给予该软件开发者的专属保护。效用上来说，软件许可是软件开发者与其用户之间的一份合约，用来保证在符合许可范围的情况下，用户将不会受到控告。进入孵化器前后，我们花费了大量的时间来梳理整个项目的外部依赖以确保整个项目的合规性。需要说明的是，开源软件选择怎样的License并不意外着项目本身就一定合规。而ASF严苛的版本检查最大程度地保证了软件License的合规性，以及软件合理合法的流通分发。

本次发布版本说明​

本次发布我们主要带来了以下特性：​

• 对微内核插件化的架构内核部分进行了大量优化，内核以 Java 为主，并对命令行参数解析，插件加载等做了大量改进，同时插件扩展可根据用户（或贡献者）所擅长的语言去做开发，极大程度地降低了插件开发门槛。
• 全面支持 Flink ，但同时用户也可自由选择底层引擎，本次更新也为大家带来了大量的Flink插件，也欢迎大家后续贡献相关插件。
• 提供本地开发极速启动环境支持（example），贡献者或用户可以在不更改任何代码的情况下快速丝滑启动，方便本地快速开发调试体验。对于需要自定义插件的贡献者或者用户来讲，这无疑是个令人激动的好消息。事实上，我们在发布前的测试中，也有大量贡献者采用这种方式快速对插件进行测试。
• 提供了Docker容器安装，用户可以极快地通过Docker部署安装使用Apache SeaTunnel，未来我们也会围绕Docker&K8s做出大量迭代，欢迎大家讨论交流。

具体发布说明： [Feature]

• 使用 JCommander来做命令行参数解析，使得开发者更关注逻辑本身。
• Flink从1.9升级至1.13.5，保持兼容旧版本，同时为后续CDC做好铺垫。
• 支持 Doris 、Hudi、Phoenix、Druid等Connector 插件，完整的插件支持你可以在这里找到 plugins-supported-by-seatunnel.
• 本地开发极速启动环境支持，你可以在使用example模块，不修改任何代码的前提下快速启动，方便开发者本地调试体验。
• 支持通过 Docker 容器安装和试用 Apache SeaTunnel。
• Sql 组件支持 SET语句，支持配置变量。
• Config模块重构，减少贡献者理解成本，同时保证项目的代码合规（License）。
• 项目结构重新调整，以适应新的Roadmap。
• CI&CD的支持，代码质量自动化管控，（后续会有更多的计划来支持CI&CD开发）。

【致谢】​

感谢以下参与贡献的同学(为GitHub ID，排名不分先后)：

Al-assad, BenJFan, CalvinKirs, JNSimba, JiangTChen, Rianico, TyrantLucifer, Yves-yuan, ZhangchengHu0923, agendazhang, an-shi-chi-fan, asdf2014, bigdataf, chaozwn, choucmei, dailidong, dongzl, felix-thinkingdata, fengyuceNv, garyelephant, kalencaya, kezhenxu94, legendtkl, leo65535, liujinhui1994, mans2singh, marklightning, mosence, nielifeng, ououtt, ruanwenjun, simon824, totalo, wntp, wolfboys, wuchunfu, xbkaishui, xtr1993, yx91490, zhangbutao, zhaomin1423, zhongjiajie, zhuangchong, zixi0825.

同时也诚挚的感谢我们的Mentor：Zhenxu Ke，Willem Jiang， William Guo，LiDong Dai ，Ted Liu, Kevin，JB 在这个过程中给予的帮助

未来几个版本的规划：​

• CDC的支持；
• 监控体系的支持；
• UI系统的支持；
• 更多的 Connector 支持，以及更高效的Sink支持，如ClickHouse，很快会在下个版本跟大家见面。

后续Feature是由社区共同决定的，我们也在这里呼吁大家一同参与社区后续建设。 欢迎大家关注以及贡献：）

社区发展​

【近期概况】​

自进入Apache孵化器以来，贡献者从13 人增长至 55 人，且持续保持上升趋势，平均周commits维持在20+，来自不同公司的三位贡献者（Lei Xie, HuaJie Wang,Chunfu Wu,）通过他们对社区的贡献被邀请成为Committer。我们举办了两场MeetUp，来自B站，OPPO、唯品会等企业讲师分享了SeaTunnel在他们在企业中的大规模生产落地实践(后续我们也会保持每月一次的meetup，欢迎各位使用SeaTunnel的用户或者贡献者分享SeaTunnel和你们的故事)。

【Apache SeaTunnel(Incubating)的用户】​

Note:仅包含已登记用户 Apache SeaTunnel(Incubating) 目前登记用户如下，如果您也在使用Apache SeaTunnel，欢迎在Who is using SeaTunne! 中登记！

【PPMC感言】​

Apache SeaTunnel(Incubating) PPMC LiFeng Nie在谈及首个Apache版本发布的时候说，从进入Apache Incubator的第一天，我们就一直在努力学习Apache Way以及各种Apache政策，第一个版本发布的过程花费了大量的时间（主要是合规性），但我们认为这种时间是值得花费的，这也是我们选择进入Apache的一个很重要的原因，我们需要让用户用得放心，而Apache无疑是最佳选择，其 License 近乎苛刻的检查会让用户尽可能地避免相关的合规性问题，保证软件合理合法的流通。另外，其践行Apache Way，例如公益使命、实用主义、社区胜于代码、公开透明与共识决策、任人唯贤等，可以帮助 SeaTunnel 社区更加开放、透明，向多元化方向发展。

分享嘉宾：唯品会 资深大数据工程师 王玉 讲稿整理：张德通

导读: 唯品会早在1.0版本时就引用了SeaTunnel，我们使用SeaTunnel进行一些Hive到ClickHouse之间数据交互的工作。 今天的介绍会围绕下面几点展开：

• ClickHouse数据导入的需求和痛点；
• ClickHouse出仓入仓工具选型；
• Hive to ClickHouse；
• ClickHouse to Hive；
• SeaTunnel与唯品会数据平台的集成；
• 未来展望；

ClickHouse数据导入的需求和痛点

1.唯品会数据OLAP架构​

图中是唯品会OLAP架构，我们负责的模块是图中的数据服务和计算引擎两大部分。底层依赖的数据仓库分为离线数仓、实时数仓和湖仓。计算引擎方面，我们使用Presto、Kylin和Clickhouse。虽然Clickhouse是一个存储一体的OLAP数据库，我们为了利用Clickhouse的优秀计算性能而将它归入了计算引擎部分。基于OLAP组件之上，我们提供了SQL类数据服务和非SQL的唯品会自主分析，为不同智能服务。例如非SQL服务是为BI和商务提供更贴近业务的数据分析的服务。在数据服务至上抽象了多个数据应用。

2.需求​

我们通过Presto Connector和Spark组件，把底层的Hive、Kudu、Alluxio组件打通。大数据组件之间可以互相导入导出数据，可以根据数据分析的需求和场景任意利用合适的组件分析数据。但我们引入Clickhouse时，它是一个数据孤岛，数据的导入和导出比较困难。Hive和Clickhouse之间需要做很多工作才能实现导入导出。我们的第一个数据导入导出需求就是提升导入导出效率，把Clickhouse纳入大数据体系中。

第二个需求是Presto跑SQL比较慢，图中是一个慢SQL的例子。图中的SQL where条件设置了日期、时间范围和具体过滤条件，这类SQL使用由于Presto使用分区粒度下推，运行比较慢。即使用Hive的Bucket表和分桶等其他方式优化后也是几秒的返回时间、不能满足业务要求。这种情况下，我们需要利用Clickhouse做离线的OLAP计算加速。

我们的实时数据是通过Kafka、Flink SQL方式写入到Clickhouse中。但分析时只用实时数据是不够的，需要用Hive维度表和已经ETL计算号的T+1实时表一起在Clickhouse中做加速运输。这需要把Hive的数据导入到Clickhouse中，这就是我们的第三个需求。

3.痛点​

首先，我们引入一项数据组件时要考虑其性能。Hive表粒度是五分钟，是否有组件可以支撑五分钟内完成一个短小ETL流程并把ETL结果导入到Clickhouse中？第二，我们需要保证数据质量，数据的准确性需要有保障。Hive和Clickhouse的数据条数需要保障一致性，如果数据质量出问题能否通过重跑等机制修复数据？第三，数据导入需要支持的数据类型是否完备？不同数据库之间的数据类型和一些机制不同，我们有HiperLogLog和BitMap这类在某一存储引擎中利用得比较多得数据类型，是否可以正确传输和识别，且可以较好地使用。

ClickHouse和Hive出仓入仓工具的选型

基于数据业务上的痛点，我们对数据出仓入仓工具进行了对比和选择。我们主要在开源工具中进行选择，没有考虑商业出入仓工具，主要对比DataX、SeaTunnel和编写Spark程序并用jdbc插入ClickHouse这三个方案中取舍。 SeaTunnel和Spark依赖唯品会自己的Yarn集群，可以直接实现分布式读取和写入。DataX是非分布式的，且Reader、Writer之间的启动过程耗时时间长，性能普通，SeaTunnel和Spark处理数据的性能可以达到DataX的数倍。 十亿以上的数据可以平稳地在SeaTunnel和Spark中运行，DataX在数据量大以后性能压力大，处理十亿以上数据吃力。 在读写插件扩展性方面，SeaTunnel支持了多种数据源，支持用户开发插件。SeaTunnel支持了数据导入Redis。 稳定性上，SeaTunnel和DataX由于是自成体系的工具，稳定性会更好。Spark的稳定性方面需要关注代码质量。

我们的曝光表数据量每天在几十亿级，我们有5min内完成数据处理的性能要求，我们我们存在数据导入导出到Redis的需求，我们需要导入导出工具可以接入到数据平台上进行任务调度。 出于数据量级、性能、可扩展性、平台兼容性几方面的考虑，我们选择了SeaTunnel作为我们的数仓导入导出工具。

Hive数据导入到ClickHouse

下面将介绍我们对SeaTunnel的使用。 图中是一张Hive表，它是我们三级的商品维度表，包含品类商品、维度品类和用户人群信息。表的主键是一个三级品类ct_third_id，下面的value是两个uid的位图，是用户id的bitmap类型，我们要把这个Hive表导入到Clickhouse。

SeaTunnel安装简单，官网文档有介绍如何安装。下图中是SeaTunnel的配置，配置中env、source和sink是必不可少的。env部分，图中的例子是Spark配置，配置了包括并发度等，可以调整这些参数。source部分是数据来源，这里配置了Hive数据源，包括一条Hive Select语句，Spark运行source配置中的SQL把数据读出，此处支持UDF进行简单ETL；sink部分配置了Clickhouse，可以看到output_type=rowbinary，rowbinary是唯品会自研加速方案；pre_sql和check_sql是自研的用于数据校验的功能，后面也会详细介绍；clickhouse.socket_timeout和bulk_size都是可以根据实际情况进行调整的。

运行SeaTunnel，执行sh脚本文件、配置conf文件地址和yarn信息，后即可。 运行过程中会产生Spark日志，运行成功和运行中错误都可以在日志中查看。

为了更贴合业务，唯品会对SeaTunnel做了一些改进。我们的ETL任务都是需要重跑的，我们支持了pre_sql和check_sql实现数据的重跑和对数。主要流程是在数据准备好后，执行pre_sql进行预处理，在Clickhouse中执行删除旧分区数据、存放到某一目录下在失败时恢复该分区、rename这类操作。check_sql会检验，校验通过后整个流程结束；如果检验不通过，根据配置进行重跑，重跑不通过则报警到对应负责人。

唯品会基于1.0版本SeaTunnel增加了RowBinary做加速，也让HuperLogLog和BinaryBitmap的二进制文件能更容易地从Hive导入到Clickhouse。我们在ClickHouse-jdbc、bulk_size、Hive-source几处进行了修改。使用CK-jdbc的extended api，以rowbinary方式将数据写入CK，bulk_size引入了以rowbinary方式写入CK的控制逻辑，Hive-source RDD以HashPartitioner进行分区将数据打散，防止数据倾斜。

我们还让SeaTunnel支持了多类型，为了圈人群的功能，需要在Clickhouse、Preso、Spark中实现对应的方法。我们在Clickhouse-jdbc中增加支持Batch特性的Callback、HttpEntity、RowBinaryStream，在Clickhouse-jdbc和Clickhouse-sink代码中增加了bitmap类型映射，在Presto和Spark中实现了Clickhouse的Hyperloglog和Bitmap的function的UDF。 前面的配置中，Clickhouse-sink部分可以指定表名，这里有写入本地表和分布式表的差异。写入分布式表的性能比写入本地表差对Clickhouse集群的压力会更大，但在计算曝光表、流量表，ABTest等场景中需要两表Join，两张表量级均在几十亿。这时我们希望Join key落在本机，Join成本更小。我们建表时在Clickhouse的分布式表分布规则中配置murmurHash64规则，然后在Seatunnel的sink里直接配置分布式表，把写入规则交给Clickhouse，利用了分布式表的特性进行写入。写入本地表对Clickhouse的压力会更小，写入的性能也会更好。我们在Seatunnel里，根据sink的本地表，去Clickhouse的System.cluster表里获取表的分布信息和机器分布host。然后根据均分规则写入这些host。把数据分布式写入的事情放到Seatunnel里来做。 针对本地表和分布式表的写入，我们未来的改造方向是在Seatunnel实现一致性哈希，直接按照一定规则写如Clickhouse、不依赖Clickhouse自身做数据分发，改善Clickhouse高CPU负载问题。

ClickHouse数据导入到Hive

我们有圈人群的需求，每天唯品会为供应商圈20万个人群，比如80后、高富帅、白富美的人群集合。这些在Clickhouse中的Bitmap人群信息需要导出到Hive表，在Hive中与其他ETL任务进行配合，最后推到PIKA交给外部媒体使用。我们使SeaTunnel将Clickhouse Bitmap人群数据反推到Hive。

图中是SeaTunnel配置，我们把source配置为Clickhouse、sink配置为Hive，数据校验也配置在Hive内。

由于我们接入SeaTunnel较早，我们对一些模块间进行了加工，包括新增plugin-spark-sink-hive模块、plugin-spark-source-ClickHouse模块，重写Spark Row相关方法，使其能封装经过Schem映射后的Clickhouse数据，重新构造StructField并生成最终需要落地Hive的DataFrame。最新版本已经有了很多source和sink组件，在SeaTunnel使用上更方便。现在也可以在SeaTunnel中直接集成Flink connector。

SeaTunnel与唯品会数据平台的集成

各个公司都有自己的调度系统，例如白鲸、宙斯。唯品会的调度工具是数坊，调度工具中集成了数据传输工具。下面是调度系统架构图，其中包含各类数据的出入仓。

SeaTunnel任务类型集成到平台中，图中是数坊的定时任务截图，可以看到选中的部分，是一个配置好的SeaTunnel任务，负责人、最近一次耗时，前后依赖任务的血缘信息，消耗的资源信息。下面展示了历史运行实例信息。

我们把SeaTunnel集成到了调度系统中，数坊调度Master会根据任务类型把任务分配到对应的Agent上，根据Agent负载情况分配到合适的机器上运行，管控器把前台的任务调度配置和信息拉取到后生成SeaTunnel cluster，在类似于k8s pod、cgroup隔离的虚拟环境内进行执行。运行结果会由调度平台的数据质量监控判断任务是否完成、是否运行成功，失败时进行重跑和告警。

SeaTunnel本身是一个工具化的组件，是为了进行数据血缘，数据质量，历史记录，高警监控，还包括资源分配这些信息的管控。我们把SeaTunnel集成到平台中，可以利用平台优势利用好SeaTunnel。 圈存人群中利用了SeaTunnel进行处理。我们通过打点数据，把圈存人群按照路径和使用情况分为不同的人，或称千人千面，把用户打上标签，圈出的某一类人群推送给用户、分析师和供应商。

流量进入Kafka，通过Flink入仓，再通过ETL形成用户标签表，用户标签表生成后，我们通过Presto实现了的BitMap方法，把数据打成Hive中的宽表。用户通过在人群系统页面中框选词条创建任务，提交腾群，生成SQL查询Clickhouse BitMap。Clickhouse的BitMap查询速度非常快，由天生优势，我们需要把Hive的BitMap表通过SeaTunnel导入到Clickhouse中。圈完人群后我们需要把表落地，形成Clickhouse的一个分区或一条记录，再把生成的结果BitMap表通过SeaTunnel存储到Hive中去。最后同步工具会将Hive的BitMap人群结果同步给外部媒体仓库Pika。每天圈20w个人群左右。 整个过程中SeaTunnel负责把数据从Hive导出到Clickhouse，Clickhouse的ETL流程完成后SeaTunnel把数据从Clickhouse导出到Hive。 为了完成这样的需求，我们在Presto和Spark端现ClickHouse的Hyperloglog和BitMap的function的UDF；我们还开发Seatunnel接口，使得用户在ClickHouse里使用Bitmap方法圈出来的人群，可以直接通过Seatunnel写入Hive表，无需中间落地步骤。用户也可以在Hive里通过spark圈人群或者反解人群bitmap，调用SeaTunnel接口，使数据直接传输到ClickHouse的结果表，而无需中间落地。

后续工作

后续我们会进一步改善Clickhouse写入数据时CPU负载高的问题，下一步会在SeaTunnel中实现Clickhouse数据源和读取端的CK-local模式，读写分离，减轻Clickhouse压力。未来我们也会增加更多sink支持，如数据推送到Pika和相应的数据检查。

ClickHouse 是面向 OLAP 的分布式列式 DBMS。我们部门目前已经把所有数据分析相关的日志数据存储至 ClickHouse 这个优秀的数据仓库之中，当前日数据量达到了 300 亿。

之前介绍的有关数据处理入库的经验都是基于实时数据流，数据存储在 Kafka 中，我们使用 Java 或者 Golang 将数据从 Kafka 中读取、解析、清洗之后写入 ClickHouse 中，这样可以实现数据的快速接入。然而在很多同学的使用场景中，数据都不是实时的，可能需要将 HDFS 或者是 Hive 中的数据导入 ClickHouse。有的同学通过编写 Spark 程序来实现数据的导入，那么是否有更简单、高效的方法呢。

目前开源社区上有一款工具 Seatunnel，项目地址 https://github.com/apache/incubator-seatunnel，可以快速地将 HDFS 中的数据导入 ClickHouse。

HDFS To ClickHouse​

假设我们的日志存储在 HDFS 中，我们需要将日志进行解析并筛选出我们关心的字段，将对应的字段写入 ClickHouse 的表中。

Log Sample​

我们在 HDFS 中存储的日志格式如下， 是很常见的 Nginx 日志

10.41.1.28 github.com 114.250.140.241 0.001s "127.0.0.1:80" [26/Oct/2018:03:09:32 +0800] "GET /Apache/Seatunnel HTTP/1.1" 200 0 "-" - "Dalvik/2.1.0 (Linux; U; Android 7.1.1; OPPO R11 Build/NMF26X)" "196" "-" "mainpage" "443" "-" "172.16.181.129"

ClickHouse Schema​

我们的 ClickHouse 建表语句如下，我们的表按日进行分区

CREATE TABLE cms.cms_msg
(
    date Date, 
    datetime DateTime, 
    url String, 
    request_time Float32, 
    status String, 
    hostname String, 
    domain String, 
    remote_addr String, 
    data_size Int32, 
    pool String
) ENGINE = MergeTree PARTITION BY date ORDER BY date SETTINGS index_granularity = 16384

Seatunnel with ClickHouse​

接下来会给大家详细介绍，我们如何通过 Seatunnel 满足上述需求，将 HDFS 中的数据写入 ClickHouse 中。

Seatunnel​

Seatunnel 是一个非常易用，高性能，能够应对海量数据的实时数据处理产品，它构建在Spark之上。Seatunnel 拥有着非常丰富的插件，支持从 Kafka、HDFS、Kudu 中读取数据，进行各种各样的数据处理，并将结果写入 ClickHouse、Elasticsearch 或者 Kafka 中。

Prerequisites​

首先我们需要安装 Seatunnel，安装十分简单，无需配置系统环境变量

1. 准备 Spark 环境
2. 安装 Seatunnel
3. 配置 Seatunnel

以下是简易步骤，具体安装可以参照 Quick Start

cd /usr/local

wget https://archive.apache.org/dist/spark/spark-2.2.0/spark-2.2.0-bin-hadoop2.7.tgz
tar -xvf https://archive.apache.org/dist/spark/spark-2.2.0/spark-2.2.0-bin-hadoop2.7.tgz

wget https://github.com/InterestingLab/seatunnel/releases/download/v1.1.1/seatunnel-1.1.1.zip

unzip seatunnel-1.1.1.zip

cd seatunnel-1.1.1
vim config/seatunnel-env.sh

# 指定Spark安装路径
SPARK_HOME=${SPARK_HOME:-/usr/local/spark-2.2.0-bin-hadoop2.7}

seatunnel Pipeline​

我们仅需要编写一个 seatunnel Pipeline 的配置文件即可完成数据的导入。

配置文件包括四个部分，分别是 Spark、Input、filter 和 Output。

Spark​

这一部分是 Spark 的相关配置，主要配置 Spark 执行时所需的资源大小。

spark {
  spark.app.name = "seatunnel"
  spark.executor.instances = 2
  spark.executor.cores = 1
  spark.executor.memory = "1g"
}

Input​

这一部分定义数据源，如下是从 HDFS 文件中读取 text 格式数据的配置案例。

input {
    hdfs {
        path = "hdfs://nomanode:8020/rowlog/accesslog"
        table_name = "access_log"
        format = "text"
    }
}

Filter​

在 Filter 部分，这里我们配置一系列的转化，包括正则解析将日志进行拆分、时间转换将 HTTPDATE 转化为 ClickHouse 支持的日期格式、对 Number 类型的字段进行类型转换以及通过 SQL 进行字段筛减等

filter {
    # 使用正则解析原始日志
    grok {
        source_field = "raw_message"
        pattern = '%{IP:ha_ip}\\s%{NOTSPACE:domain}\\s%{IP:remote_addr}\\s%{NUMBER:request_time}s\\s\"%{DATA:upstream_ip}\"\\s\\[%{HTTPDATE:timestamp}\\]\\s\"%{NOTSPACE:method}\\s%{DATA:url}\\s%{NOTSPACE:http_ver}\"\\s%{NUMBER:status}\\s%{NUMBER:body_bytes_send}\\s%{DATA:referer}\\s%{NOTSPACE:cookie_info}\\s\"%{DATA:user_agent}\"\\s%{DATA:uid}\\s%{DATA:session_id}\\s\"%{DATA:pool}\"\\s\"%{DATA:tag2}\"\\s%{DATA:tag3}\\s%{DATA:tag4}'
    }

    # 将"dd/MMM/yyyy:HH:mm:ss Z"格式的数据转换为
    # "yyyy/MM/dd HH:mm:ss"格式的数据
    date {
        source_field = "timestamp"
        target_field = "datetime"
        source_time_format = "dd/MMM/yyyy:HH:mm:ss Z"
        target_time_format = "yyyy/MM/dd HH:mm:ss"
    }

    # 使用SQL筛选关注的字段，并对字段进行处理
    # 甚至可以通过过滤条件过滤掉不关心的数据
    sql {
        table_name = "access"
        sql = "select substring(date, 1, 10) as date, datetime, hostname, url, http_code, float(request_time), int(data_size), domain from access"
    }
}

Output​

最后我们将处理好的结构化数据写入 ClickHouse

output {
    clickhouse {
        host = "your.clickhouse.host:8123"
        database = "seatunnel"
        table = "access_log"
        fields = ["date", "datetime", "hostname", "uri", "http_code", "request_time", "data_size", "domain"]
        username = "username"
        password = "password"
    }
}

Running seatunnel​

我们将上述四部分配置组合成为我们的配置文件 config/batch.conf。

vim config/batch.conf

spark {
  spark.app.name = "seatunnel"
  spark.executor.instances = 2
  spark.executor.cores = 1
  spark.executor.memory = "1g"
}

input {
    hdfs {
        path = "hdfs://nomanode:8020/rowlog/accesslog"
        table_name = "access_log"
        format = "text"
    }
}

filter {
    # 使用正则解析原始日志
    grok {
        source_field = "raw_message"
        pattern = '%{IP:ha_ip}\\s%{NOTSPACE:domain}\\s%{IP:remote_addr}\\s%{NUMBER:request_time}s\\s\"%{DATA:upstream_ip}\"\\s\\[%{HTTPDATE:timestamp}\\]\\s\"%{NOTSPACE:method}\\s%{DATA:url}\\s%{NOTSPACE:http_ver}\"\\s%{NUMBER:status}\\s%{NUMBER:body_bytes_send}\\s%{DATA:referer}\\s%{NOTSPACE:cookie_info}\\s\"%{DATA:user_agent}\"\\s%{DATA:uid}\\s%{DATA:session_id}\\s\"%{DATA:pool}\"\\s\"%{DATA:tag2}\"\\s%{DATA:tag3}\\s%{DATA:tag4}'
    }

    # 将"dd/MMM/yyyy:HH:mm:ss Z"格式的数据转换为
    # "yyyy/MM/dd HH:mm:ss"格式的数据
    date {
        source_field = "timestamp"
        target_field = "datetime"
        source_time_format = "dd/MMM/yyyy:HH:mm:ss Z"
        target_time_format = "yyyy/MM/dd HH:mm:ss"
    }

    # 使用SQL筛选关注的字段，并对字段进行处理
    # 甚至可以通过过滤条件过滤掉不关心的数据
    sql {
        table_name = "access"
        sql = "select substring(date, 1, 10) as date, datetime, hostname, url, http_code, float(request_time), int(data_size), domain from access"
    }
}

output {
    clickhouse {
        host = "your.clickhouse.host:8123"
        database = "seatunnel"
        table = "access_log"
        fields = ["date", "datetime", "hostname", "uri", "http_code", "request_time", "data_size", "domain"]
        username = "username"
        password = "password"
    }
}

执行命令，指定配置文件，运行 Seatunnel，即可将数据写入 ClickHouse。这里我们以本地模式为例。

./bin/start-seatunnel.sh --config config/batch.conf -e client -m 'local[2]'

Conclusion​

在这篇文章中，我们介绍了如何使用 Seatunnel 将 HDFS 中的 Nginx 日志文件导入 ClickHouse 中。仅通过一个配置文件便可快速完成数据的导入，无需编写任何代码。除了支持 HDFS 数据源之外，Seatunnel 同样支持将数据从 Kafka 中实时读取处理写入 ClickHouse 中。我们的下一篇文章将会介绍，如何将 Hive 中的数据快速导入 ClickHouse 中。

当然，Seatunnel 不仅仅是 ClickHouse 数据写入的工具，在 Elasticsearch 以及 Kafka等 数据源的写入上同样可以扮演相当重要的角色。

希望了解 Seatunnel 和 ClickHouse、Elasticsearch、Kafka 结合使用的更多功能和案例，可以直接进入官网 https://seatunnel.apache.org/

-- Power by InterestingLab

ClickHouse是面向OLAP的分布式列式DBMS。我们部门目前已经把所有数据分析相关的日志数据存储至ClickHouse这个优秀的数据仓库之中，当前日数据量达到了300亿。

在之前的文章 如何快速地把HDFS中的数据导入ClickHouse 中我们提到过使用 Seatunnel https://github.com/apache/incubator-seatunnel 对HDFS中的数据经过很简单的操作就可以将数据写入ClickHouse。HDFS中的数据一般是非结构化的数据，那么针对存储在Hive中的结构化数据，我们应该怎么操作呢？

Hive to ClickHouse​

假定我们的数据已经存储在Hive中，我们需要读取Hive表中的数据并筛选出我们关心的字段，或者对字段进行转换，最后将对应的字段写入ClickHouse的表中。

Hive Schema​

我们在Hive中存储的数据表结构如下，存储的是很常见的Nginx日志

CREATE TABLE `nginx_msg_detail`(
   `hostname` string,
   `domain` string,
   `remote_addr` string,
   `request_time` float,
   `datetime` string,
   `url` string,
   `status` int,
   `data_size` int,
   `referer` string,
   `cookie_info` string,
   `user_agent` string,
   `minute` string)
 PARTITIONED BY (
   `date` string,
   `hour` string)

ClickHouse Schema​

我们的ClickHouse建表语句如下，我们的表按日进行分区

CREATE TABLE cms.cms_msg
(
    date Date,
    datetime DateTime,
    url String,
    request_time Float32,
    status String,
    hostname String,
    domain String,
    remote_addr String,
    data_size Int32
) ENGINE = MergeTree PARTITION BY date ORDER BY (date, hostname) SETTINGS index_granularity = 16384

Seatunnel with ClickHouse​

接下来会给大家介绍，我们如何通过 Seatunnel 将Hive中的数据写入ClickHouse中。

Seatunnel​

Seatunnel 是一个非常易用，高性能，能够应对海量数据的实时数据处理产品，它构建在Spark之上。Seatunnel 拥有着非常丰富的插件，支持从Kafka、HDFS、Kudu中读取数据，进行各种各样的数据处理，并将结果写入ClickHouse、Elasticsearch或者Kafka中。

Seatunnel的环境准备以及安装步骤这里就不一一赘述了，具体安装步骤可以参考上一篇文章或者访问 Seatunnel Docs

Seatunnel Pipeline​

我们仅需要编写一个Seatunnel Pipeline的配置文件即可完成数据的导入。

配置文件包括四个部分，分别是Spark、Input、filter和Output。

Spark​

这一部分是Spark的相关配置，主要配置Spark执行时所需的资源大小。

spark {
  // 这个配置必需填写
  spark.sql.catalogImplementation = "hive"
  spark.app.name = "seatunnel"
  spark.executor.instances = 2
  spark.executor.cores = 1
  spark.executor.memory = "1g"
}

Input​

这一部分定义数据源，如下是从Hive文件中读取text格式数据的配置案例。

input {
    hive {
        pre_sql = "select * from access.nginx_msg_detail"
        table_name = "access_log"
    }
}

看，很简单的一个配置就可以从Hive中读取数据了。其中pre_sql是从Hive中读取数据SQL，table_name是将读取后的数据，注册成为Spark中临时表的表名，可为任意字段。

需要注意的是，必须保证hive的metastore是在服务状态。

在Cluster、Client、Local模式下运行时，必须把hive-site.xml文件置于提交任务节点的$HADOOP_CONF目录下

Filter​

在Filter部分，这里我们配置一系列的转化，我们这里把不需要的minute和hour字段丢弃。当然我们也可以在读取Hive的时候通过pre_sql不读取这些字段

filter {
    remove {
        source_field = ["minute", "hour"]
    }
}

Output​

最后我们将处理好的结构化数据写入ClickHouse

output {
    clickhouse {
        host = "your.clickhouse.host:8123"
        database = "seatunnel"
        table = "nginx_log"
        fields = ["date", "datetime", "hostname", "url", "http_code", "request_time", "data_size", "domain"]
        username = "username"
        password = "password"
    }
}

Running Seatunnel​

我们将上述四部分配置组合成为我们的配置文件config/batch.conf。

vim config/batch.conf

spark {
  spark.app.name = "seatunnel"
  spark.executor.instances = 2
  spark.executor.cores = 1
  spark.executor.memory = "1g"
  // 这个配置必需填写
  spark.sql.catalogImplementation = "hive"
}
input {
    hive {
        pre_sql = "select * from access.nginx_msg_detail"
        table_name = "access_log"
    }
}
filter {
    remove {
        source_field = ["minute", "hour"]
    }
}
output {
    clickhouse {
        host = "your.clickhouse.host:8123"
        database = "seatunnel"
        table = "access_log"
        fields = ["date", "datetime", "hostname", "uri", "http_code", "request_time", "data_size", "domain"]
        username = "username"
        password = "password"
    }
}

执行命令，指定配置文件，运行 Seatunnel，即可将数据写入ClickHouse。这里我们以本地模式为例。

./bin/start-seatunnel.sh --config config/batch.conf -e client -m 'local[2]'

Conclusion​

在这篇文章中，我们介绍了如何使用 Seatunnel 将Hive中的数据导入ClickHouse中。仅仅通过一个配置文件便可快速完成数据的导入，无需编写任何代码，十分简单。

希望了解 Seatunnel 与ClickHouse、Elasticsearch、Kafka、Hadoop结合使用的更多功能和案例，可以直接进入官网 https://seatunnel.apache.org/

-- Power by InterestingLab

说到数据写入 Elasticsearch，最先想到的肯定是Logstash。Logstash因为其简单上手、可扩展、可伸缩等优点被广大用户接受。但是尺有所短，寸有所长，Logstash肯定也有它无法适用的应用场景，比如：

• 海量数据ETL
• 海量数据聚合
• 多源数据处理

为了满足这些场景，很多同学都会选择Spark，借助Spark算子进行数据处理，最后将处理结果写入Elasticsearch。

我们部门之前利用Spark对Nginx日志进行分析，统计我们的Web服务访问情况，将Nginx日志每分钟聚合一次最后将结果写入Elasticsearch，然后利用Kibana配置实时监控Dashboard。Elasticsearch和Kibana都很方便、实用，但是随着类似需求越来越多，如何快速通过Spark将数据写入Elasticsearch成为了我们的一大问题。

今天给大家推荐一款能够实现数据快速写入的黑科技 Seatunnel https://github.com/apache/incubator-seatunnel 一个非常易用，高性能，能够应对海量数据的实时数据处理产品，它构建在Spark之上，简单易用，灵活配置，无需开发。

Kafka to Elasticsearch​

和Logstash一样，Seatunnel同样支持多种类型的数据输入，这里我们以最常见的Kakfa作为输入源为例，讲解如何使用 Seatunnel 将数据快速写入Elasticsearch

Log Sample​

原始日志格式如下:

127.0.0.1 elasticsearch.cn 114.250.140.241 0.001s "127.0.0.1:80" [26/Oct/2018:21:54:32 +0800] "GET /article HTTP/1.1" 200 123 "-" - "Dalvik/2.1.0 (Linux; U; Android 7.1.1; OPPO R11 Build/NMF26X)"

Elasticsearch Document​

我们想要统计，一分钟每个域名的访问情况，聚合完的数据有以下字段:

domain String
hostname String
status int
datetime String
count int

Seatunnel with Elasticsearch​

接下来会给大家详细介绍，我们如何通过 Seatunnel 读取Kafka中的数据，对数据进行解析以及聚合，最后将处理结果写入Elasticsearch中。

Seatunnel​

Seatunnel 同样拥有着非常丰富的插件，支持从Kafka、HDFS、Hive中读取数据，进行各种各样的数据处理，并将结果写入Elasticsearch、Kudu或者Kafka中。

Prerequisites​

首先我们需要安装seatunnel，安装十分简单，无需配置系统环境变量

1. 准备Spark环境
2. 安装 Seatunnel
3. 配置 Seatunnel

以下是简易步骤，具体安装可以参照 Quick Start

cd /usr/local
wget https://archive.apache.org/dist/spark/spark-2.2.0/spark-2.2.0-bin-hadoop2.7.tgz
tar -xvf https://archive.apache.org/dist/spark/spark-2.2.0/spark-2.2.0-bin-hadoop2.7.tgz
wget https://github.com/InterestingLab/seatunnel/releases/download/v1.1.1/seatunnel-1.1.1.zip
unzip seatunnel-1.1.1.zip
cd seatunnel-1.1.1

vim config/seatunnel-env.sh
# 指定Spark安装路径
SPARK_HOME=${SPARK_HOME:-/usr/local/spark-2.2.0-bin-hadoop2.7}

Seatunnel Pipeline​

与Logstash一样，我们仅需要编写一个Seatunnel Pipeline的配置文件即可完成数据的导入，相信了解Logstash的朋友可以很快入手 Seatunnel 配置。

配置文件包括四个部分，分别是Spark、Input、filter和Output。

Spark​

这一部分是Spark的相关配置，主要配置Spark执行时所需的资源大小。

spark {
  spark.app.name = "seatunnel"
  spark.executor.instances = 2
  spark.executor.cores = 1
  spark.executor.memory = "1g"
  spark.streaming.batchDuration = 5
}

Input​

这一部分定义数据源，如下是从Kafka中读取数据的配置案例，

kafkaStream {
    topics = "seatunnel-es"
    consumer.bootstrap.servers = "localhost:9092"
    consumer.group.id = "seatunnel_es_group"
    consumer.rebalance.max.retries = 100
}

Filter​

在Filter部分，这里我们配置一系列的转化，包括正则解析将日志进行拆分、时间转换将HTTPDATE转化为Elasticsearch支持的日期格式、对Number类型的字段进行类型转换以及通过SQL进行数据聚合

filter {
    # 使用正则解析原始日志
    # 最开始数据都在raw_message字段中
    grok {
        source_field = "raw_message"
        pattern = '%{NOTSPACE:hostname}\\s%{NOTSPACE:domain}\\s%{IP:remote_addr}\\s%{NUMBER:request_time}s\\s\"%{DATA:upstream_ip}\"\\s\\[%{HTTPDATE:timestamp}\\]\\s\"%{NOTSPACE:method}\\s%{DATA:url}\\s%{NOTSPACE:http_ver}\"\\s%{NUMBER:status}\\s%{NUMBER:body_bytes_send}\\s%{DATA:referer}\\s%{NOTSPACE:cookie_info}\\s\"%{DATA:user_agent}'
   }
    # 将"dd/MMM/yyyy:HH:mm:ss Z"格式的数据转换为
    # Elasticsearch中支持的格式
    date {
        source_field = "timestamp"
        target_field = "datetime"
        source_time_format = "dd/MMM/yyyy:HH:mm:ss Z"
        target_time_format = "yyyy-MM-dd'T'HH:mm:ss.SSS+08:00"
    }
    ## 利用SQL对数据进行聚合
    sql {
        table_name = "access_log"
        sql = "select domain, hostname, int(status), datetime, count(*) from access_log group by domain, hostname, status, datetime"
    }
 }

Output​

最后我们将处理好的结构化数据写入Elasticsearch。

output {
    elasticsearch {
        hosts = ["localhost:9200"]
        index = "seatunnel-${now}"
        es.batch.size.entries = 100000
        index_time_format = "yyyy.MM.dd"
    }
}

Running Seatunnel​

我们将上述四部分配置组合成为我们的配置文件 config/batch.conf。

vim config/batch.conf

spark {
  spark.app.name = "seatunnel"
  spark.executor.instances = 2
  spark.executor.cores = 1
  spark.executor.memory = "1g"
  spark.streaming.batchDuration = 5
}
input {
    kafkaStream {
        topics = "seatunnel-es"
        consumer.bootstrap.servers = "localhost:9092"
        consumer.group.id = "seatunnel_es_group"
        consumer.rebalance.max.retries = 100
    }
}
filter {
    # 使用正则解析原始日志
    # 最开始数据都在raw_message字段中
    grok {
        source_field = "raw_message"
        pattern = '%{IP:hostname}\\s%{NOTSPACE:domain}\\s%{IP:remote_addr}\\s%{NUMBER:request_time}s\\s\"%{DATA:upstream_ip}\"\\s\\[%{HTTPDATE:timestamp}\\]\\s\"%{NOTSPACE:method}\\s%{DATA:url}\\s%{NOTSPACE:http_ver}\"\\s%{NUMBER:status}\\s%{NUMBER:body_bytes_send}\\s%{DATA:referer}\\s%{NOTSPACE:cookie_info}\\s\"%{DATA:user_agent}'
   }
    # 将"dd/MMM/yyyy:HH:mm:ss Z"格式的数据转换为
    # Elasticsearch中支持的格式
    date {
        source_field = "timestamp"
        target_field = "datetime"
        source_time_format = "dd/MMM/yyyy:HH:mm:ss Z"
        target_time_format = "yyyy-MM-dd'T'HH:mm:00.SSS+08:00"
    }
    ## 利用SQL对数据进行聚合
    sql {
        table_name = "access_log"
        sql = "select domain, hostname, status, datetime, count(*) from access_log group by domain, hostname, status, datetime"
    }
 }
output {
    elasticsearch {
        hosts = ["localhost:9200"]
        index = "seatunnel-${now}"
        es.batch.size.entries = 100000
        index_time_format = "yyyy.MM.dd"
    }
}

执行命令，指定配置文件，运行 Seatunnel，即可将数据写入Elasticsearch。这里我们以本地模式为例。

./bin/start-seatunnel.sh --config config/batch.conf -e client -m 'local[2]'

最后，写入Elasticsearch中的数据如下，再配上Kibana就可以实现Web服务的实时监控了^_^.

"_source": {
    "domain": "elasticsearch.cn",
    "hostname": "localhost",
    "status": "200",
    "datetime": "2018-11-26T21:54:00.000+08:00",
    "count": 26
  }

Conclusion​

在这篇文章中，我们介绍了如何通过 Seatunnel 将Kafka中的数据写入Elasticsearch中。仅仅通过一个配置文件便可快速运行一个Spark Application，完成数据的处理、写入，无需编写任何代码，十分简单。

当数据处理过程中有遇到Logstash无法支持的场景或者Logstah性能无法达到预期的情况下，都可以尝试使用 Seatunnel 解决问题。

希望了解 Seatunnel 与Elasticsearch、Kafka、Hadoop结合使用的更多功能和案例，可以直接进入官网 https://seatunnel.apache.org/

我们近期会再发布一篇《如何用Spark和Elasticsearch做交互式数据分析》，敬请期待.

Contract us​

• 邮件列表 : dev@seatunnel.apache.org. 发送任意内容至 dev-subscribe@seatunnel.apache.org， 按照回复订阅邮件列表。
• Slack: 发送 Request to join SeaTunnel slack 邮件到邮件列表 (dev@seatunnel.apache.org), 我们会邀请你加入（在此之前请确认已经注册Slack）.
• bilibili B站 视频

TiDB 是一款定位于在线事务处理/在线分析处理的融合型数据库产品，实现了一键水平伸缩，强一致性的多副本数据安全，分布式事务，实时 OLAP 等重要特性。

TiSpark 是 PingCAP 为解决用户复杂 OLAP 需求而推出的产品。它借助 Spark 平台，同时融合 TiKV 分布式集群的优势。

直接使用 TiSpark 完成 OLAP 操作需要了解 Spark，还需要一些开发工作。那么，有没有一些开箱即用的工具能帮我们更快速地使用 TiSpark 在 TiDB 上完成 OLAP 分析呢？

目前开源社区上有一款工具 Seatunnel，项目地址 https://github.com/apache/incubator-seatunnel ，可以基于Spark，在 TiSpark 的基础上快速实现 TiDB 数据读取和 OLAP 分析。

使用 Seatunnel 操作TiDB​

在我们线上有这么一个需求，从 TiDB 中读取某一天的网站访问数据，统计每个域名以及服务返回状态码的访问次数，最后将统计结果写入 TiDB 另外一个表中。 我们来看看 Seatunnel 是如何实现这么一个功能的。

Seatunnel​

Seatunnel 是一个非常易用，高性能，能够应对海量数据的实时数据处理产品，它构建在 Spark 之上。Seatunnel 拥有着非常丰富的插件，支持从 TiDB、Kafka、HDFS、Kudu 中读取数据，进行各种各样的数据处理，然后将结果写入 TiDB、ClickHouse、Elasticsearch 或者 Kafka 中。

准备工作​

1. TiDB 表结构介绍​

Input（存储访问日志的表）

CREATE TABLE access_log (
    domain VARCHAR(255),
    datetime VARCHAR(63),
    remote_addr VARCHAR(63),
    http_ver VARCHAR(15),
    body_bytes_send INT,
    status INT,
    request_time FLOAT,
    url TEXT
)

+-----------------+--------------+------+------+---------+-------+
| Field           | Type         | Null | Key  | Default | Extra |
+-----------------+--------------+------+------+---------+-------+
| domain          | varchar(255) | YES  |      | NULL    |       |
| datetime        | varchar(63)  | YES  |      | NULL    |       |
| remote_addr     | varchar(63)  | YES  |      | NULL    |       |
| http_ver        | varchar(15)  | YES  |      | NULL    |       |
| body_bytes_send | int(11)      | YES  |      | NULL    |       |
| status          | int(11)      | YES  |      | NULL    |       |
| request_time    | float        | YES  |      | NULL    |       |
| url             | text         | YES  |      | NULL    |       |
+-----------------+--------------+------+------+---------+-------+

Output（存储结果数据的表）

CREATE TABLE access_collect (
    date VARCHAR(23),
    domain VARCHAR(63),
    status INT,
    hit INT
)

+--------+-------------+------+------+---------+-------+
| Field  | Type        | Null | Key  | Default | Extra |
+--------+-------------+------+------+---------+-------+
| date   | varchar(23) | YES  |      | NULL    |       |
| domain | varchar(63) | YES  |      | NULL    |       |
| status | int(11)     | YES  |      | NULL    |       |
| hit    | int(11)     | YES  |      | NULL    |       |
+--------+-------------+------+------+---------+-------+

2. 安装 Seatunnel​

有了 TiDB 输入和输出表之后， 我们需要安装 Seatunnel，安装十分简单，无需配置系统环境变量

1. 准备 Spark环境
2. 安装 Seatunnel
3. 配置 Seatunnel

以下是简易步骤，具体安装可以参照 Quick Start

# 下载安装Spark
cd /usr/local
wget https://archive.apache.org/dist/spark/spark-2.1.0/spark-2.1.0-bin-hadoop2.7.tgz
tar -xvf https://archive.apache.org/dist/spark/spark-2.1.0/spark-2.1.0-bin-hadoop2.7.tgz
wget
# 下载安装seatunnel
https://github.com/InterestingLab/seatunnel/releases/download/v1.2.0/seatunnel-1.2.0.zip
unzip seatunnel-1.2.0.zip
cd seatunnel-1.2.0

vim config/seatunnel-env.sh
# 指定Spark安装路径
SPARK_HOME=${SPARK_HOME:-/usr/local/spark-2.1.0-bin-hadoop2.7}

实现 Seatunnel 处理流程​

我们仅需要编写一个 Seatunnel 配置文件即可完成数据的读取、处理、写入。

Seatunnel 配置文件由四个部分组成，分别是 Spark、Input、Filter 和 Output。Input 部分用于指定数据的输入源，Filter 部分用于定义各种各样的数据处理、聚合，Output 部分负责将处理之后的数据写入指定的数据库或者消息队列。

整个处理流程为 Input -> Filter -> Output，整个流程组成了 Seatunnel 的 处理流程（Pipeline）。

以下是一个具体配置，此配置来源于线上实际应用，但是为了演示有所简化。

Input (TiDB)​

这里部分配置定义输入源，如下是从 TiDB 一张表中读取数据。

input {
    tidb {
        database = "nginx"
        pre_sql = "select * from nginx.access_log"
        table_name = "spark_nginx_input"
    }
}

Filter​

在Filter部分，这里我们配置一系列的转化, 大部分数据分析的需求，都是在Filter完成的。Seatunnel 提供了丰富的插件，足以满足各种数据分析需求。这里我们通过 SQL 插件完成数据的聚合操作。

filter {
    sql {
        table_name = "spark_nginx_log"
        sql = "select count(*) as hit, domain, status, substring(datetime, 1, 10) as date from spark_nginx_log where substring(datetime, 1, 10)='2019-01-20' group by domain, status, substring(datetime, 1, 10)"
    }
}

Output (TiDB)​

最后， 我们将处理后的结果写入TiDB另外一张表中。TiDB Output是通过JDBC实现的

output {
    tidb {
        url = "jdbc:mysql://127.0.0.1:4000/nginx?useUnicode=true&characterEncoding=utf8"
        table = "access_collect"
        user = "username"
        password = "password"
        save_mode = "append"
    }
}

Spark​

这一部分是 Spark 的相关配置，主要配置 Spark 执行时所需的资源大小以及其他 Spark 配置。

我们的 TiDB Input 插件是基于 TiSpark 实现的，而 TiSpark 依赖于 TiKV 集群和 Placement Driver (PD)。因此我们需要指定 PD 节点信息以及 TiSpark 相关配置spark.tispark.pd.addresses和spark.sql.extensions。

spark {
  spark.app.name = "seatunnel-tidb"
  spark.executor.instances = 2
  spark.executor.cores = 1
  spark.executor.memory = "1g"
  # Set for TiSpark
  spark.tispark.pd.addresses = "localhost:2379"
  spark.sql.extensions = "org.apache.spark.sql.TiExtensions"
}

运行 Seatunnel​

我们将上述四部分配置组合成我们最终的配置文件 conf/tidb.conf

spark {
    spark.app.name = "seatunnel-tidb"
    spark.executor.instances = 2
    spark.executor.cores = 1
    spark.executor.memory = "1g"
    # Set for TiSpark
    spark.tispark.pd.addresses = "localhost:2379"
    spark.sql.extensions = "org.apache.spark.sql.TiExtensions"
}
input {
    tidb {
        database = "nginx"
        pre_sql = "select * from nginx.access_log"
        table_name = "spark_table"
    }
}
filter {
    sql {
        table_name = "spark_nginx_log"
        sql = "select count(*) as hit, domain, status, substring(datetime, 1, 10) as date from spark_nginx_log where substring(datetime, 1, 10)='2019-01-20' group by domain, status, substring(datetime, 1, 10)"
    }
}
output {
    tidb {
        url = "jdbc:mysql://127.0.0.1:4000/nginx?useUnicode=true&characterEncoding=utf8"
        table = "access_collect"
        user = "username"
        password = "password"
        save_mode = "append"
    }
}

执行命令，指定配置文件，运行 Seatunnel ，即可实现我们的数据处理逻辑。

• Local

./bin/start-seatunnel.sh --config config/tidb.conf --deploy-mode client --master 'local[2]'

• yarn-client

./bin/start-seatunnel.sh --config config/tidb.conf --deploy-mode client --master yarn

• yarn-cluster

./bin/start-seatunnel.sh --config config/tidb.conf --deploy-mode cluster -master yarn

如果是本机测试验证逻辑，用本地模式（Local）就可以了，一般生产环境下，都是使用yarn-client或者yarn-cluster模式。

检查结果​

mysql> select * from access_collect;
+------------+--------+--------+------+
| date       | domain | status | hit  |
+------------+--------+--------+------+
| 2019-01-20 | b.com  |    200 |   63 |
| 2019-01-20 | a.com  |    200 |   85 |
+------------+--------+--------+------+
2 rows in set (0.21 sec)

总结​

在这篇文章中，我们介绍了如何使用 Seatunnel 从 TiDB 中读取数据，做简单的数据处理之后写入 TiDB 另外一个表中。仅通过一个配置文件便可快速完成数据的导入，无需编写任何代码。

除了支持 TiDB 数据源之外，Seatunnel 同样支持Elasticsearch, Kafka, Kudu, ClickHouse等数据源。

于此同时，我们正在研发一个重要功能，就是在 Seatunnel 中，利用 TiDB 的事务特性，实现从 Kafka 到 TiDB 流式数据处理，并且支持端（Kafka）到端（TiDB）的 Exactly-Once 数据一致性。

希望了解 Seatunnel 和 TiDB，ClickHouse、Elasticsearch、Kafka结合使用的更多功能和案例，可以直接进入官网 https://seatunnel.apache.org/

联系我们​

• 邮件列表 : dev@seatunnel.apache.org. 发送任意内容至 dev-subscribe@seatunnel.apache.org， 按照回复订阅邮件列表。
• Slack: 发送 Request to join SeaTunnel slack 邮件到邮件列表 (dev@seatunnel.apache.org), 我们会邀请你加入（在此之前请确认已经注册Slack）.
• bilibili B站 视频

-- Power by InterestingLab

前言​

StructuredStreaming是Spark 2.0以后新开放的一个模块，相比SparkStreaming，它有一些比较突出的优点：
  一、它能做到更低的延迟;
  二、可以做实时的聚合，例如实时计算每天每个商品的销售总额；
  三、可以做流与流之间的关联，例如计算广告的点击率，需要将广告的曝光记录和点击记录关联。
以上几点如果使用SparkStreaming来实现可能会比较麻烦或者说是很难实现，但是使用StructuredStreaming实现起来会比较轻松。

如何使用StructuredStreaming​

可能你没有详细研究过StructuredStreaming，但是发现StructuredStreaming能很好的解决你的需求，如何快速利用StructuredStreaming来解决你的需求？目前社区有一款工具 Seatunnel，项目地址：https://github.com/apache/incubator-seatunnel , 可以高效低成本的帮助你利用StructuredStreaming来完成你的需求。

Seatunnel​

Seatunnel 是一个非常易用，高性能，能够应对海量数据的实时数据处理产品，它构建在Spark之上。Seatunnel 拥有着非常丰富的插件，支持从Kafka、HDFS、Kudu中读取数据，进行各种各样的数据处理，并将结果写入ClickHouse、Elasticsearch或者Kafka中

准备工作​

首先我们需要安装 Seatunnel，安装十分简单，无需配置系统环境变量

1. 准备Spark环境
2. 安装 Seatunnel
3. 配置 Seatunnel

以下是简易步骤，具体安装可以参照 Quick Start

cd /usr/local
wget https://archive.apache.org/dist/spark/spark-2.2.0/spark-2.2.0-bin-hadoop2.7.tgz
tar -xvf https://archive.apache.org/dist/spark/spark-2.2.0/spark-2.2.0-bin-hadoop2.7.tgz
wget https://github.com/InterestingLab/seatunnel/releases/download/v1.3.0/seatunnel-1.3.0.zip
unzip seatunnel-1.3.0.zip
cd seatunnel-1.3.0

vim config/seatunnel-env.sh
# 指定Spark安装路径
SPARK_HOME=${SPARK_HOME:-/usr/local/spark-2.2.0-bin-hadoop2.7}

Seatunnel Pipeline​

我们仅需要编写一个 Seatunnel Pipeline的配置文件即可完成数据的导入。

配置文件包括四个部分，分别是Spark、Input、filter和Output。

Spark​

这一部分是Spark的相关配置，主要配置Spark执行时所需的资源大小。

spark {
  spark.app.name = "seatunnel"
  spark.executor.instances = 2
  spark.executor.cores = 1
  spark.executor.memory = "1g"
}

Input​

下面是一个从kafka读取数据的例子

kafkaStream {
    topics = "seatunnel"
    consumer.bootstrap.servers = "localhost:9092"
    schema = "{\"name\":\"string\",\"age\":\"integer\",\"addrs\":{\"country\":\"string\",\"city\":\"string\"}}"
}

通过上面的配置就可以读取kafka里的数据了 ，topics是要订阅的kafka的topic，同时订阅多个topic可以以逗号隔开，consumer.bootstrap.servers就是Kafka的服务器列表，schema是可选项，因为StructuredStreaming从kafka读取到的值(官方固定字段value)是binary类型的，详见http://spark.apache.org/docs/latest/structured-streaming-kafka-integration.html 但是如果你确定你kafka里的数据是json字符串的话，你可以指定schema，input插件将按照你指定的schema解析

Filter​

下面是一个简单的filter例子

filter{
    sql{
        table_name = "student"
        sql = "select name,age from student"
    }
}

table_name是注册成的临时表名，以便于在下面的sql使用

Output​

处理好的数据往外输出，假设我们的输出也是kafka

output{
    kafka {
        topic = "seatunnel"
        producer.bootstrap.servers = "localhost:9092"
        streaming_output_mode = "update"
        checkpointLocation = "/your/path"
    }
}

topic 是你要输出的topic， producer.bootstrap.servers是kafka集群列表，streaming_output_mode是StructuredStreaming的一个输出模式参数，有三种类型append|update|complete，具体使用参见文档http://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#output-modes

checkpointLocation是StructuredStreaming的checkpoint路径，如果配置了的话，这个目录会存储程序的运行信息，比如程序退出再启动的话会接着上次的offset进行消费。

场景分析​

以上就是一个简单的例子，接下来我们就来介绍的稍微复杂一些的业务场景

场景一：实时聚合场景​

假设现在有一个商城，上面有10种商品，现在需要实时求每天每种商品的销售额，甚至是求每种商品的购买人数（不要求十分精确）。 这么做的巨大的优势就是海量数据可以在实时处理的时候，完成聚合，再也不需要先将数据写入数据仓库，再跑离线的定时任务进行聚合， 操作起来还是很方便的。

kafka的数据如下

{"good_id":"abc","price":300,"user_id":123456,"time":1553216320}

那我们该怎么利用 Seatunnel 来完成这个需求呢，当然还是只需要配置就好了。

#spark里的配置根据业务需求配置
spark {
  spark.app.name = "seatunnel"
  spark.executor.instances = 2
  spark.executor.cores = 1
  spark.executor.memory = "1g"
}

#配置input
input {
    kafkaStream {
        topics = "good_topic"
        consumer.bootstrap.servers = "localhost:9092"
        schema = "{\"good_id\":\"string\",\"price\":\"integer\",\"user_id\":\"Long\",\"time\":\"Long\"}"
    }
}

#配置filter    
filter {
    
    #在程序做聚合的时候，内部会去存储程序从启动开始的聚合状态，久而久之会导致OOM,如果设置了watermark，程序自动的会去清理watermark之外的状态
    #这里表示使用ts字段设置watermark，界限为1天

    Watermark {
        time_field = "time"
        time_type = "UNIX"              #UNIX表示时间字段为10为的时间戳，还有其他的类型详细可以查看插件文档
        time_pattern = "yyyy-MM-dd"     #这里之所以要把ts对其到天是因为求每天的销售额，如果是求每小时的销售额可以对其到小时`yyyy-MM-dd HH`
        delay_threshold = "1 day"
        watermark_field = "ts"          #设置watermark之后会新增一个字段，`ts`就是这个字段的名字
    }
    
    #之所以要group by ts是要让watermark生效，approx_count_distinct是一个估值，并不是精确的count_distinct
    sql {
        table_name = "good_table_2"
        sql = "select good_id,sum(price) total, approx_count_distinct(user_id) person from good_table_2 group by ts,good_id"
    }
}

#接下来我们选择将结果实时输出到Kafka
output{
    kafka {
        topic = "seatunnel"
        producer.bootstrap.servers = "localhost:9092"
        streaming_output_mode = "update"
        checkpointLocation = "/your/path"
    }
}

如上配置完成，启动 Seatunnel，就可以获取你想要的结果了。

场景二：多个流关联场景​

假设你在某个平台投放了广告，现在要实时计算出每个广告的CTR(点击率)，数据分别来自两个topic，一个是广告曝光日志，一个是广告点击日志, 此时我们就需要把两个流数据关联到一起做计算，而 Seatunnel 最近也支持了此功能，让我们一起看一下该怎么做：

点击topic数据格式

{"ad_id":"abc","click_time":1553216320,"user_id":12345}

曝光topic数据格式

{"ad_id":"abc","show_time":1553216220,"user_id":12345}

#spark里的配置根据业务需求配置
spark {
  spark.app.name = "seatunnel"
  spark.executor.instances = 2
  spark.executor.cores = 1
  spark.executor.memory = "1g"
}

#配置input
input {
    
    kafkaStream {
        topics = "click_topic"
        consumer.bootstrap.servers = "localhost:9092"
        schema = "{\"ad_id\":\"string\",\"user_id\":\"Long\",\"click_time\":\"Long\"}"
        table_name = "click_table"
    }
    
    kafkaStream {
        topics = "show_topic"
        consumer.bootstrap.servers = "localhost:9092"
        schema = "{\"ad_id\":\"string\",\"user_id\":\"Long\",\"show_time\":\"Long\"}"
        table_name = "show_table"
    }
}

filter {
    
    #左关联右表必须设置watermark
    #右关左右表必须设置watermark
    #http://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#inner-joins-with-optional-watermarking
    Watermark {
              source_table_name = "click_table" #这里可以指定为某个临时表添加watermark，不指定的话就是为input中的第一个
              time_field = "time"
              time_type = "UNIX"               
              delay_threshold = "3 hours"
              watermark_field = "ts" 
              result_table_name = "click_table_watermark" #添加完watermark之后可以注册成临时表，方便后续在sql中使用
    }
    
    Watermark {
                source_table_name = "show_table" 
                time_field = "time"
                time_type = "UNIX"               
                delay_threshold = "2 hours"
                watermark_field = "ts" 
                result_table_name = "show_table_watermark" 
     }
    
    
    sql {
        table_name = "show_table_watermark"
        sql = "select a.ad_id,count(b.user_id)/count(a.user_id) ctr from show_table_watermark as a left join click_table_watermark as b on a.ad_id = b.ad_id and a.user_id = b.user_id "
    }
    
}

#接下来我们选择将结果实时输出到Kafka
output {
    kafka {
        topic = "seatunnel"
        producer.bootstrap.servers = "localhost:9092"
        streaming_output_mode = "append" #流关联只支持append模式
        checkpointLocation = "/your/path"
    }
}

通过配置，到这里流关联的案例也完成了。

结语​

通过配置能很快的利用StructuredStreaming做实时数据处理，但是还是需要对StructuredStreaming的一些概念了解，比如其中的watermark机制，还有程序的输出模式。

最后，Seatunnel 当然还支持spark streaming和spark 批处理。 如果你对这两个也感兴趣的话，可以阅读我们以前发布的文章《如何快速地将Hive中的数据导入ClickHouse》、 《优秀的数据工程师，怎么用Spark在TiDB上做OLAP分析》、 《如何使用Spark快速将数据写入Elasticsearch》

希望了解 Seatunnel 和 HBase, ClickHouse、Elasticsearch、Kafka、MySQL 等数据源结合使用的更多功能和案例，可以直接进入官网 https://seatunnel.apache.org/

联系我们​

• 邮件列表 : dev@seatunnel.apache.org. 发送任意内容至 dev-subscribe@seatunnel.apache.org， 按照回复订阅邮件列表。
• Slack: 发送 Request to join SeaTunnel slack 邮件到邮件列表 (dev@seatunnel.apache.org), 我们会邀请你加入（在此之前请确认已经注册Slack）.
• bilibili B站 视频