## 说明
### 背景
使用 Seatunnel 执行数据同步. 部署参考 部署 Apache-Seatunnel 服务
### 问题
- 需要在任务报错或者其他关键事件发送钉钉消息通知
- SeaTunnel 本身不支持消息通知, 必须依赖 DolphinScheduler 或其他外部工具
### 方案
- 利用SeaTunnel提供的事件监听器功能
- 编写自定义插件, 捕捉报错事件, 发送消息通知
- 群机器人配置通过命令行提交
### 部署
- 如果不想写代码打包, 只需要报错通知, 跳过开发插件 步骤, 下载 jar包即可
- 如果需要自定义通知内容和其他事件处理, 自行调整代码
## 开发插件
可以从 https://github.com/ts7ming/SeatunnelExt 获取(或https://gitee.com/ts7ming/SeatunnelExt)
### 项目结构
包名 `com.ts7ming` 自定义即可

```
│  pom.xml
│
└─src
    └─main
        ├─java
        │  └─com
        │      └─ts7ming
        │              DingTalkEventListener.java
        │
        └─resources
            └─META-INF
                └─services
                        org.apache.seatunnel.api.event.EventHandler
```
pom.xml
- 这里用了2.3.13 版本, 根据实际情况调整
```
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.ts7ming</groupId>
    <artifactId>SeatunnelExt</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <seatunnel.version>2.3.13</seatunnel.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.apache.seatunnel</groupId>
            <artifactId>seatunnel-api</artifactId>
            <version>${seatunnel.version}</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.seatunnel</groupId>
            <artifactId>seatunnel-engine-common</artifactId>
            <version>${seatunnel.version}</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.30</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.13.2</version>
            <scope>test</scope>
        </dependency>
    </dependencies>
</project>
```
DingTalkEventListener.java
- SeaTunnel支持的事件如下

