4. 批量数据同步：基于 Apache SeaTunnel 的 API 到数据湖同步

好，咱们进入第四章。前面几章我们把流批一体的理论、API 数据源的接入方式都聊了一遍。这一章，咱们要动真格的了——用 Apache SeaTunnel 把 API 里的数据，批量同步到数据湖里。

说白了，就是解决一个很实际的问题：你有个第三方系统，只给你暴露了 HTTP 接口，你怎么把它的几百万、几千万条历史数据，高效、稳定地搬到你的 Hive 或 Hudi 表里？

我个人习惯，遇到这种场景，第一反应就是 SeaTunnel。为什么？因为它配置化程度高，不用写一行 Spark 或 Flink 代码，改个配置文件就能跑。我曾在项目里用它同步过某电商平台的订单数据，每天几千万条，跑了半年没出过问题。

4.1 为什么选 SeaTunnel 做 API 批量同步？

你可能会问，用 Flink 或 Spark 自己写个程序不行吗？当然行。但你想过没有，如果对接的 API 有几十个，每个都要写一遍连接、解析、写入的逻辑，那维护成本得多高？

SeaTunnel 的核心优势在于：

• 插件化架构：Source、Transform、Sink 都是插件，API 连接器、数据湖连接器开箱即用。
• 配置驱动：所有逻辑写在配置文件里，改配置就等于改逻辑。
• 全量+增量一体：通过配置就能实现全量同步和增量同步的切换。
• Exactly-Once 语义：配合 Checkpoint，数据不会丢也不会重复。

4.2 整体架构：API → SeaTunnel → 数据湖

我们先画一张图，把整体流程看清楚。这张图我每次培训都会拿出来讲，因为它把数据流向讲得很明白。

这张图里，我特意把「配置文件」和「调度器」也画出来了。因为在实际生产中，SeaTunnel 很少手动跑，都是挂在调度系统上定时执行。

4.3 全量同步：一次性拉取所有数据

全量同步，说白了就是「不管三七二十一，把 API 里所有数据都拉下来」。适合第一次建表，或者数据量不大的场景。

4.3.1 配置示例

我们来看一个实际的配置文件。这个配置我曾在某项目的初始化阶段用过，把 CRM 系统的客户数据全量同步到 Hive。

env {
  job.mode = "BATCH"
  job.name = "crm_customer_full_sync"
}

source {
  Http {
    url = "https://api.example.com/v1/customers"
    method = "GET"
    headers {
      Authorization = "Bearer ${API_}"
      Content-Type = "application/json"
    }
    format = "json"
    schema {
      fields {
        customer_id = "string"
        name = "string"
        email = "string"
        created_at = "string"
        updated_at = "string"
      }
    }
    // 分页配置
    pagination {
      type = "PageNumber"
      page_size = 1000
      page_field = "page"
      size_field = "page_size"
      total_field = "total"
    }
  }
}

transform {
  // 可以加一些数据清洗逻辑
  Filter {
    field = "customer_id"
    condition = "IS NOT NULL"
  }
}

sink {
  Hive {
    table = "ods.crm_customer"
    metastore_uri = "thrift://hive-metastore:9083"
    save_mode = "overwrite"
  }
}

4.3.2 分页处理

API 一般不会一次性返回所有数据，都会做分页。SeaTunnel 的 HTTP 连接器内置了分页支持，你只需要配置分页类型和字段名。

分页类型	说明	适用场景
PageNumber	通过页码和每页大小控制	RESTful API 最常见
Cursor	通过游标或偏移量控制	数据量大、实时性高的 API
Offset	通过起始偏移量和限制条数控制	兼容旧版 API

我曾经遇到过一个坑：某 API 的分页字段叫 pageNum，但文档写的是 page。结果配置错了，跑了半天只拉了第一页数据。嗯，这里要注意，一定要先手工调一下 API，确认分页字段名和返回值结构。

4.4 增量同步：只拉取变化的数据

全量同步搞定了，但你不能每天全量拉一遍吧？数据量大了以后，全量同步会越来越慢，而且浪费带宽和计算资源。

这时候就需要增量同步了。增量同步的核心思路是：只拉取上次同步之后发生变化的数据。

4.4.1 基于时间戳的增量

最常用的方式，就是利用 API 的 updated_at 或 modified_time 字段。你想想看，只要 API 支持按时间范围过滤，增量同步就变得很简单。

source {
  Http {
    url = "https://api.example.com/v1/customers"
    method = "GET"
    headers {
      Authorization = "Bearer ${API_}"
    }
    format = "json"
    // 增量参数
    request_params {
      updated_after = "${last_sync_time}"
      updated_before = "${current_time}"
    }
    schema {
      fields {
        customer_id = "string"
        name = "string"
        email = "string"
        updated_at = "string"
      }
    }
    pagination {
      type = "PageNumber"
      page_size = 1000
    }
  }
}

sink {
  Hudi {
    table = "ods.crm_customer"
    base_path = "/data/hudi/crm_customer"
    table_type = "COPY_ON_WRITE"
    record_key = "customer_id"
    precombine_field = "updated_at"
    save_mode = "append"
  }
}

4.4.2 增量状态的维护

增量同步有个绕不开的问题：last_sync_time 存在哪？

我建议的做法是：

• 方案一：存在数据库或 Redis 里，每次同步前读取，同步后更新。
• 方案二：利用 SeaTunnel 的 variable 功能，从外部传入。
• 方案三：查目标表的最大时间戳，作为增量起点。

我个人习惯用方案三，因为它不需要额外的存储组件。代码里写一句 SQL 就能拿到最大时间戳，然后传给 SeaTunnel 作为参数。

4.5 全量与增量的切换策略

在实际项目中，全量和增量不是非此即彼的关系。我通常的做法是：

1. 首次运行：全量同步，把历史数据一次性拉过来。
2. 日常运行：增量同步，每隔 10 分钟或每小时拉一次变化数据。
3. 定期全量：每周或每月做一次全量同步，用于数据校正。

这个策略，说白了就是「增量保实时，全量保准确」。你想想看，如果增量同步因为网络抖动丢了几条数据，定期全量就能把它补回来。

4.6 性能优化与避坑

最后，我分享几个实战中总结出来的优化点：

问题	优化方案	我的经验
API 限流	配置 rate_limit 参数，控制请求频率	某 API 限制 100 次/分钟，我设了 80 次/分钟，稳如老狗
数据倾斜	在 Transform 阶段做 repartition	按日期字段分区，避免单个分区数据量过大
连接超时	设置 connect_timeout 和 read_timeout	默认 30 秒，我一般设 60 秒，防止网络抖动导致任务失败
数据湖小文件	Hudi 或 Iceberg 的 compaction 策略	增量同步容易产生小文件，建议每小时做一次 compaction

好了，这一章的内容就到这里。配置化实现 API 到数据湖的同步，说白了就是「配好 Source、配好 Sink、选对模式」。全量用 overwrite，增量用 append，再加上分页和限流，基本就能应对大部分场景了。