Flink SQL自研连接器实战：从UDP日志源到动态表连接器全解析

看到“自定义数据源”这几个字，不少人的第一反应是“官方连接器还不够用吗”。这个反应我特别理解——Kafka、JDBC、Hive、ClickHouse、ES，该有的几乎都有了，日常要接的数据源十有八九都能从官方清单里找到。但真正把Flink SQL推到业务深处之后，你迟早会遇到几类“看起来很简单、官方却偏偏没有”的源：内部老平台只给你一个UDP端口、某个自研网关只暴露HTTP回调、监控组件生成的文本落在共享目录里。数据量不大、格式也不复杂，可你就是没法用一个现成连接器把它直接变成一张SQL表。这时候，自己动手写一个Flink SQL Connector，不是炫技，是因为根本没有捷径。

这篇文章不打算去解析Kafka连接器那种配置项多到让人头皮发麻的标杆实现，而是以“UDP日志源”这个最小但完整的案例，带你走一遍自研连接器的完整链路：从Factory自动发现、DynamicTableSource，到运行时SourceFunction，再到SQL验证、维表扩展和发布避坑。读完你就能把一类自定义数据源变成一条CREATE TABLE就能直接用的SQL表，后续想扩展更多协议、更多格式，也有一个清晰的底子可以往上搭。

1. 为什么需要自研Flink SQL Connector：场景、代价与决策边界

1.1 官方连接器覆盖不到的“最后一公里”

我在实际项目里遇到过几次特别典型的“最后一公里”需求。第一类是老系统接口。某个内部平台运行了七八年，对外只暴露UDP端口，运维同学把数据往这个端口一丢就算完事。换到现在的大数据体系里，你想让这件事进入Flink SQL，官方连接器列表里根本找不到UDP这个选项。

第二类是半结构化文本。很多监控组件、探针脚本会把日志写到固定目录或者通过syslog转发，你拿到的不是标准JSON也不是CSV，而是“自定义分隔符的混合行”。Kafka连接器虽然能读原始字节，但前置还得有一个采集进程把这些文本推送到Kafka，链路多一跳，维护就多一层。

第三类是需要特殊鉴权的内部HTTP接口。这类接口只对内网某个服务开放，调用时要带签名、要滑动窗口限流，这些逻辑根本没法用官方提供的HTTP连接器解决，更别提官方压根没有通用HTTP连接器。

这些场景的共同特点是：数据量不大，但接入模式非常明确，而且业务希望直接用SQL去查。自研一个连接器，本质上就是把“我要怎么把这个源变成一张表”这件事固化下来，后续任何团队想用这张表，一行DDL就够了。

1.2 自研Connector vs 旁路脚本：成本收益怎么算

也有人会问：既然数据量不大，为什么不写一个Python脚本定时落库，或者直接用DataStream API写一个Source再接Table？这两种方案当然都成立，但长期看都有隐性成本。

旁路脚本的问题在于它和Flink任务体系是割裂的。脚本要单独部署、单独监控、单独处理失败重跑，一旦数据格式变了，脚本和SQL任务往往要分两次改。而如果直接在DataStream API里写Source，虽然能绕开Connector开发，但每次新需求都要重新写一遍接入逻辑，也很难让不同团队通过标准SQL复用这张表。

Connector方案把问题收敛到了一个可复用的组件里。写完一次，connector='udp-log'这个标识就永久成了公司内部的“标准数据源能力”，任何人有权限就可以通过建表语句接入。从团队协作和长期维护的角度看，这笔投入是值得的。

当然我也得说句公道话：如果业务比较急，而且只有一两个任务在用，旁路脚本确实更快。什么时候该自研，请务必先做完1.3节里的判断，别一上来就钻Connector开发的牛角尖。

1.3 什么时候不该自研：先看四种替代方案

我总结过一套比较实用的决策顺序，开发前先过一遍：