![](https://developer.qcloudimg.com/http-save/yehe-11851421/caaa4d35ea4d55bad74160396cff6fec.jpg)


```
package com.ts7ming;

import lombok.extern.slf4j.Slf4j;
import org.apache.seatunnel.api.event.Event;
import org.apache.seatunnel.api.event.EventHandler;
import org.apache.seatunnel.api.event.EventType;
import org.apache.seatunnel.engine.common.job.JobStatus;
import org.apache.seatunnel.engine.common.job.JobStateEvent;
import org.apache.seatunnel.api.source.event.ReaderOpenEvent;
import org.apache.seatunnel.api.sink.event.WriterCloseEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableAddColumnEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableColumnsEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableDropColumnEvent;
import org.apache.seatunnel.api.table.schema.event.AlterTableModifyColumnEvent;
import javax.crypto.Mac;
import javax.crypto.spec.SecretKeySpec;
import java.io.OutputStream;
import java.net.HttpURLConnection;
import java.net.URL;
import java.net.URLEncoder;
import java.nio.charset.StandardCharsets;
import java.util.Base64;

@Slf4j
public class DingTalkEventListener implements EventHandler {
    private static final String WEBHOOK_URL = System.getProperty("dingtalk.webhook.url", "https://oapi.dingtalk.com/robot/send?access_token=YOUR_ACCESS_TOKEN");
    private static final String SECRET = System.getProperty("dingtalk.secret", "YOUR_SECRET");
    @Override
    public void handle(Event event) {
        EventType eventType = event.getEventType();
        if (eventType == EventType.JOB_STATUS) {
            handleJobStateEvent((JobStateEvent) event);
        } 
//        else if (eventType.name().equals("SCHEMA_CHANGE_ADD_COLUMN")) {
//            handleAddColumnEvent((AlterTableAddColumnEvent) event);
//        }
//        else if (eventType.name().equals("SCHEMA_CHANGE_UPDATE_COLUMNS")) {
//            handleUpdateColumnEvent((AlterTableColumnsEvent) event);
//        }
//        else if (eventType.name().equals("SCHEMA_CHANGE_DROP_COLUMN")) {
//            handleDropColumnEvent((AlterTableDropColumnEvent) event);
//        }
//        else if (eventType.name().equals("SCHEMA_CHANGE_MODIFY_COLUMN")) {
//            handleModifyColumnEvent((AlterTableModifyColumnEvent) event);
//        }
        else {
            log.debug("忽略未处理的事件类型: {}", eventType);
        }
    }

    private void handleJobStateEvent(JobStateEvent jobEvent) {
        String jobId = jobEvent.getJobId();
        String jobName = jobEvent.getJobName();
        JobStatus status = jobEvent.getJobStatus();
        long eventTime = jobEvent.getCreatedTime();

        switch (status) {
            case FAILED:
                sendAlert("【任务失败】jobId: " + jobId + ", jobName: " + jobName);
                break;
            case FINISHED:
                //sendAlert("任务完成: " + jobId + ", jobName: " + jobName);
                break;
            //其他需要处理的事件
            // case READER_OPEN:
            //     break;
            // case WRITER_CLOSE:
            //     break;
            default:
                log.debug("任务状态变更 | jobId: {}, 状态: {}, 时间: {}",jobId, status, eventTime);
                // 不发送通知
        }
    }

    private void handleAddColumnEvent(AlterTableAddColumnEvent event) {
        String tableName = event.getTableIdentifier().getTableName();
        String columnName = event.getColumn() != null ? event.getColumn().getName() : "未知列";
        sendAlert("【表结构变更】表名: " + tableName + ", 新增列: " + columnName);
    }
    private void handleUpdateColumnEvent(AlterTableColumnsEvent event) {
        String tableName = event.getTableIdentifier().getTableName();
        sendAlert("【表结构变更】表名: " + tableName + ", 更新内容: " + event);
    }
    private void handleDropColumnEvent(AlterTableDropColumnEvent event) {
        String tableName = event.getTableIdentifier().getTableName();
        String columnName = event.getColumn() != null ? event.getColumn() : "未知列";
        sendAlert("【表结构变更】表名: " + tableName + ", 删除列: " + columnName);
    }
    private void handleModifyColumnEvent(AlterTableModifyColumnEvent event) {
        String tableName = event.getTableIdentifier().getTableName();
        String columnName = event.getColumn() != null ? event.getColumn().getName() : "未知列";
        sendAlert("【表结构变更】表名: " + tableName + ", 修改列: " + columnName);
    }

    private void sendAlert(String content) {
          sendDingTalkMessage(content);
    }

    void sendDingTalkMessage(String message) {
        try {
            long timestamp = System.currentTimeMillis();
            String sign = generateSign(timestamp, SECRET);
            String fullUrl = WEBHOOK_URL + "&timestamp=" + timestamp + "&sign=" + sign;
            String escapedMessage = message.replace("\", "\\")
                                          .replace("\"", "\\"")
                                          .replace("\n", "\n")
                                          .replace("\r", "\r")
                                          .replace("\t", "\t");
            String jsonPayload = String.format("{\"msgtype\":\"text\",\"text\":{\"content\":\"%s\"}}", escapedMessage);
            URL url = new URL(fullUrl);
            HttpURLConnection conn = (HttpURLConnection) url.openConnection();
            conn.setRequestMethod("POST");
            conn.setRequestProperty("Content-Type", "application/json");
            conn.setDoOutput(true);
            conn.setConnectTimeout(5000);
            conn.setReadTimeout(5000);

            try (OutputStream os = conn.getOutputStream()) {
                os.write(jsonPayload.getBytes(StandardCharsets.UTF_8));
                os.flush();
            }
            
            int responseCode = conn.getResponseCode();
            if (responseCode == 200) {
                log.info("钉钉消息发送成功: {}", message);
            } else {
                log.error("钉钉消息发送失败，响应码: {}, 消息: {}", responseCode, message);
            }
        } catch (Exception e) {
            log.error("发送钉钉消息异常: {}", message, e);
        }
    }
    private String generateSign(long timestamp, String secret) throws Exception {
        String stringToSign = timestamp + "\n" + secret;
        Mac mac = Mac.getInstance("HmacSHA256");
        mac.init(new SecretKeySpec(secret.getBytes(StandardCharsets.UTF_8), "HmacSHA256"));
        byte[] signData = mac.doFinal(stringToSign.getBytes(StandardCharsets.UTF_8));
        return URLEncoder.encode(new String(Base64.getEncoder().encode(signData)), "UTF-8");
    }
}
```
org.apache.seatunnel.api.event.EventHandler
```
com.ts7ming.DingTalkEventListener
```
打包
```
mvn clean package
```
## 部署插件
直接下载可用 jar (如果不想自己打包)
```
cd /opt/apache-seatunnel/lib
wget https://github.com/ts7ming/SeatunnelExt/releases/download/v1/SeatunnelExt-1.0-SNAPSHOT.jar
# 网络差的话用 gitee
wget https://gitee.com/ts7ming/SeatunnelExt/releases/download/v1/SeatunnelExt-1.0-SNAPSHOT.jar

# 如果用其他用户下载, 注意权限
chown -R seatunnel:seatunnel /opt/apache-seatunnel
```
上传
- Jar包上传到Seatunnel 根目录 lib 下 (例如: /opt/apache-seatunnel/lib/)

重启 Seatunnel 服务
```
systemctl stop seatunnel-master.service
systemctl stop seatunnel-worker.service

systemctl start seatunnel-master.service
systemctl start seatunnel-worker.service
```
检查插件是否加载
```
grep "DingTalk" /opt/apache-seatunnel/logs/seatunnel-engine-master.log

# 结果应该有如下信息
INFO  [o.a.s.e.s.CoordinatorService  ] [pool-4-thread-1] - [localhost]:5801 [seatunnel] [5.1] Loaded event handlers: [com.ts7ming.DingTalkEventListener@20eaeaed, org.apache.seatunnel.api.event.LoggingEventHandler@59c99cb9]
```
## 运行任务
注意: -D参数一定要在 --config前面 (SeaTunnel启动脚本解析参数的顺序不够灵活)
```
sh bin/seatunnel.sh --async -Ddingtalk.webhook.url="https://oapi.dingtalk.com/robot/send?access_token=钉钉群Token" -Ddingtalk.secret="钉钉群secret" --config 任务配置.conf  -n "任务名称"
```
Done!

![](https://developer.qcloudimg.com/http-save/yehe-11851421/4b87757b847b92a956216cc2dbcd10e1.jpg)




使用 Seatunnel 执行数据同步. 部署参考 部署 Apache-Seatunnel 服务

Apache SeaTunnel 设置钉钉通知详细教程，亲测可用！

大数据

后端

SeaTunnel数据同步任务如何实现钉钉消息通知？无需依赖外部调度工具，通过自定义事件监听器插件自动捕获任务失败等关键事件。支持下载现成jar包快速部署，也可定制开发。插件自动监控Job状态变化，失败时立即发送告警到钉钉群，提供完整配置参数和部署步骤。实现SeaTunnel任务监控自动化，提升运维效率。

数据同步

2026采购季 | AI焕新·智启新局

datainlong

bigdata-class

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

Apache SeaTunnel 设置钉钉通知详细教程，亲测可用！-腾讯云开发者社区-腾讯云

Apache SeaTunnel 设置钉钉通知详细教程，亲测可用！

Apache SeaTunnel 设置钉钉通知详细教程，亲测可用！

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