SeaTunnel Email Sink 连接器详解：将数据以附件形式发送邮件-北京尧图网络

SeaTunnel Email Sink 连接器详解将数据以附件形式发送邮件【免费下载链接】seatunnelSeaTunnel is a multimodal, high-performance, distributed, massive data integration tool.项目地址: https://gitcode.com/GitHub_Trending/se/seatunnel本文以 SeaTunnel 开源仓库中的 Email Sink 官方文档为主线结合connector-email模块的源码实现系统讲解 Email Sink 连接器的能力边界、全部配置参数、底层发送机制与完整实战示例。阅读完本文后你将掌握如何在 SeaTunnel 作业中把任意数据源如 FakeSource、Kafka、JDBC 等的查询结果缓冲为 CSV 附件并通过 SMTP 发送到一个或多个收件人邮箱同时理解其空表不发送邮件、多表独立发信等设计细节可直接照搬到真实业务场景报表推送、告警通知、数据分发。连接器概述Email Sink 是 SeaTunnel 内置的 Sink 连接器之一核心职责是将接收到的数据写成附件文件并发送到一个或多个邮箱地址。从官方文档描述看其工作方式可以概括为三点连接器会把每张表的数据缓冲到一个带分隔符的附件文件里每行一条记录不含表头在 Writer 关闭时为每张表各发送一封邮件如果某张表没有数据则不会为该表发送邮件。这意味着 Email Sink 非常适合批处理结束后的结果通知场景把当日报表、告警明细、数据对账结果等以 CSV 附件的形式定时推送到运维或业务人员的邮箱。引擎支持Email Sink 同时支持以下三种运行引擎与 Email.md 中声明一致SparkFlinkSeaTunnel Zeta支持版本官方文档标注的测试版本为1.5.6仅供参考。从模块构建文件 connector-email/pom.xml 可以看到该版本号对应的是其底层邮件发送依赖com.sun.mail:javax.mail的版本properties email.version1.5.6/email.version /properties ... dependency groupIdcom.sun.mail/groupId artifactIdjavax.mail/artifactId version${email.version}/version /dependency也就是说邮件发送能力由 JavaMailjavax.mailAPI 提供连接器本身只负责数据的缓冲、格式化与 JavaMail 会话的装配。能力特性特性支持情况精确一次Exactly-Once❌ 不支持多表写入Multi-Table Write✅ 支持定时刷新Scheduled Refresh❌ 不支持以上链接指向概念文档 connector-v2-features.md由于不支持精确一次Email Sink 适合对投递可靠性要求不苛刻的通知类场景而支持多表写入意味着一个作业可以同时向同一批收件人分别发送多张表的附件。工作原理从数据行到邮件附件为了准确理解各配置项的作用有必要先看连接器的核心实现。Email Sink 模块位于seatunnel-connectors-v2/connector-email其核心类包括类职责EmailSinkFactory.java插件工厂声明factoryIdentifier EmailSink并定义参数校验规则EmailSink.javaSink 入口实现SupportMultiTableSink创建 WriterEmailSinkWriter.java核心 Writer缓冲数据行、生成附件文件、调用 JavaMail 发信EmailSinkOptions.java全部选项定义与默认值EmailSinkConfig.java把ReadonlyConfig解析为强类型的配置对象数据缓冲与格式化write 阶段在 EmailSinkWriter.write() 中每一行数据被追加到一个StringBuffer字段按上游 schema 的顺序遍历null值被写成空字符串源码注释明确写到 Handle null field values to avoid NPE非空字段直接toString()字段之间以email_field_delimiter分隔行末追加换行符一旦写入过至少一行hasData标记置为true。也就是说附件内容是一个典型的无表头、分隔符分隔、每行一条记录的文本文件。发信时机与空数据判定close 阶段在 EmailSinkWriter.close() 中连接器首先检查hasData// Only send email if there was data written successfully if (!hasData) { log.info(No data to send, skipping email); return; }如果没有任何数据行直接跳过既不生成附件也不发送邮件。这正是文档中只有上游至少写入一行数据时Email Sink 才会发送邮件的实现依据。附件生成createFile()方法把StringBuffer内容写入到email_attachment_name指定的本地文件中文件不存在则创建随后通过FileDataSource作为附件附加到MimeMultipart邮件体中DataSource source new FileDataSource(filename); messageBodyPart.setDataHandler(new DataHandler(source)); messageBodyPart.setFileName(filename); multipart.addBodyPart(messageBodyPart);SMTP 会话与 SSL 处理close()中装配 JavaMailProperties并通过MailSSLSocketFactory设置setTrustAllHosts(true)信任所有 SMTP 主机。关键逻辑是if (config.getEmailSmtpAuth()) { properties.put(mail.smtp.ssl.enable, true); session Session.getDefaultInstance(properties, new Authenticator() { Override protected PasswordAuthentication getPasswordAuthentication() { return new PasswordAuthentication(config.getEmailFromAddress(), config.getEmailAuthorizationCode()); } }); } else { session Session.getDefaultInstance(properties); }这段源码印证了文档中的认证与 SSL 说明当email_smtp_auth true时自动开启 SSLmail.smtp.ssl.enable true并以email_from_address作为用户名、email_authorization_code作为密码进行认证当email_smtp_auth false时走普通无 SSL 的 SMTP不做认证。收件人解析则直接对email_to_address按英文逗号,切分逐个构造InternetAddress后通过message.setRecipients(Message.RecipientType.TO, addresses)一次性设置多个收件人最后调用Transport.send(message)发送。异常与错误码发送失败时会抛出 EmailConnectorException对应的错误码定义在 EmailConnectorErrorCode.java 中public enum EmailConnectorErrorCode implements SeaTunnelErrorCode { SEND_EMAIL_FAILED(EMAIL-01, Send email failed); }附件文件写入失败时则抛出CommonError.fileOperationFailed(Email, create, fileName, e)。排查问题时看到EMAIL-01即可定位到发送邮件失败。参数选项详解以下选项表来自 Email.md与 EmailSinkOptions.java 中的定义一一对应名称类型是否必须默认值email_from_addressstring是-email_to_addressstring是-email_hoststring是-email_transport_protocolstring是-email_smtp_authboolean是-email_smtp_portint否465email_authorization_codestring是-email_message_headlinestring是-email_message_contentstring是-email_attachment_namestring否emailsink.csvemail_field_delimiterstring否,multi_table_sink_replicaint否1common-options否-参数校验规则定义在 EmailSinkFactory.optionRule()前 8 个选项发件人、收件人、SMTP 主机、传输协议、认证开关、授权码、标题、正文为必填email_smtp_port可选但有边界校验greaterThan(0).and(lessOrEqual(65535))即取值必须在1到65535之间含边界multi_table_sink_replica来自 Sink 通用选项SinkConnectorCommonOptions.MULTI_TABLE_SINK_REPLICA。email_from_address [string]必填发件人邮箱地址。在启用 SMTP 认证时它还同时作为 SMTP 登录用户名使用见上文源码。email_to_address [string]必填接收邮件的地址支持多个邮箱地址以英文逗号,分隔。示例receiver-1example.com,receiver-2example.com。从源码看收件人解析正是config.getEmailToAddress().split(,)后逐一构造地址因此多个收件人之间不要混用中文逗号或空格。email_host [string]必填连接的 SMTP 服务器地址例如smtp.qq.com、smtp.gmail.com、smtp.163.com或自建邮件服务器的域名 / IP。email_transport_protocol [string]必填发送邮件使用的传输协议通常为smtp或smtps。对应 JavaMail 属性mail.transport.protocol。email_smtp_auth [boolean]必填是否启用 SMTP 认证。结合源码可以明确其行为设为true时连接器会自动开启 SSLmail.smtp.ssl.enable并信任所有 SMTP 主机以email_from_address作为用户名、email_authorization_code作为密码进行认证设为false时连接器通过普通 SMTP 发送邮件不进行认证。email_smtp_port [int]可选默认 465SMTP 服务器端口取值必须在1到65535之间包含边界值工厂校验会提前失败而不是等发送时才报错。默认值465为 SMTPS 端口需与email_smtp_auth true配合使用。如果使用不带认证的普通 SMTP请填写与服务匹配的端口例如25或本地测试服务的3025。email_authorization_code [string]必填当email_smtp_auth true时用于与email_from_address一起对 SMTP 会话进行认证的授权码或专用密码。可以从邮箱设置中获取例如 QQ 邮箱、Gmail、163 邮箱等提供的专用密码 / 授权码。连接器要求必须配置该项属于必填选项。当email_smtp_auth false时可以配置为空字符串示例中即为email_authorization_code 。⚠️ 注意请使用邮箱服务商提供的授权码专用密码而不是邮箱账号的登录密码。email_message_headline [string]必填邮件的标题主题对应MimeMessage.setSubject(...)。email_message_content [string]必填邮件消息的正文对应MimeBodyPart.setText(...)。正文会以纯文本形式展示在邮件中与附件并列。email_attachment_name [string]可选默认 emailsink.csv邮件附件的文件名。默认值为emailsink.csv。连接器会先把数据写到本地这个文件里再作为附件发送。注意附件文件名建议使用.csv等常见后缀便于收件方直接打开。email_field_delimiter [string]可选默认 ,附件文件中用于分隔字段的分隔符默认逗号,。单元测试 EmailSinkWriterTest.java 中验证了自定义分隔符如|的解析正确性。附件格式约定来自文档附件不包含表头字段会按上游 schema 的顺序写入null值会写成空字符串。multi_table_sink_replica [int]可选默认 1多表写入时每张表使用的 Sink Writer 副本数默认值为1。这是 Sink 连接器的通用选项之一完整说明见 Sink 常用选项。common optionsSink 插件常用参数请参考 Sink 常用选项了解详情例如plugin_input指定上游数据集、parallelism覆盖 env 并行度等。空数据行为只有上游至少写入一行数据时Email Sink 才会发送邮件。如果上游表没有数据则不会发送邮件。这一行为在EmailSinkWriter.close()中通过hasData标记实现详见上文发信时机与空数据判定其对应的单元测试也覆盖了 null 值写入场景。完整配置示例以下三个示例完整继承自 Email.md可直接作为作业配置模板使用。示例一发送单表数据到多个收件人该示例使用不需要认证的 SMTP 服务并发送一封收件人列表由email_to_address决定的邮件。使用 FakeSource 生成 100 行数据id/name/age三列经 EmailSink 处理后发送env { parallelism 1 job.mode BATCH } source { FakeSource { tables_configs [ { row.num 100 schema { table test.table1 columns [ { name id type bigint }, { name name type string }, { name age type int } ] } } ] } } sink { EmailSink { email_from_address senderexample.com email_to_address receiver-1example.com,receiver-2example.com email_host smtp.example.com email_transport_protocol smtp email_smtp_auth false email_smtp_port 3025 email_authorization_code email_message_headline test-title email_message_content test-content email_attachment_name report.csv email_field_delimiter | } }要点说明无认证模式下email_authorization_code配为空字符串端口使用普通 SMTP 端口3025示例中的本地测试服务端口email_field_delimiter |演示了把附件字段分隔符改为竖线email_attachment_name report.csv自定义了附件文件名。示例二发送多表数据Email Sink 支持多表输入。以下配置中两个上游表会让每个收件人各收到两封邮件每张表一封附件分别包含各自表的数据source { FakeSource { tables_configs [ { row.num 100 schema { table test.table1 fields { id bigint name string age int } } }, { row.num 100 schema { table test.table2 fields { id bigint name string age int } } } ] } } sink { EmailSink { email_from_address senderexample.com email_to_address receiver-3example.com,receiver-4example.com email_host smtp.example.com email_transport_protocol smtp email_smtp_auth false email_smtp_port 3025 email_authorization_code email_message_headline test-title email_message_content test-content } }该示例与示例一的差异在于未显式配置email_attachment_name和email_field_delimiter因此附件文件名取默认值emailsink.csv、分隔符取默认逗号,。示例三使用 SMTP 认证发送以 QQ 邮箱为例本示例通过 QQ 邮箱等需要认证的 SMTP 服务器发送邮件。当email_smtp_auth true时连接器会自动开启 SSL并使用email_from_address和email_authorization_code进行认证。请将授权码替换为你在邮箱设置中生成的授权码env { parallelism 1 job.mode BATCH } source { FakeSource { row.num 100 schema { fields { id bigint name string age int } } } } sink { EmailSink { email_from_address xxxxxxxxqq.com email_to_address xxxxxxxxqq.com email_host smtp.qq.com email_transport_protocol smtp email_smtp_auth true email_authorization_code your-authorization-code email_message_headline test-title email_message_content test-content } } 认证与 SSL 提示来自原文档当email_smtp_auth true时连接器会开启 SSLmail.smtp.ssl.enable并信任所有 SMTP 主机因此默认端口465SMTPS适用于 QQ 邮箱、Gmail、163 邮箱等服务商。请使用邮箱的**授权码专用密码**作为email_authorization_code而不是账号登录密码。当email_smtp_auth false时连接器通过不带 SSL 的普通 SMTP 发送邮件请选择匹配的普通 SMTP 端口例如25或本地测试服务的3025。常见问题FAQEmail Sink 如何格式化邮件附件连接器会将每张表接收到的数据行缓冲到带分隔符的附件文件中通过email_attachment_name配置默认为emailsink.csv并在 Writer 关闭时将附件随邮件发出。附件不含表头字段按上游 schema 顺序排列null值写成空字符串行间以换行符分隔。如果没有接收到任何数据行还会发送邮件吗不会。如果某张表没有接收到任何数据行不会生成附件也不会为该表发送邮件对应源码中hasData判空逻辑。是否可以配置多个收件人邮箱可以。在email_to_address中配置多个收件人邮箱使用英文逗号,分隔例如user1example.com,user2example.com。为什么发送失败报 EMAIL-01 错误EMAIL-01Send email failed是连接器在Transport.send(message)抛出异常时包装的错误码定义于 EmailConnectorErrorCode.java。常见诱因包括授权码错误、SMTP 端口不匹配认证模式下未使用 465 或未开启 SSL、发件地址与登录用户名不一致、目标 SMTP 服务器网络不可达等可结合异常堆栈中的EmailConnectorException信息逐项排查。如何运行验证在仓库根目录下使用 SeaTunnel Zeta 引擎即可本地验证上述配置# 假设将上述示例保存为 email.conf ./bin/seatunnel.sh --config email.confbin/seatunnel.sh位于构建产物目录中配置模板可参考仓库 config/v2.batch.config.template示例中的FakeSource为内置数据源无需外部依赖即可生成测试数据。建议先在本地起一个简易 SMTP 服务如示例中的3025端口或无认证的中继验证附件格式与收件人拆分是否符合预期再切换到真实邮箱的 SMTPS 465 端口授权码的生产配置。源码结构与延伸阅读Email Sink 的完整模块源码位于 seatunnel-connectors-v2/connector-email包含 6 个主类与 2 个测试类选项与配置EmailSinkOptions.java、EmailSinkConfig.javaSink 实现EmailSink.java、EmailSinkWriter.java、EmailSinkFactory.java单元测试EmailSinkWriterTest.java覆盖 null 值写入、自定义分隔符、EmailFactoryTest.java构建依赖connector-email/pom.xml基于javax.mail 1.5.6相关文档还可继续阅读Email Sink 官方文档英文Sink 常用选项Sink 写入模式与 Save ModeConnector V2 能力特性说明Email 连接器的版本变更记录见 docs/zh/connectors/changelog/connector-email.md总体而言Email Sink 是一个轻量、易用的通知类 Sink不追求精确一次但凭借对多表写入、多收件人、自定义附件格式与 SMTP 认证/SSL 的完整支持足以胜任绝大多数数据结果邮件推送的场景。在实际落地时只需重点把握认证模式选 465 授权码、无认证模式选普通端口空授权码这两个要点即可快速完成配置。【免费下载链接】seatunnelSeaTunnel is a multimodal, high-performance, distributed, massive data integration tool.项目地址: https://gitcode.com/GitHub_Trending/se/seatunnel创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考

