DolphinScheduler集成Kafka与Flink：构建定时调度与实时计算全链路

1. 项目概述：Dolphinscheduler 与 Flink 是怎么串起来的

最近在做一个订单事件流处理的项目，业务侧要求既能按时触发数据生产任务，又希望下游的实时计算任务稳定消费消息。我搭了一套基于 DolphinScheduler Standalone 单机部署的调度环境，配合 Kafka 做消息推送，再用 Flink 消费 Kafka 里的订单事件，做简单的统计清洗后落库。整套链路从调度平台到实时计算全部串通后，发现这套方案特别适合中小团队快速验证流式任务，也适合在没有完整集群资源的时候先摸清工作流调度和 Flink 消费的配合方式。

这个方案解决的核心问题其实很朴素：日常大数据开发里，定时任务、数据上报、消息生产和实时消费往往是分开维护的，要么用 crontab 硬抗，要么依赖各种自研脚本，出了问题都不知道是哪一个环节挂了。Dolphinscheduler 作为可视化的工作流调度平台，正好把“定时触发、依赖编排、运行日志、告警通知”这些脏活累活接管了；而 Flink 作为流式计算引擎，负责把 Kafka 里不断涌进来的消息消化掉。两者通过 Kafka 这个“消息管道”组合在一起，就形成了一个从定时调度到实时处理都比较完整的闭环。

这篇文章适合谁看？如果你正准备给团队搭建第一个调度平台，或者你手里还没用过 DolphinScheduler，却想让 FLink 任务有一个相对规范的入口，那直接照着我这个流程走就可以。我会把版本选型、Standalone 部署、Kafka 环境准备、Flink 消费代码、工作流编排以及我踩过的各种坑全部写出来。不需要你在这些组件上都有很深的积累，我会尽量把每一步的前因后果说清楚，哪怕你是第一次接触 DolphinScheduler，也可以跟着落地。

1.1 为什么选 Standalone 模式

DolphinScheduler 的部署模式有好几种：单机 Standalone、伪集群、Kubernetes 部署、以及基于 ZooKeeper 的高可用集群部署。不少人上来就想搞 High Availability 模式，觉得生产环境必须高可用。但我这次刻意选了 Standalone，主要有两个原因。

第一，这个项目的数据量还没到必须多活的程度。一套 Standalone 环境足以支撑几十个工作流的日常调度，即使后续流量变大，Dolphinscheduler 支持平滑迁移到集群模式，没必要一开始就把架构复杂度拉满。第二，Standalone 模式内置了 H2 数据库，不需要额外部署 MySQL、ZooKeeper，下载解压后执行一条启动命令就能用，这对快速验证业务逻辑来说效率高得多。

实际上，很多团队的生产环境也是从 Standalone 起步的。先跑通业务，再考虑控制台高可用、任务队列分片等高级能力，这个路线非常务实。而且 Standalone 模式下调度平台占用的资源很少，给开发环境、测试环境用尤其合适。

1.2 数据链路设计与选型理由

先看我最终搭出来的链路：

定时调度触发一个 Shell 节点，由 Shell 脚本调用 Kafka 生产端向指定 Topic 推送模拟订单事件；随后调度进入第二个节点，由 DolphinScheduler 向 Flink 集群提交一个流式任务；Flink 从 Kafka 中消费订单消息，做解析与统计，再把结果输出到控制台或下游存储。整个过程会把每一个节点的状态、日志、执行耗时都记录在 DolphinScheduler 上。

为什么中间要隔一个 Kafka？直接让 DolphinScheduler 调 Flink 处理数据不是更短？这里面涉及一个架构上的关键点：调度平台和流计算引擎的定位不同。Dolphinscheduler 适合管理“有始有终”的批任务和短任务，而 Flink 消费 Kafka 是持续运行的长任务。把消息先放到 Kafka，由 Kafka 做削峰填谷和消息堆积缓冲，下游 Flink 根据自己的消费能力拉取数据，两者之间就解耦了。即使 Flink 任务重启、升级，上游生产端也不需要停下来，消息不会丢。

用 Kafka 做消息推送还有一层好处：方便回溯。消费者出问题的时候，只要 Kafka 里的数据没有过期，就可以通过重置消费位点重新消费，这在联调阶段特别实用。可以说，调度平台负责“什么时候做”，Kafka 负责“消息怎么传”，Flink 负责“传过来怎么算”，三者各司其职，才是一个成熟的数据流转形态。

1.3 部署规划与预期效果

