数据工程大数据批处理流处理【免费下载链接】seatunnelSeaTunnel is a next-generation super high-performance, distributed, massive data integration tool.项目地址https://gitcode.com/gh_mirrors/sea/seatunnel点击查看免费下载SeaTunnel 除了内置的 SeaTunnel Engine 之外还提供了完整的 Flink 运行时适配层允许你复用 Apache Flink 这一高性能分布式流处理引擎来承载 SeaTunnel 作业。本文围绕官方文档 docs/en/other-engine/flink.md 展开结合仓库中seatunnel-flink-starter的源码实现系统讲解如何在 SeaTunnel 配置文件中注入 Flink 原生参数flink.前缀机制、如何编写一个从 FakeSource 到 Console 的完整 Flink 作业以及如何通过示例工程和命令行两种方式将作业跑起来。读完本文你将掌握 SeaTunnel-on-Flink 的核心配置规则、参数类型限制、废弃参数迁移方式与完整的启动链路。Flink 引擎在 SeaTunnel 中的定位Flink 是一款高性能、分布式的流处理引擎具备精确一次Exactly-Once语义、原生流批一体、丰富的 State/Checkpoint 能力。SeaTunnel 将 Flink 视为可选的计算引擎之一作业依然用 SeaTunnel 的 HOCON 配置来描述env/source/transform/sink但运行时由 Flink 的StreamExecutionEnvironment与StreamTableEnvironment来调度执行。在仓库中这一能力由seatunnel-core/seatunnel-flink-starter模块提供它包含三个子模块seatunnel-flink-13-starter面向 Flink 1.13 的启动器seatunnel-flink-15-starter面向 Flink 1.15 的启动器seatunnel-flink-starter-common两个版本共享的公共实现包括参数解析FlinkCommandArgs.java、环境构建AbstractFlinkRuntimeEnvironment.java与 Flink 配置注入逻辑EnvironmentUtil.java。在作业配置中注入 Flink 原生参数flink.前缀机制SeaTunnel 配置文件中env块内的参数分为两类一类是 SeaTunnel 的通用参数如parallelism、checkpoint.interval另一类是希望直接透传给 Flink 的原生配置项。凡是 Flink 原生参数键名必须以flink.开头SeaTunnel 启动器会把前缀剥离后写入 Flink 的Configuration最终生效于StreamExecutionEnvironment。例如为作业开启非对齐检查点Unaligned Checkpointenv { parallelism 1 flink.execution.checkpointing.unaligned.enabled true }这里flink.execution.checkpointing.unaligned.enabled被剥掉flink.前缀后等价于 Flink 配置项execution.checkpointing.unaligned.enabled对应 Flink 官方配置ExecutionCheckpointingOptions.ENABLE_UNALIGNED。源码实现前缀剥离与过滤逻辑这一注入过程由 EnvironmentUtil.java 的initConfiguration方法完成遍历env配置的全部键值对只处理以flink.开头的键通过confKey.replaceFirst(flink., )去掉前缀后调用configuration.setString(...)写入 Flink 的Configuration键值统一以字符串形式写入最终由 Flink 自己按目标配置项的类型完成解析。需要特别注意的是initConfiguration会跳过以flink.table.exec开头的键——这类表执行参数由另一个方法initTableEnvironmentConfigurationEnvironmentUtil.java单独处理它把flink.table.exec.*还原为table.exec.*后写入TableEnvironment的Configuration从而控制 Flink Table API 的执行行为。也就是说flink.前缀同时打通了 DataStream 层与 Table 层两套配置体系。参数值类型限制枚举需在 Flink 配置文件中指定当前版本的flink.前缀透传仅支持 Integer / Boolean / String / Duration 四类取值枚举类型如检查点模式、状态后端枚举等暂不支持在作业配置中直接书写需要放到 Flink 的flink-conf.yaml中配置。这是当前实现上的明确边界规划参数时应提前区分可以用flink.前缀注入数值、布尔值、普通字符串、时长类配置应写入 Flink 配置文件execution.checkpointing.mode等枚举类配置。从源码看透传时所有值都会被unwrapped().toString()转换为字符串EnvironmentUtil.java因此字符串与 Duration 类参数可以安全透传而枚举语义的解析完全交给 Flink 侧。更多可透传的 Flink 原生参数示例仓库中的 Flink starter 测试配置 test_flink_run_parameter.conf 给出了多个实际可用的flink.前缀参数示例env { parallelism 1 flink.execution.checkpointing.interval 5000 flink.execution.checkpointing.unaligned.enabled true flink.execution.checkpointing.aligned-checkpoint-timeout 100000 flink.jobstore.cache-size 52428801 flink.state.backend.rocksdb.predefined-options SPINNING_DISK_OPTIMIZED_HIGH_MEM }它们分别对应检查点间隔、非对齐检查点开关、对齐检查点超时、JobStore 缓存大小与 RocksDB 状态后端预设选项。参照 Flink 官方配置项列表几乎所有flink-conf.yaml中的键都可以用这种方式按作业粒度覆盖无需修改集群级配置。env 中的 pipeline 参数除flink.前缀外env中还支持pipeline.jars与pipeline.classpaths它们会被写入 Flink 的PipelineOptions.JARS/PipelineOptions.CLASSPATHS见 EnvironmentUtil.java用于在作业层面声明额外的依赖 Jar 与 classpath。通用参数、Flink 原生参数与废弃参数的取舍env块内同时存在几套风格不同的参数理解它们的优先级与迁移关系有助于写出干净、可维护的作业配置。通用参数与 Flink 原生参数并存以检查点为例AbstractFlinkRuntimeEnvironment.java 的setCheckpoint方法展示了参数读取顺序首选 SeaTunnel 公共参数checkpoint.interval即EnvCommonOptions.CHECKPOINT_INTERVAL否则回退到废弃键execution.checkpoint.interval最后兜底默认值10000L10 秒checkpoint.timeout对应execution.checkpoint.timeout废弃别名废弃键execution.checkpoint.mode支持exactly-once与at-least-once两个取值。因此官方文档示例中这样的写法是标准用法——公共参数与flink.前缀原生参数各司其职env { # 通用参数 parallelism 1 checkpoint.interval 5000 # flink 特殊参数 flink.execution.checkpointing.mode EXACTLY_ONCE flink.execution.checkpointing.timeout 600000 }其中checkpoint.interval 5000由 SeaTunnel 侧解析并调用environment.enableCheckpointing(interval)而flink.execution.checkpointing.mode、flink.execution.checkpointing.timeout属于 Flink 原生配置通过前缀机制透传。两者面向的是 Flink 检查点体系的不同入口可以共存。旧版execution.*前缀参数已废弃ConfigKeyName.java 中集中声明了一批带Deprecated注解的旧键名包括但不限于execution.parallelism、execution.max-parallelismexecution.time-characteristic、execution.buffer.timeoutexecution.checkpoint.interval、execution.checkpoint.mode、execution.checkpoint.timeout、execution.checkpoint.data-uriexecution.max-concurrent-checkpoints、execution.checkpoint.cleanup-mode、execution.checkpoint.min-pause、execution.checkpoint.fail-on-errorexecution.restart.strategy、execution.restart.attempts、execution.restart.delayBetweenAttempts、execution.restart.failureInterval、execution.restart.failureRate、execution.restart.delayIntervalexecution.query.state.max-retention、execution.query.state.min-retention、execution.state.backend这些键目前仍被兼容运行时会打印“will be deprecated, please use the flink. prefix…”的告警日志见 EnvironmentUtil.java但官方推荐将同类能力迁移到两种写法之一公共参数如parallelism、checkpoint.interval或flink.前缀原生参数如flink.execution.checkpointing.interval。重启策略与状态后端等进阶 env 参数从 AbstractFlinkRuntimeEnvironment.java 与 EnvironmentUtil.java 可以看到 SeaTunnel 对 Flink 环境的封装还包含重启策略execution.restart.strategy支持no、fixed-delay需配合execution.restart.attempts与execution.restart.delayBetweenAttempts、failure-rate需配合execution.restart.failureInterval、execution.restart.failureRate、execution.restart.delayInterval并且配置校验逻辑checkRestartStrategy会在缺参时直接返回错误并行度优先读公共参数parallelism其次兼容废弃键execution.parallelism时间特征execution.time-characteristic支持event-time、ingestion-time、processing-time状态后端通过execution.checkpoint.data-uri指定存储路径配合execution.state.backend rocksdb时自动构造RocksDBStateBackend(FsStateBackend)作业模式当job.mode BATCH时环境被设置为RuntimeExecutionMode.BATCH且检查点会被跳过Flink 在批模式下无需检查点。编写一个简单的 Flink 作业FakeSource 到 Console下面是在 Flink 上运行的最简作业由 FakeSource 随机生成 16 行数据并打印到控制台。该示例完整覆盖了env/source/transform/sink四个区块schema 中展示了 SeaTunnel 支持的全部常见数据类型含嵌套row与map/arrayenv { # 通用参数 parallelism 1 checkpoint.interval 5000 # flink 特殊参数 flink.execution.checkpointing.mode EXACTLY_ONCE flink.execution.checkpointing.timeout 600000 } source { FakeSource { row.num 16 result_table_name fake_table schema { fields { c_map mapstring, string c_array arrayint c_string string c_boolean boolean c_int int c_bigint bigint c_double double c_bytes bytes c_date date c_decimal decimal(33, 18) c_timestamp timestamp c_row { c_map mapstring, string c_array arrayint c_string string c_boolean boolean c_int int c_bigint bigint c_double double c_bytes bytes c_date date c_decimal decimal(33, 18) c_timestamp timestamp } } } } } transform { # 如需了解 transform 插件的完整列表与配置方式 # 可参阅仓库 docs/zh/transform-v2 或 docs/en/transform-v2 目录 } sink { Console {} }运行后Flink 会按parallelism 1并行度执行每 5 秒触发一次检查点以 Exactly-Once 模式处理 FakeSource 产出的 16 行结构化数据并逐行输出到控制台。result_table_name会把源表注册为 Flink Table 环境中的临时视图供后续 transform / sink 以 SQL 语义引用。在工程内直接运行作业SeaTunnelApiExample如果你已经将仓库代码拉取到本地最快的方式是直接运行示例模块seatunnel-examples/seatunnel-flink-connector-v2-example中的org.apache.seatunnel.example.flink.v2.SeaTunnelApiExample主类来完成作业的启动。该示例类位于 SeaTunnelApiExample.java核心逻辑是默认读取 classpath 下的/examples/fake_to_console.conf也支持通过命令行第一个参数指定其他配置文件路径构造FlinkCommandArgs设置配置文件路径、checkConfig false、variables null调用SeaTunnel.run(flinkCommandArgs.buildCommand())直接在当前 JVM 中提交执行。示例配套的作业配置为 fake_to_console.conf内容与上面的示例基本一致job.mode BATCH、parallelism 2、FakeSource 产出name/age两字段并输出到 Console。运行方式# 在项目根目录下编译并运行示例 ./mvnw -pl seatunnel-examples/seatunnel-flink-connector-v2-example -am compile exec:java \ -Dexec.mainClassorg.apache.seatunnel.example.flink.v2.SeaTunnelApiExample这种方式适合本地开发调试作业直接以内嵌 Flink 环境运行无需预先部署独立的 Flink 集群。通过命令行提交作业Starter 脚本与部署模式在真实生产环境中作业通常提交到独立的 Flink 集群。SeaTunnel 的 Flink Starter 采用“先生成 flink 提交命令再交由${FLINK_HOME}/bin/flink执行”的架构。命令生成链路启动脚本 start-seatunnel-flink-15-connector-v2.shFlink 1.13 对应 start-seatunnel-flink-13-connector-v2.sh的执行流程是以org.apache.seatunnel.core.starter.flink.FlinkStarter为主类启动把用户传入的全部参数转发给它FlinkStarter.main解析参数并调用buildCommands()拼出最终的 flink 命令字符串后打印到标准输出脚本捕获输出若退出码为 234 则打印帮助为 0 则执行最后一行命令即真正的${FLINK_HOME}/bin/flink ...提交指令。从 FlinkStarter.java 的buildCommands可以看到生成的命令结构${FLINK_HOME}/bin/flink \ deploy-mode \ [--target master] \ [原始 flink 参数...] \ -c org.apache.seatunnel.core.starter.flink.SeaTunnelFlink \ starter.jar \ --config 配置文件路径 \ [--check] \ --name 作业名 \ [--encrypt | --decrypt] \ [-i 变量替换...]作业真正的主类是org.apache.seatunnel.core.starter.flink.SeaTunnelFlinkSeaTunnelFlink.java它解析FlinkCommandArgs后通过SeaTunnel.run(...)执行作业。若用户在命令行通过--name指定了作业名还会覆盖配置文件env块中的job.name见 FlinkTaskExecuteCommand.java。支持的部署模式与提交目标根据 FlinkCommandArgs.java 中的参数定义SeaTunnel-on-Flink 支持部署模式-e/--deploy-moderun、run-application提交目标--master/--targetlocal、remote、yarn-session、yarn-per-job、kubernetes-session、yarn-application、kubernetes-application。例如提交到本地集群./bin/start-seatunnel-flink-15-connector-v2.sh \ --deploy-mode run \ --target local \ --config /path/to/your-job.conf提交到 YARN Session 集群./bin/start-seatunnel-flink-15-connector-v2.sh \ -e run --master yarn-session \ -m yarn-cluster \ --config /path/to/your-job.conf参数解析器还内置了校验当提交目标或部署模式超出上述枚举时会直接抛出IllegalArgumentException并提示合法取值FlinkCommandArgs.java避免无效命令进入 Flink 侧。参数注入链路与注意事项小结综合来看一条 SeaTunnel-on-Flink 作业的配置会经历如下注入链路FlinkTaskExecuteCommand读取配置文件并解析为Config命令行--name覆盖env.job.nameFlinkRuntimeEnvironment.prepare()依次调用createStreamEnvironment()与createStreamTableEnvironment()EnvironmentUtil.initConfiguration将flink.*排除flink.table.exec.*写入 Stream 环境配置initTableEnvironmentConfiguration将flink.table.exec.*写入 Table 环境配置公共参数parallelism、checkpoint.interval等与废弃的execution.*键由AbstractFlinkRuntimeEnvironment直接映射到 Flink APISource / Transform / Sink 插件由 Flink 执行处理器SourceExecuteProcessor、TransformExecuteProcessor、SinkExecuteProcessor逐段编译为 DataStream 算子链执行。使用中需要留意的几个事实性细节枚举类 Flink 配置请写入 Flink 的flink-conf.yaml作业配置中的flink.前缀透传目前只覆盖 Integer / Boolean / String / Duration检查点默认间隔为 10 秒未配置checkpoint.interval时的兜底值job.mode BATCH时会跳过检查点这是 Flink 批模式下的正常行为不是故障旧版execution.*参数仍可用但已标记废弃运行时会打印迁移提示建议逐步迁移到公共参数或flink.前缀写法作业级参数通过flink.前缀按作业粒度生效适合多租户场景下对不同作业差异化调优无需改动集群级flink-conf.yaml。围绕 Flink 引擎的更多部署细节可参考官方文档 docs/en/other-engine/flink.md 及其中文版 docs/zh/other-engine/flink.mdSpark 引擎的对应说明见 docs/en/other-engine/spark.md。赞分享数据工程大数据批处理流处理【免费下载链接】seatunnelSeaTunnel is a next-generation super high-performance, distributed, massive data integration tool.项目地址https://gitcode.com/gh_mirrors/sea/seatunnel点击查看免费下载相关推荐SeaTunnel 基于 Flink 引擎的本地快速上手部署、配置与运行实战SeaTunnel 基于 Flink 引擎的本地快速上手部署、配置与运行实战 导读 本指南面向已经拥有或计划使用 Apache Flink 运行环境的团队讲数据集成ETL大数据批处理流处理变更数据捕获Apache SeaTunnel 基于 Flink 引擎的快速入门指南Apache SeaTunnel 基于 Flink 引擎的快速入门指南 前言 Apache SeaTunnel 是一个高性能、分布式、易扩展的数据集成平台支持数据集成ETL大数据批处理流处理变更数据捕获Apache SeaTunnel 基于 Flink 快速上手指南部署、配置与运行数据同步作业Apache SeaTunnel 基于 Flink 快速上手指南部署、配置与运行数据同步作业 本篇指南面向希望在 Apache Flink https://l数据工程大数据批处理流处理上一篇Kubernetes Python 客户端 V1ContainerStatus 模型详解解读 Pod 容器运行状态与就绪探针下一篇DataHub Agents 实战指南在元数据图上构建、调度并治理你的 AI Agent创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考