SeaTunnel Email Sink 连接器详解：将数据以附件形式发送邮件

相关推荐

yarn.lock 地址不对怎么办？从 signature 哈希到依赖锁定机制全解析

nas-tools 2026完整指南：三步部署你的NAS媒体库管理工具

TodoMVC 应用规范（App Specification）完全解读：从目录结构到路由与持久化的实现指南

更多资讯

装备体系作战试验分布式仿真系统：架构选型、模型集成与排错验证

高通Chromatix 7 ISP调优实战：从环境搭建到模块参数详解

汽车零部件物流系统集成与实时协同技术实践

多租户AI智能客服系统架构实战：从Dify到Spring AI的隔离与编排

AI大模型就业黄金期：岗位需求与转型指南

MATLAB实现一维信号分类的CNN实战指南

今日更新

校园网认证计费模式详解：AAA与代拨架构、原理及排障

Spring Boot+Vue垃圾分类毕设：分层、权限与排错实战

把 Cline 的 Base URL 改到 TaoToken 之后，DeepSeek 仍能修调试配置

本周精选

AI驱动的自动化专利分析平台技术解析

LeetCode-Go 题解：1252. Cells with Odd Values in a Matrix（奇数值单元格计数）双解法剖析

在线答疑系统Java毕设实战：Spring Boot前后端分离与状态流转

本月热门

基于YOLO的焊缝缺陷检测：从数据标注到工业部署全流程实践

分布式服务的决策记录

创业产品的首版范围

读完文章，想聊聊您的网站？