整个部署规划如下：在一台开发服务器上安装 JDK、Kafka、DolphinScheduler、Flink。DolphinScheduler 负责管理调度，Kafka 负责消息中转，Flink 作为流处理引擎独立运行。预期效果是：DolphinScheduler 控制台上能够看到完整的工作流运行历史，每次运行都能在日志里查到消息生产与消费的链路记录。

这套环境部署完成后，你可以继续往 Flink 里加各种业务逻辑，比如 CEP 复杂事件处理、窗口聚合、维表关联，DolphinScheduler 只需要负责触发就可以了。也就是说，后续扩展空间依然很大，不会因为一开始图简单就用死。

2. 环境准备：版本选型定好后成功一半

大数据组件有一个特点：版本之间兼容性问题最让人头疼。我在动手之前花了一点时间确认版本组合，这一步其实比部署本身更值得关注。组件版本如果选错，后面启动 Kafka、提交 Flink 任务时会出现各种莫名其妙的报错。

2.1 JDK 与基础组件版本建议

DolphinScheduler 3.x 系列要求 JDK 8 及以上，我使用的是 JDK 8，这也是目前大数据生态里最保守、最稳妥的选择。Kafka 我选了 2.8 版本，配套的 Scala 版本是 2.12，日常使用足够稳定。Flink 选择 1.15.2，这个版本对 Kafka 连接器的支持很成熟，社区资料也丰富。

有一点要提前说明：如果你本机之前装了比较新的 JDK，比如 JDK 17，个别 DolphinScheduler 版本可能会有兼容性问题。建议统一用 JDK 8 来跑调度平台，省去后续排查时区、加密套件之类杂七杂八的问题。

2.2 版本兼容性对照表

这里我把这次使用的组件版本和用途整理成一张表，大家可以参考：

Flink 和 Kafka 连接器的版本尤其要注意：Flink 1.15 对应 flink-connector-kafka 1.15.2。不能随便拿一个旧版 Kafka Connector 塞给 Flink 1.15 用，否则会抛序列化或者类加载异常。

2.3 下载与目录规划

我习惯把所有大数据组件统一放在 /data 目录下，每个组件一个独立文件夹，日志统一走组件自己的 log 目录。这样后续查找日志、扩容、备份都比较方便。

下载时建议通过各组件官网的镜像站获取。DolphinScheduler 下载二进制包后，解压到 /data/dolphinscheduler；Kafka 解压到 /data/kafka；Flink 解压到 /data/flink。全部准备好后，先配置环境变量，把 JAVA_HOME 固定指向 JDK 8。

这里有一个小建议：JAVA_HOME 不要依赖 /etc/profile 里可能被覆盖的默认配置，而是在 /etc/profile.d/env.sh 里单独写死：

export JAVA_HOME=/data/jdk1.8.0_202 export PATH=$JAVA_HOME/bin:$PATH

写完后 source 一下，再用 java -version 确认一下。这一步虽然简单，但能避免后面好几个组件同时抢默认 JDK 的问题。

3. DolphinScheduler Standalone 单机部署实操

3.1 初始化配置

DolphinScheduler 3.2.1 的 Standalone 模式设计得很轻量，默认配置已经能直接跑起来。解压后的目录结构里，bin/start.sh 就是启动脚本。不过，我还是建议先看一眼 conf/application.yaml，确认服务端口和数据存储方式。Standalone 默认监听 12345 端口，数据库使用内置 H2。

如果只想快速体验，完全不需要改配置。但如果你想把调度平台暴露给其他同事访问，要注意服务器防火墙的端口放行，比如云服务器安全组需要放行 12345 端口。这一点和组件本身无关，但最容易让人误以为服务启动失败。

启动时执行：

cd /data/dolphinscheduler/ds bin/start.sh

启动完成后访问：http://服务器IP:12345/dolphinscheduler/ui，默认账号 admin，密码 dolphinscheduler123。第一次登录会要求修改密码，这个地方别跳过，否则默认密码太容易被猜到。

3.2 启动过程与日志排查

如果你发现访问不了页面，第一件事不是改配置，而是看日志。Standalone 模式的日志在 logs/ 目录下，重点看 dolphinscheduler-standalone-server.log 或类似命名的文件。正常情况下启动会经过数据库初始化、服务注册、API 服务拉起几个阶段，耗时一般在 20 秒到 60 秒之间。