• 能中转就中转。如果数据源前方可以加一个Kafka Producer或者JDBC写入，优先走官方连接器。多一跳并不是坏事，有时候反而能解耦生产端和消费端。
• 周期性拉取可以先用UDF/DataStream挡一下。如果数据源是HTTP接口、且半小时拉一次也能接受，完全可以在Flink作业里写一个UDF做周期拉取，不必直接开发Connector。
• 小数据量维表优先用Lookup方案。如果这个源主要是给流任务做关联用的，而且数据量撑得住同步查询，那可以走4节的Lookup连接器路线，未必需要Scan Table Source。
• 能用CDC框架套壳就套壳。很多老系统虽然没有标准协议，但其底层数据源是数据库或者文件，Flink CDC本身已经覆盖了大量变更捕获场景，先看看能不能直接用。

如果以上四种都不合适，且你确实需要一个“持续读取外部数据源并把它当流式表”的能力，那自研Connector就是板上钉钉的事。

2. 动手编码前要把机制吃透：Factory发现、动态表与运行时Provider

2.1 CREATE TABLE ... WITH 'connector'='x' 的背后

很多新手第一次写连接器，最大的困惑不是怎么实现Source，而是“为什么我写了一个类，Flink就能自动找到它”。这背后的机制其实很简单：Flink没有在代码里硬编码任何连接器列表，而是通过Java标准的ServiceLoader机制，在类路径下扫描META-INF/services目录，找到所有org.apache.flink.table.factories.Factory的实现类。

当你执行一条建表语句时，Flink做的事情大概是这样：

1. 解析SQL，得到一个CatalogTable，里面包含了WITH子句里所有的配置项和Schema定义。
2. TableEnvironment把CatalogTable交给Planner。
3. Planner通过connector这个配置项的值作为标识，扫描并匹配所有可用的Factory。
4. 命中的Factory创建对应的DynamicTableSource或DynamicTableSink对象。
5. Planner再通过Source对象提供的运行时Provider，把实际的DataStream接入到作业图中。

所以“connector='udp-log'”里的udp-log，本质上就是工厂发现时的唯一标识。这个标识要求全小写，一般用连字符分隔单词。

2.2 新老两套接口：不要在错误的版本上写旧代码

Flink的表连接器接口经历过一次比较大的演进。Flink 1.9时代用的是TableFactory和TableSource那套旧接口，但这套接口在后续版本里已经逐渐被废弃和移除。从Flink 1.11开始，官方推荐的新链路是DynamicTableSourceFactory+DynamicTableSource，这套接口也被称为“动态表连接器API”。

动态表连接器API的核心角色有四个：

需要注意的是，如果你现在用的是Flink 1.14之后的版本，建议直接基于DynamicTableSourceFactory和ScanTableSource来写。网上还能搜到很多基于旧接口的老博客，照着写大概率编译都过不了。至于运行时Provider，SourceFunctionProvider在Flink 1.15之后已经不是唯一选择，新的SourceProvider也支持基于新Source API的接入方式，但SourceFunctionProvider仍然兼容可用，本文案例为了降低理解门槛，就用它来做演示。

2.3 Scan、Lookup、CDC三类Source的API差异

自研连接器前，先搞清楚你要实现的是哪一类Source，这三类在API设计上差异很大：

• Scan Table Source：持续扫描外部数据源，把数据作为无界或有界流读进来。典型场景是读Kafka、读文件、读UDP端口。需要实现ScanTableSource接口。
• Lookup Table Source：临时去外部系统查一次，主要用于维表关联。典型场景是流任务JOIN一个MySQL配置表。需要实现LookupTableSource接口。
• CDC Source：捕获上游数据库或消息系统的增量变更，输出包含Insert/Update/Delete语义的Changelog流。它在内部往往同时具备Scan和Lookup的一些特性。

如果你的目标是“把外部数据变成一张可以被持续查询的表”，那大多数时候实现的是Scan Table Source。文章后面会先重点讲Scan，再在第四节展开Lookup。

3. 实战：从零开发UDP日志Source连接器，全程代码链路

3.1 工程与依赖准备：别把运行时依赖打进去

先建一个标准的Maven工程，artifact可以叫flink-connector-udp-log。这个连接器最终要放到Flink运行时环境里，所以依赖上有一个非常重要的原则：Flink自身已经提供的包，一律用provided作用域，不要打到jar里。

