背景介绍

SeaTunnel 是一个开源、易用、多模态、超高性能的分布式数据集成平台，支持实时海量数据同步。具体项目介绍和使用方法可以参考官方文档。本文将介绍如何在流式计算 Flink 版中部署并使用 SeaTunnel 产品。

注意：SeaTunnel 属于开源产品。流式计算 Flink 版不会对 SeaTunnel 开源项目本身的问题提供任何支持和 SLA 保障。本文档仅供部署参考，请用户使用前仔细学习并理解社区文档。

前提条件

在使用产品前，需要注意以下几点

1. 需要开通流式计算 Flink 版，并购买资源池资源
2. 需要下载 SeaTunnel 产品，并且熟悉产品的使用方法

使用方法

SeaTunnel 部署，我们主要包含以下几个步骤：

1. 下载并且安装 SeaTunnel 插件
2. 按照 SeaTunnel 官方文档进行参数配置
3. 创建 Flink JAR 作业，并且上传相关资源文件、配置自定义参数
4. 启动任务，观察任务状态

准备 SeaTunnel

参考官方文档，下载 SeaTunnel 安装包，本文以 2.3.12 版本为例

export version="2.3.12"
wget "https://archive.apache.org/dist/seatunnel/${version}/apache-seatunnel-${version}-bin.tar.gz"
tar -xzvf "apache-seatunnel-${version}-bin.tar.gz"

在 SeaTunnel 项目目录下，安装插件（如果只依赖 Faker 和 Console 这两个测试 Connector，本步骤可以跳过）：

sh bin/install-plugin.sh 2.3.12

将生成目录文件压缩打包：

export version="2.3.12"
cd ..
tar czf apache-seatunnel-${version}-all.tar.gz apache-seatunnel-${version}

因为该压缩包较大，所以我们提前通过 TOS 控制台上传到 TOS 桶上，以供后续 Flink 服务使用：

上传完成后，记录下 tos 的存储路径，例如 tos://<bucket_name>/path/to/apache-seatunnel-${version}-all.tar.gz

任务配置

步骤一：在流式计算 Flink 版控制台创建 Flink JAR 作业，因为 SeaTunnel 版本支持情况，建议选择 Flink 1.16 版本测试使用。

注意：本步骤以 Flink 1.16 的流式同步任务为例。如果 SeaTunnel 任务类型为批式作业，请结合实际情况调整

步骤二：上传 JAR 包并设置相关启动参数：

其中几个需要设置的参数

1. Jar URI：需要上传 SeaTunnel 安装包 starter 目录下的 seatunnel-flink-15-starter.jar 作业
2. Entrypoint Class： 填写主类全路径名 org.apache.seatunnel.core.starter.flink.SeaTunnelFlink
3. Entrypoint main args：填写命令参数 --config /opt/tiger/workdir/v2.streaming.conf。其中
   1. /opt/tiger/workdir/是容器内固定的文件加载路径，无需修改
   2. v2.streaming.conf 是 SeaTunnel 的配置文件名，可以根据您的配置文件名调整

步骤三：准备并上传 SeaTunnel 的配置文件， 本文以 v2.streaming.conf 为例，本地生成如下配置文件。其中配置的含义是生成一个流式 Flink 作业，通过 Fake Source 生成数据，写入 TOS 的 Paimon 数据表（具体参数含义请参考 SeaTunnel 的官方文档）

env {
  parallelism = 1
  job.mode = "STREAMING"
}

source {
  FakeSource {
    plugin_output = "fake"
    row.num = 16
    schema = {
      fields {
        name = "string"
        age = "int"
      }
    }
  }
}

transform {
  FieldMapper {
    plugin_input = "fake"
    plugin_output = "fake1"
    field_mapper = {
      age = age
      name = new_name
    }
  }
}

sink {
  Paimon {
    warehouse = "tos://<Bucket_Name>/<Path>/"
    database = "st"
    table = "st_test_sink"
  }
}

步骤四：上传配置文件，以及引用上一步准备的 SeaTunnel 依赖文件包

1. 配置文件：通过文件上传并选择上一步准备的 v2.streaming.conf 的配置文件
2. SeaTunnel 依赖文件：通过填写 TOS 地址，设置为第一步 TOS 桶中的指定路径 tos://<bucket_name>/path/to/apache-seatunnel-${version}-all.tar.gz

自定义参数

在自定义参数中设置如下参数

containerized.master.env.SEATUNNEL_HOME: /opt/tiger/workdir/apache-seatunnel-2.3.12
containerized.taskmanager.env.SEATUNNEL_HOME: /opt/tiger/workdir/apache-seatunnel-2.3.12
kubernetes.container-start-command-prefix: tar -zvxf /opt/tiger/workdir/apache-seatunnel-2.3.12-all.tar.gz -C /opt/tiger/workdir

设置参数的示意如图：

如果写入 Paimon 或者 TOS 等路径可以选择额外设置以下参数，加速 TOS 写入能力：

flink.plugins.filesystem.tos.proton.enabled: true

任务启动

检查无误后，保存并发布任务

进入任务运维界面，选择全新启动/重启任务，后续更新任务的话可以从最新状态启动：

等待一个 Checkpoint 的周期，可以从 TOS 的桶文件，确认数据已经写入 Paimon：

常见问题

问题一：SeaTunnel 文件过大，上传文件管理的时候会超时。启动 Flink 容器的时候会有些慢。
主要原因：SeaTunnel 包含了大量的 Connector 文件，会大大增加
解决方案：

1. 当前 Flink 的文件管理模块不建议管理超过 500MB 以上的 JAR 包，可以参考步骤一，将 SeaTunnel 的压缩包提前上传到 TOS。
2. SeaTunnel 中 connectors 目录可以进行裁剪，将不需要的 connector 从安装包中删除，减小 tar 包体积大小。

问题二：SeaTunnel 是否支持所有的火山引擎云产品？
答案：SeaTunnel 属于开源产品，是否已经和火山引擎的云产品完成对接需要咨询相关云产品和 SeaTunnel 社区维护者。请在使用 Flink 产品前自行调研相关能力。