我在第一次启动时遇到过一个问题：进程起来了，但页面一直白屏。查看日志才发现是端口被占用。因为本机之前有个服务占用了 12345 端口，DolphinScheduler 的 API 服务没起来，只有前端静态资源服务起来了。换端口之后一切正常。所以各位如果遇到启动后页面白屏或接口报错，先排查端口占用。

另外多说一句，Standalone 模式虽然内置了 H2 数据库，但数据只能存在本地文件里。如果后续要引入正式环境，建议切换成 MySQL，并部署三节点高可用模式。初学阶段用 H2 完全没有问题。

3.3 队列、租户与用户权限配置

DolphinScheduler 里有一个概念和常见的调度平台不太一样，就是“租户”。工作流执行任务时，必须指定一个租户。租户实际上对应 Linux 系统里的一个用户。

所以你需要先在服务器上创建一个 Linux 用户，比如叫 ds：

useradd ds

接着登录 DolphinScheduler 控制台，在“安全中心 -> 租户管理”里创建一个租户，租户名称填 ds，队列填 root.default。再进入“用户管理”，把 admin 用户绑定到这个租户上。不做这一步，你创建的工作流提交执行时会直接报“租户不存在”或“执行用户不存在”。

队列的概念也和 Yarn 队列相关，如果你没有独立 Yarn 集群，使用默认队列即可。但如果你后续接入了 Hadoop/Yarn，建议在“队列管理”里单独创建真实队列，比如 root.ds_dev，这样任务资源可以隔离，不至于全挤在 default 队列里。

4. Kafka 环境准备与消息通道验证

4.1 启动 Kafka 服务

Kafka 2.8 版本还支持基于 ZooKeeper 的模式，单机环境直接用自带的 ZooKeeper 就行。先启动 ZooKeeper：

cd /data/kafka bin/zookeeper-server-start.sh config/zookeeper.properties &

然后启动 Kafka Broker：

bin/kafka-server-start.sh config/server.properties &

这里需要注意 Kafka 的 listeners 和 advertised.listeners 配置。默认情况下 Kafka 只监听 localhost:9092，如果你用服务器 IP 从另一个机器访问 Producer/Consumer 会失败。单机测试倒无所谓，但如果你计划从本机的 Flink 所在容器或远程调试工具连接，最好把 advertised.listeners 改成服务器内网 IP。

修改 config/server.properties：

listeners=PLAINTEXT://0.0.0.0:9092 advertised.listeners=PLAINTEXT://你的服务器内网IP:9092

改完重启 Kafka。这个配置我踩过坑，如果不改，本机使用 localhost 连接一切正常，换 IP 就超时。

4.2 创建 Topic 与生产消费验证

创建 Topic 的命令如下：

bin/kafka-topics.sh --create \ --bootstrap-server localhost:9092 \ --replication-factor 1 \ --partitions 3 \ --topic ods_order_event

Topic 名称我习惯用“ods_”开头，表示 ODS 层原始数据。分区数我设置了 3，这个数字不是随便写的：分区数直接决定了 Kafka 的并行度上限，也影响下游 Flink 的并行度设置。单机环境下分区数 3 已经能提供足够的并发，同时不会产生过多的文件句柄开销。

创建完成后用 producer 和 consumer 验证一下消息通道：

# 终端1 启动生产者 bin/kafka-console-producer.sh \ --bootstrap-server localhost:9092 \ --topic ods_order_event # 终端2 启动消费者 bin/kafka-console-consumer.sh \ --bootstrap-server localhost:9092 \ --topic ods_order_event \ --from-beginning

如果生产者输入消息后，消费者终端能看到，说明 Kafka 链路是通的。这一步非常重要，千万别跳过。我见过不少人在 Flink 连不上 Kafka 后到处排查，回头一看 Kafka 自己都接收不了消息。

4.3 Group 与 Offset 的基础认知

在进入 Flink 之前，还要补充一个关键概念：Consumer Group。Kafka 消费消息时，同一个 Group 里的消费者共同分摊分区，不同 Group 之间互不影响。Flink 作为消费者时也要指定 Group ID，这个 Group ID 会决定消费位点的保存位置。

日常开发里，如果有多个 Flink 作业想消费同一个 Topic，且各自需要从头消费，它们必须使用不同的 Group ID。如果两个作业共用一个 Group ID，Kafka 会把分区分配给其中一个消费者，另一个作业可能完全收不到数据。这个现象在联调阶段非常常见，别等出了问题再来困惑。

5. Flink 消费 Kafka 的实现与提交细节