<properties> <flink.version>1.17.1</flink.version> </properties> <dependencies> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-common</artifactId> <version>${flink.version}</version> <scope>provided</scope> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-java</artifactId> <version>${flink.version}</version> <scope>provided</scope> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-api-java-bridge</artifactId> <version>${flink.version}</version> <scope>provided</scope> </dependency> </dependencies>

这里最核心的是flink-table-common，它包含了Factory接口和ConfigOption等基础类。flink-table-api-java-bridge主要是为了在你的工程里能引用RowData、GenericRowData这些表运行时类型。

版本方面，编译用的Flink版本最好和你线上运行的Flink完全一致，尤其是Flink 1.14到1.17之间，表连接器API有过一些细微调整。如果不一致，最常见的报错是NoSuchMethodError或者NoClassDefFoundError，这类问题比逻辑Bug更难排查。

3.2 工厂类：UdpLogTableSourceFactory

工厂类是连接器和SQL语句之间的第一层桥梁。它要做三件事：告诉Flink自己的标识是什么、声明哪些配置项是必填/可选、根据配置创建Source对象。

package com.example.connector.udp; import org.apache.flink.configuration.ConfigOption; import org.apache.flink.configuration.ConfigOptions; import org.apache.flink.configuration.ReadableConfig; import org.apache.flink.table.factories.DynamicTableSourceFactory; import org.apache.flink.table.factories.FactoryUtil; import org.apache.flink.table.api.ValidationException; import org.apache.flink.table.connector.source.DynamicTableSource; import java.nio.charset.Charset; import java.util.HashSet; import java.util.Set; public class UdpLogTableSourceFactory implements DynamicTableSourceFactory { public static final ConfigOption<String> HOST = ConfigOptions.key("host") .stringType() .noDefaultValue() .withDescription("UDP监听地址"); public static final ConfigOption<Integer> PORT = ConfigOptions.key("port") .intType() .noDefaultValue() .withDescription("UDP监听端口"); public static final ConfigOption<String> CHARSET = ConfigOptions.key("charset") .stringType() .defaultValue("UTF-8") .withDescription("解码字符集"); @Override public String factoryIdentifier() { return "udp-log"; } @Override public Set<ConfigOption<?>> requiredOptions() { Set<ConfigOption<?>> options = new HashSet<>(); options.add(HOST); options.add(PORT); return options; } @Override public Set<ConfigOption<?>> optionalOptions() { Set<ConfigOption<?>> options = new HashSet<>(); options.add(CHARSET); return options; } @Override public DynamicTableSource createDynamicTableSource(Context context) { FactoryUtil.TableFactoryHelper helper = FactoryUtil.createTableFactoryHelper(this, context); helper.validate(); ReadableConfig options = helper.getOptions(); String host = options.get(HOST); Integer port = options.get(PORT); Charset charset = Charset.forName(options.get(CHARSET)); return new UdpLogDynamicTableSource(host, port, charset); } }

这里有两个细节值得注意。

requiredOptions和optionalOptions看起来只是声明，但Flink会基于它们在helper.validate()阶段做参数校验。如果用户在DDL里漏写了port，会直接得到一条清晰的ValidationException，而不是到运行时才报错。这个特性对连接器使用者非常友好，写成连接器时一定要用好。

factoryIdentifier()返回的"udp-log"必须与建表语句WITH里的connector='udp-log'完全一致。如果这里返回的是udp_log，建表时写udp-log就匹配不上，Flink会直接说找不到工厂。

3.3 DynamicTableSource与ScanRuntimeProvider：把运行时交出去

工厂创建出来的UdpLogDynamicTableSource要继承ScanTableSource。这个类的作用是描述“这个外部数据源的表语义”，包括它输出的是Insert流还是完整的Changelog流，以及如何把运行时组件交给Planner。