5.1 工程依赖与打包方式

Flink 消费 Kafka 的代码并不复杂，但依赖管理是有讲究的。我用 Maven 创建一个 Java 工程，依赖如下：

<properties> <maven.compiler.source>1.8</maven.compiler.source> <maven.compiler.target>1.8</maven.compiler.target> </properties> <dependencies> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-java</artifactId> <version>1.15.2</version> <scope>provided</scope> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-kafka</artifactId> <version>1.15.2</version> </dependency> </dependencies> <build> <plugins> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-shade-plugin</artifactId> <version>3.4.1</version> <executions> <execution> <phase>package</phase> <goals> <goal>shade</goal> </goals> </execution> </executions> </plugin> </plugins> </build>

这里的关键是把 flink-streaming-java 设为 provided，因为 Flink 运行环境自带这些核心类；而 flink-connector-kafka 必须打包进最终 Jar，否则提交到 Flink 集群后会报找不到 KafkaSource 相关类的错误。maven-shade-plugin 会把连接器的依赖一起打进去。

5.2 核心消费逻辑代码

我写了一个最简单的 Demo：消费订单事件，按订单类型做计数统计。消息格式假设是“时间戳,订单类型,金额”，比如：

1720000000000,normal,99.00 1720000001000,vip,199.00

代码核心逻辑如下：

package com.example; import org.apache.flink.api.common.eventtime.WatermarkStrategy; import org.apache.flink.api.common.serialization.SimpleStringSchema; import org.apache.flink.api.common.typeinfo.Types; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.connector.kafka.source.KafkaSource; import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.util.Collector; public class KafkaFlinkDemo { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.enableCheckpointing(30000L); KafkaSource<String> source = KafkaSource.<String>builder() .setBootstrapServers("localhost:9092") .setTopics("ods_order_event") .setGroupId("flink_consumer_group") .setStartingOffsets(OffsetsInitializer.latest()) .setValueOnlyDeserializer(new SimpleStringSchema()) .build(); DataStream<String> stream = env.fromSource( source, WatermarkStrategy.noWatermarks(), "kafka-order-source" ); stream.flatMap((String line, Collector<Tuple2<String, Integer>> out) -> { String[] fields = line.split(","); if (fields.length >= 2) { out.collect(Tuple2.of(fields[1].trim(), 1)); } }).returns(Types.TUPLE(Types.STRING, Types.INT)) .keyBy(item -> item.f0) .sum(1) .print(); env.execute("kafka-flink-demo"); } }

在写这段逻辑时，我特意把 enableCheckpointing 打开了。这个开关非常重要。Flink 只有开启了 Checkpoint，才会定期把 Kafka 消费位点保存到状态后端，作业重启后才能从最近的位点继续消费。如果不开启，Flink 作业每次重启都可能出现重复消费，消息量大的时候统计结果会偏。

5.3 提交方式与参数选择

打包完成后，得到类似 dolphinscheduler-flink-demo-1.0-SNAPSHOT.jar 的文件。有两种常见提交方式：一种是直接在本机用 flink run 命令以 Standalone 模式运行，另一种是提交到 Yarn 集群。

由于我这次没有把 Flink 跑在完整 Hadoop Yarn 上，所以直接用了 Flink 默认的 Standalone 模式。先把 Flink 集群启动起来：

cd /data/flink/flink-1.15.2 bin/start-cluster.sh

然后提交作业：

bin/flink run \ -m localhost:8081 \ -c com.example.KafkaFlinkDemo \ /data/jars/dolphinscheduler-flink-demo-1.0-SNAPSHOT.jar

这里 -m 指的是 JobManager 地址。默认端口是 8081。可以看到，Flink 本身也有一个 Web UI，从这个界面可以观察作业的吞吐量、延迟和 Checkpoint 状态。如果你的服务器内存有限，可以在提交前调整一下 TaskManager 的内存，稍后我在常见问题里详细说。

6. 工作流编排：把推送和消费链路自动化起来

到这里，各个组件都已经就绪。接下来要解决的问题是：如何用 DolphinScheduler 把“定时推送消息到 Kafka”和“启动 Flink 消费任务”编排成一个可以重复执行的工作流。

6.1 在 DolphinScheduler 中创建项目与资源

第一步先把 Flink 的 Jar 包上传到 DolphinScheduler 的资源中心。进入“资源中心 -> 文件管理”，点击上传文件，把刚才打好的 Jar 包传上去。资源中心的作用是统一管理脚本、Jar 包等文件，工作流节点执行时可以引用这些资源。

为什么要把 Jar 放到资源中心，而不是直接写死绝对路径？因为 DolphinScheduler 支持多 Worker 节点，如果以后扩展到多台机器，每台机器上的工作目录可能不一样。把 Jar 放到资源中心后，DolphinScheduler 会把这个文件分发到实际执行任务的 Worker 上，代码里不需要关心具体路径。这个设计对于后期扩容非常友好。

接着创建一个 Shell 脚本资源，比如叫 kafka_push.sh，内容如下：

#!/bin/bash msg="$(date +%s),normal,99.00" echo "$msg" | /data/kafka/bin/kafka-console-producer.sh \ --bootstrap-server localhost:9092 \ --topic ods_order_event

这个脚本的作用是向 Kafka 推送一条测试消息。实际项目中，你可以把消息内容替换成业务系统上报的真实数据，也可以把脚本换成 Java Producer 程序。

6.2 创建项目与工作流定义

回到 DolphinScheduler 首页，进入“项目 -> 项目列表”，创建一个新的项目，比如叫“kafka_flink_demo”。进入项目后，在“工作流定义”页面点击“创建工作流”，拖拽出两个 Shell 节点。

第一个节点叫“push_kafka”，运行参数指向资源 kafka_push.sh，并勾选该资源。第二个节点叫“start_flink_job”，它是整个链路的消费端，也是一个 Shell 节点，脚本内容如下：

#!/bin/bash export FLINK_HOME=/data/flink/flink-1.15.2 export PATH=$PATH:$FLINK_HOME/bin flink run \ -m localhost:8081 \ -c com.example.KafkaFlinkDemo \ ${file}/dolphinscheduler-flink-demo-1.0-SNAPSHOT.jar

这里的 ${file} 是 DolphinScheduler 的资源引用占位符。当你在节点的“资源”属性里勾选了上传的 Jar 包后，运行时这个占位符会被替换成 Worker 节点上的实际文件路径。我个人建议把资源勾选和脚本里的 ${file} 配套使用，这样最稳妥，不会出现找不到 Jar 的问题。

然后在这两个节点之间画一条依赖连线，让第二个节点依赖第一个节点成功执行。保存工作流后，点击“上线”按钮，再选择“运行一次”，就可以触发一次手动执行。

6.3 执行日志怎么看

工作流运行后，可以在“工作流实例”页面看到实例状态。点击实例名称可以进入 DAG 页面，每个节点都有独立的日志入口。这个日志非常详细，包含了 Shell 脚本的执行输出、错误堆栈以及 DolphinScheduler 的工具函数输出。

看日志时我习惯先看当前节点是成功还是失败。如果失败，直接展开节点的日志，搜关键字 Exception 或 ERROR。举个真实例子：我第一次运行 Flink 节点时，日志里报“Caused by: java.lang.ClassNotFoundException: org.apache.flink.connector.kafka.source.KafkaSource”。原因就是打包时没有把 flink-connector-kafka 依赖打进去。重新用 shade 插件打包后问题解决。

另一个常见问题是 Shell 脚本执行完但 Flink 作业还在后台跑。DolphinScheduler 在执行 Shell 节点时，等待的是 Shell 进程退出。如果你用 flink run 默认前台模式，Shell 会一直等待 Flink 作业结束，节点状态也会一直处于运行中。这其实是正常的，但如果你希望 Shell 节点只在“提交”这个动作结束后就退出，可以在 flink run 后面加 -d 参数，让作业 detach 到后台。这一点要根据实际业务权衡。

6.4 定时调度与长任务边界

工作流上线后，可以设置定时调度。比如每 5 分钟执行一次，Cron 表达式为：

0 0/5 * * * ?

如果每次调度都重新提交一个 Flink 作业，显然不合理，因为 Flink 作业是长驻的。我在这个项目里的实践方式是：工作流只负责定时推送 Kafka 消息，Flink 作业由第一次调度时启动后常驻运行。具体做法是把“start_flink_job”节点设置为仅在第一次运行时执行，或者干脆把启动 Flink 和定时推送分成两个工作流。

为什么这么设计？因为 DolphinScheduler 擅长管理短生命周期任务，而 Flink 的流式作业是常驻进程。强行让调度平台去启动和停止常驻进程，会造成作业频繁重启，反而影响稳定性和 Checkpoint 恢复。正确的姿势是：调度平台负责数据产生侧的定时节奏，Flink 消费侧保持独立运行。这是我在项目里最有价值的一个架构判断。