package com.example.connector.udp; import org.apache.flink.table.connector.ChangelogMode; import org.apache.flink.table.connector.source.DynamicTableSource; import org.apache.flink.table.connector.source.ScanTableSource; import org.apache.flink.table.connector.source.SourceFunctionProvider; import org.apache.flink.table.data.RowData; import org.apache.flink.table.connector.source.ScanTableSource.ScanRuntimeProvider; import org.apache.flink.table.connector.source.DynamicTableSource.Context; import java.nio.charset.Charset; public class UdpLogDynamicTableSource implements ScanTableSource { private final String host; private final int port; private final Charset charset; public UdpLogDynamicTableSource(String host, int port, Charset charset) { this.host = host; this.port = port; this.charset = charset; } @Override public ChangelogMode getChangelogMode() { return ChangelogMode.insertOnly(); } @Override public ScanRuntimeProvider getScanRuntimeProvider(ScanContext runtimeProviderContext) { return SourceFunctionProvider.of( new UdpLogSourceFunction(host, port, charset), false ); } @Override public DynamicTableSource copy() { return new UdpLogDynamicTableSource(host, port, charset); } @Override public String asSummaryString() { return "UdpLog-Source"; } }

getChangelogMode()返回insertOnly()，因为UDP日志源只会新增数据，不会对已有数据做修改或删除。如果以后你要做一个数据库CDC连接器，这里就该返回ChangelogMode.all()或者upsert()。

SourceFunctionProvider.of()的第二个布尔参数表示这个Source是不是有界流。UDP端口是持续监听的，显然是无界流，所以传false。如果一个Source是读取一批固定文件，有明确结束边界，这里就传true。

getScanRuntimeProvider里的ScanContext还能做很多事，比如获取物理Schema、获取目标数据类型、创建TypeInformation。我们为了简单，直接输出一个固定结构的RowData，具体字段处理和Schema的对应关系在3.5节再展开。

另外，copy()方法千万别遗漏。Flink Planner可能会对同一个Source对象做多份拷贝用于优化和计划生成，如果copy()直接返回this或者返回错误实例，某些复杂查询下会出现很奇怪的运行时问题。

3.4 UDP接收线程与生命周期：RichSourceFunction的正确写法

现在到了连接器真正的“内脏”——接收UDP数据包的SourceFunction。我选择继承RichSourceFunction<RowData>，因为能在open()和close()里管理Socket资源。

package com.example.connector.udp; import org.apache.flink.configuration.Configuration; import org.apache.flink.streaming.api.functions.source.RichSourceFunction; import org.apache.flink.table.data.GenericRowData; import org.apache.flink.table.data.RowData; import java.net.DatagramPacket; import java.net.DatagramSocket; import java.net.InetSocketAddress; import java.net.SocketException; import java.net.SocketTimeoutException; import java.nio.charset.Charset; public class UdpLogSourceFunction extends RichSourceFunction<RowData> { private final String host; private final int port; private final Charset charset; private transient DatagramSocket socket; private transient volatile boolean running; public UdpLogSourceFunction(String host, int port, Charset charset) { this.host = host; this.port = port; this.charset = charset; } @Override public void open(Configuration parameters) throws Exception { socket = new DatagramSocket(new InetSocketAddress(host, port)); socket.setSoTimeout(1000); running = true; } @Override public void run(SourceContext<RowData> ctx) throws Exception { byte[] buffer = new byte[65507]; while (running) { DatagramPacket packet = new DatagramPacket(buffer, buffer.length); try { socket.receive(packet); } catch (SocketTimeoutException e) { continue; } catch (SocketException e) { if (!running) { return; } throw e; } String line = new String(packet.getData(), packet.getOffset(), packet.getLength(), charset); GenericRowData row = new GenericRowData(1); row.setField(0, line); synchronized (ctx.getCheckpointLock()) { ctx.collect(row); } } } @Override public void cancel() { running = false; if (socket != null) { socket.close(); } } @Override public void close() throws Exception { if (socket != null) { socket.close(); } } }

这段代码有几个细节是实战中踩过坑才总结出来的：