7. 常见问题与排查技巧实录

7.1 版本不一致引发的运行异常

这类问题最隐蔽。比如 Flink 1.15 作业报：

java.lang.NoSuchMethodError: org.apache.kafka.clients.consumer.KafkaConsumer.subscribe

这种一般不是代码问题，而是 Jar 包里 kafka-clients 版本和运行环境冲突。解决方案是统一 Flink Connector 版本，使用 Maven 的 dependencyManagement 锁版本，避免传递依赖把不兼容的 Kafka 客户端带进来。

还有 Flink 本身的依赖冲突，如果用户 Jar 里打入了 flink-streaming-java，提交到 Flink 集群时有可能出现类加载冲突。用 provided 声明核心依赖就能规避。

7.2 Kafka 消费不到数据怎么办

按下面的顺序排查：

1. 先看 Kafka 控制台消费者，手动发一条消息验证 Kafka 本身是否正常。
2. 用 Flink Web UI 查看作业是否处于 Running 状态，检查 Source 算子的“Records Received”指标是否一直为 0。
3. 检查 Flink 作业的 Group ID 是否和之前测试的消费者共用了同一个 Group，如果是，把 Group ID 改一个。
4. 检查 StartingOffsets，如果设置成 latest，而消息是几分钟前生产的，Flink 启动时已经错过了这些消息，所以看不到。调试阶段用 earliest 更合适。
5. 确认 bootstrap.servers 是否用了服务器内网 IP，Kafka 配置 advertised.listeners 是否匹配。

7.3 Flink 作业提交成功但 DAG 不显示

如果你在 Flink Web UI 里找不到刚刚提交的作业，大概率是提交模式的问题。默认 Standalone 模式下，作业信息在 JobManager 的 UI 里。如果 flink run 命令返回了作业 ID，但 UI 看不到，可能是 -m 参数指定的地址不对。

排查方法：直接在命令行执行 jps，看有没有 TaskManagerExecutor 进程。然后访问 JobManager 的 REST API：

curl http://localhost:8081/jobs

能看到作业列表，说明服务正常，只是 UI 访问入口不对。看不到就需要查 Flink 日志。这类问题多花几分钟查一下日志，比自己瞎猜高效得多。

7.4 调度节点一直“运行中”不结束

这个问题很多新手会遇到。Shell 节点里执行了 flink run 命令，没有加 -d 参数，Flink 作业在前台运行，Shell 就一直等待，因此节点状态一直是“运行中”。

解决办法有两个：一是给 flink run 加 -d 参数，让作业分离到后台；二是把 Flink 提交逻辑单独放到工作流里，只触发一次，不要每次调度都执行。如果 Flink 作业本身就是短期的测试，等待它执行完也无可厚非。关键是你得清楚当前节点的行为逻辑，别对“运行中”状态产生误会。

7.5 资源不足导致 Flink 作业频繁失败

Standalone 模式下，Flink 默认会启动固定数量的 TaskManager，每个 TaskManager 的内存默认值可能超过 1G 甚至更多。如果服务器内存只有 4G，同时还要跑 Kafka 和 DolphinScheduler，任务很容易因为内存不足失败。

遇到这种情况，优先修改 Flink 的配置 conf/flink-conf.yaml：

jobmanager.memory.process.size: 1024m taskmanager.memory.process.size: 1024m taskmanager.numberOfTaskSlots: 4

注意，修改配置后需要重启 Flink 集群和已经提交的作业。DolphinScheduler 本身和它执行 Shell 脚本的 Worker 进程也会占内存，尽量给 Flink 留出足够空间。实测下来，4G 内存的机器跑这套环境会比较紧张，8G 就比较舒服了。

我在实际项目中的体会是：这套方案最大的价值不在于用什么高级特性，而在于把“调度”和“实时计算”两个本来看起来毫不相干的部分无缝衔接起来。DolphinScheduler 本身学习曲线不算陡，Standalone 模式更是可以在十分钟内完成体验部署。如果你只是想把 Flink 任务纳入规范化的调度体系，或者想找一个能定时推送数据到 Kafka 再触发流计算的可视化入口，这篇文章的流程可以直接抄。最后再分享一个小技巧：把 Kafka Topic 的生命周期管理也放到 DolphinScheduler 的 Shell 节点里，定时检查 Topic 是否存在、分区数是否足够，可以在上层就把很多低级问题拦截掉，让实时链路稳定很多。