• 设置了SO_TIMEOUT，让receive()每秒钟超时一次。否则cancel()里调用socket.close()虽然能打断阻塞，但有些平台上会直接抛SocketException，导致作业以异常退出。有了超时机制配合running标志，退出会非常干净。
• 用GenericRowData而不是自定义的Result类。这是因为Flink表生态里，运行时统一以RowData作为行数据载体。你如果直接从SourceFunction里发射一个自定义Java对象，Planner那边还需要额外的转换逻辑，非常麻烦。
• synchronized (ctx.getCheckpointLock())是Flink SourceFunction的经典写法。它确保Flink在做checkpoint快照时，不会出现“状态已经记录了一部分，但数据还没完全发射出去”的错乱。

如果你以后想把UDP连接器升级成高吞吐的Kafka级别Source，这条链路还有更多可做：改用Source API、接Watermark、做分区发现等。但对内部工具性质的日志源，RichSourceFunction这个粒度已经足够。

3.5 注册Factory并跑通第一条SQL：从打包到验证

写完三个核心类后，还差最后一步：把工厂注册到META-INF/services，让Flink能发现它。

在src/main/resources/META-INF/services/目录下创建一个文件，文件名必须是org.apache.flink.table.factories.Factory，文件内容填工厂类的全限定名：

com.example.connector.udp.UdpLogTableSourceFactory

然后打包：

mvn clean package

启动SQL Client时，用-j参数把Connector jar加载进来：

flink-sql-client -j flink-connector-udp-log-1.0.0.jar

在SQL Client里执行：

CREATE TABLE udp_log ( data STRING ) WITH ( 'connector' = 'udp-log', 'host' = '0.0.0.0', 'port' = '9527' ); SELECT * FROM udp_log;

然后用一个简单的Python命令发一条UDP包测试：

python3 -c "import socket; s=socket.socket(socket.AF_INET, socket.SOCK_DGRAM); s.sendto(b'hello flink connector', ('127.0.0.1', 9527))"

如果一切正常，SQL Client会立即打印出hello flink connector这行数据。到这里，一个最简自定义连接器就算完整跑通了。

细心的读者会发现，这个连接器目前只能输出一个String类型的字段，Schema里也只能定义data STRING。如果你的实际数据是一行JSON或者CSV，并且想要把它解析成多个字段，建议走正式连接器更通用的做法：引入DecodingFormat机制，在Factory里通过helper.discoverDecodingFormat()解析format选项，把真正的反序列化交给JSON Format或CSV Format这类官方格式组件完成。这个机制写起来逻辑会多一层，但设计思路是相通的——连接器负责“从哪里读”，Format负责“怎么解析”，两者解耦后复用性会好很多。

4. 从Scan提升到Lookup：维表连接器要额外实现的几件事

4.1 为什么流任务总是绕不开维表关联

实时流任务里，最普遍的需求之一就是给事实数据补维度信息。订单流要关联用户表拿用户等级，点击流要关联商品表拿商品分类，监控日志要关联设备表拿设备名称。这类数据量往往不会特别大，但更新频率不低，非常适合做成维表。

Flink SQL里，这种临时查询外部表并返回关联结果的能力，就是Lookup Table Source。它在执行时不是持续扫描整张表，而是根据流中每条数据的关联键，按需去外部系统查询一次。

4.2 LookupTableSource与LookupRuntimeProvider的核心接口

实现维表连接器，需要让Source类实现LookupTableSource接口，而不是ScanTableSource。

package com.example.connector.lookup; import org.apache.flink.table.connector.source.DynamicTableSource; import org.apache.flink.table.connector.source.LookupTableSource; import org.apache.flink.table.connector.source.LookupTableSource.LookupContext; import org.apache.flink.table.connector.source.LookupTableSource.LookupRuntimeProvider; public class DeviceDimTableSource implements LookupTableSource { private final String connectionUrl; public DeviceDimTableSource(String connectionUrl) { this.connectionUrl = connectionUrl; } @Override public LookupRuntimeProvider getLookupRuntimeProvider(LookupContext context) { int[] joinKeys = context.getKeys(); return LookupRuntimeProvider.of(new DeviceDimLookupFunction(connectionUrl, joinKeys)); } @Override public DynamicTableSource copy() { return new DeviceDimTableSource(connectionUrl); } @Override public String asSummaryString() { return "DeviceDim-Source"; } }

LookupContext.getKeys()返回的是维表里参与关联的字段索引数组，它告诉你SQL的ON条件里到底用了哪些维度字段。你的查询函数必须按照这个顺序接收参数。

真正的查询逻辑落在TableFunction里。以最简的“查数据库”为例：

package com.example.connector.lookup; import org.apache.flink.table.data.GenericRowData; import org.apache.flink.table.data.RowData; import org.apache.flink.table.functions.FunctionContext; import org.apache.flink.table.functions.TableFunction; import java.sql.Connection; import java.sql.DriverManager; import java.sql.PreparedStatement; import java.sql.ResultSet; import java.sql.SQLException; public class DeviceDimLookupFunction extends TableFunction<RowData> { private final String connectionUrl; private transient Connection connection; private transient PreparedStatement queryStmt; public DeviceDimLookupFunction(String connectionUrl, int[] joinKeys) { this.connectionUrl = connectionUrl; } @Override public void open(FunctionContext context) throws Exception { connection = DriverManager.getConnection(connectionUrl); queryStmt = connection.prepareStatement("SELECT device_name, status FROM device_dim WHERE device_id = ?"); } public void eval(Object deviceId) { try { queryStmt.setObject(1, deviceId); try (ResultSet rs = queryStmt.executeQuery()) { while (rs.next()) { GenericRowData row = new GenericRowData(3); row.setField(0, deviceId); row.setField(1, rs.getString(1)); row.setField(2, rs.getString(2)); collect(row); } } } catch (SQLException e) { throw new RuntimeException("设备维度查询失败", e); } } @Override public void close() throws Exception { if (queryStmt != null) queryStmt.close(); if (connection != null) connection.close(); } }

这里有一个很容易踩的坑：eval()方法不是只被单个线程调用。Flink的Lookup源通常会开多个并发实例，每个实例内部再被多条数据调用，所以PreparedStatement要么只能在单线程内复用，要么就用连接池。更稳妥的做法是使用HikariCP这类连接池替代裸DriverManager连接，把获取连接的逻辑放到eval()里而不是反复创建。

在SQL里，这个维表连接器可以直接参与流式JOIN：

SELECT o.order_id, d.device_name, d.status FROM orders AS o JOIN device_dim FOR SYSTEM_TIME AS OF o.proc_time AS d ON o.device_id = d.device_id;

很多团队内部维表一开始用Scan模式或者旁路同步，后来发现实时JOIN延迟升高才想起来换Lookup。我的建议是：只要维表数据量在百万级以内、查询链路走索引能保持在几十毫秒，就直接考虑Lookup模式，它天生就是为流式关联设计的。

4.3 实时维表连接常见的三个坑

第一，未命中维表数据不会自动告警。eval()里查不到数据就直接return，Flink不会给你任何提示。实际项目里一定要配合侧输出或者自定义日志，把维表未命中的情况单独统计出来，否则数据对不上会非常被动。

第二，缓存不是越久越好。很多团队为了提升维表连接性能，会给LookupFunction加上本地缓存。缓存能显著降低外部系统QPS，但也会拉长数据可见延迟。建议用可配置的过期时间，比如默认60秒，业务上能接受再调大。

第三，同步Lookup吞吐有限。同步eval()每来一条数据就查一次，并发有限时很容易变成整个作业的瓶颈。Flink还提供了AsyncLookupFunction接口，可以用异步方式和外部存储交互。如果维表查询的确压力大，优先上异步版本，而不是单纯堆并行度。

5. 自研连接器的避坑清单与发布规范

5.1 “Could not find any factory”的五个排查方向

这是写连接器时最经典的报错：

org.apache.flink.table.api.ValidationException: Could not find any factory for identifier 'udp-log'

出现这个错，90%的情况不是代码逻辑问题，而是Factory没有被Flink加载到。我的排查顺序固定是：

1. 确认META-INF/services/org.apache.flink.table.factories.Factory文件存在，而且文件名一个字母都不能错。

2. 确认文件内容写的是工厂类的全限定名，比如com.example.connector.udp.UdpLogTableSourceFactory，不带任何后缀或空格。

3. 用unzip -p直接看jar包里的注册文件是不是真的打进去了：

   unzip -p flink-connector-udp-log-1.0.0.jar META-INF/services/org.apache.flink.table.factories.Factory

4. 确认jar已经被SQL Client或作业classpath加载了。SQL Client用-j参数加载，Flink作业则要把jar放到lib目录或者通过ADD JAR加载。

5. 确认factoryIdentifier()返回的标识和SQL里connector选项值完全一致，包括大小写和连字符。

5.2 版本、Scope与类加载：依赖越少越安全

连接器jar在生产环境里面对的类加载环境比开发环境复杂得多。Flink本身有类加载隔离机制，作业jar和框架jar是分开加载的。如果你把flink-table-common这些框架类打进了连接器jar，极有可能出现“两个Factory类”冲突，最后Flink加载的还是已被隔离的版本，导致工厂发现逻辑变得不可预测。

所以我在3.1节里强调provided作用域，不是偷懒，是这条经验值过不少生产故障换来的。除此之外，发布前还应该用jar tf检查一下最终产物，确保没有携带flink-*、log4j-*这类框架级依赖。

5.3 Watermark、并行度与无法复用的资源

自研连接器的并行度设计，比普通DataStream作业要更谨慎。以UDP Source为例，多个并行实例同时监听同一个端口是不可能做到的，所以这个Source本身就只能以并行度1运行。如果要提高吞吐，就要设计成“多个端口同时收”，或者在Source内部自己实现多Receiver线程再接合并逻辑，而不是简单调并行度。

另一个容易被忽略的是Watermark。SQL里如果写了WATERMARK FOR字段，Planner需要Source流上带有对应的Watermark信息。在SourceFunction里，你可以用ctx.collectWithTimestamp()标记事件时间，再用ctx.emitWatermark()发射Watermark。如果不做这个动作，基于事件时间窗口的查询在自定义Source上就永远触发不了窗口计算。这个问题曾经困扰过我很久，当时查遍日志都没发现异常，最后才意识到是Source没有发Watermark。

5.4 干净打包与发布：小细节决定故障率

打包时再说一个细节。如果你用了maven-shade-plugin做fat jar，一定要把META-INF/*.SF、META-INF/*.DSA、META-INF/*.RSA这些签名文件排除掉，否则在某些Java环境下会报“Invalid signature file digest”错误。你可以用这个配置片段：

<plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-shade-plugin</artifactId> <executions> <execution> <phase>package</phase> <goals> <goal>shade</goal> </goals> <configuration> <filters> <filter> <artifact>*:*</artifact> <excludes> <exclude>META-INF/*.SF</exclude> <exclude>META-INF/*.DSA</exclude> <exclude>META-INF/*.RSA</exclude> </excludes> </filter> </filters> </configuration> </execution> </executions> </plugin>

发布时，我通常建议把连接器jar放在Flink的lib目录下，而不是每次作业提交时临时ADD JAR。原因很简单：ADD JAR是动态加载，配合本地缓存和多个作业同时提交时，踩到类加载问题的概率会高不少。放在lib目录虽然需要重启集群来生效，但稳定性和可排查性都强很多。

发布前还应该准备一份简单的连接器说明，把支持参数、建表示例、性能边界写清楚。很多团队内部的自研连接器，最后不是死在代码上，而是死在没有文档，别人不敢用、不会用，再好的实现也沉淀不下来。

最后分享一个小技巧：我在排查连接器问题时，习惯先用EXPLAIN SELECT * FROM udp_log看Flink生成的执行计划。执行计划里会出现UdpLog-Source这样的描述，也就是asSummaryString()返回的内容。如果执行计划里连接器名称对不上，说明你当前任务实际加载的并不是你以为的那个jar。这个不起眼的方法，帮我省掉过很多次“明明改了代码却没生效”的困惑时间。