Flink SQL自定义Connector开发：从零实现数据源接入

做 Flink 实时计算的同学迟早会遇到一个场景官方提供的 JDBC、Kafka、File 等连接器都试了一遍发现某类业务数据源官方根本没适配。要么是自研消息队列只能通过 API 硬编码拉数据要么是某内部存储系统需要以 Source 方式接入 SQL 作业又或者你想把测试数据伪造成 Kafka topic 的形态来跑通 Flink SQL 任务。这时候就别再去翻有没有现成 Jar 了直接动手写一个“自定义数据源 Connector”才是正路。这篇文我会完整拆解 Flink SQL Connector 的开发机制从接口设计到落地实现最后附上一套可直接套用的自定义数据源接入流程适合正在做 Flink 二次开发、或者刚接触 Flink 扩展机制、被各种“找不到实现类”报错折磨过的开发者参考。1. 自定义 Connector 的整体设计与架构拆解1.1 Flink SQL 连接器到底在解决什么问题先搞清楚一点Flink SQL 本身不认识什么 MySQL、Kafka、ES它只认识一种抽象的表。用户在 SQL 里写CREATE TABLE声明字段名和类型后面接WITH参数剩下的就是让框架能根据这张表定义去外部系统读写数据。那这个“识别外部系统并完成数据传输”的桥梁就是 Connector一套遵循了 SPIService Provider Interface机制的插件。Flink 在运行时通过META-INF/services下的接口实现注册找到你写的工厂类再根据 SQL 中connector参数决定实例化哪一个实现。这个机制设计得非常巧妙用户侧的 DDL 是相对稳定的而各种数据源可以通过插件方式无限扩展不需要改 Flink 核心代码。自定义 Connector 的开发主体就是两件事一是实现一个DynamicTableSourceFactory或者DynamicTableSinkFactory负责解释 DDL 里的参数创建相应的 Source 对象二是实现一个DynamicTableSource负责真正产生和初步加工数据。对于自定义数据源来说读入数据时流的并行度、字段类型、watermark 分配、checkpoint 行为都在这个数据源里配置。1.2 为什么官方连接器不够用必须自己扩展很多团队的第一反应是为什么不能直接用FlinkKafkaConsumer或者RichSourceFunction当然可以但这样你会失去 SQL 层的所有表达能力。普通 DataStream API 里你写死了一个消费逻辑上层要想过滤、Join、聚合要么手工套 ProcessFunction要么把逻辑写在业务代码里维护成本很高。而 SQL 化的表数据源能让下游直接用标准 SQL 写分析和清洗逻辑配合 Flink SQL Planner 做过滤条件下推projection pushdown / filter pushdown性能通常比 DataStream API 手写更优。另外自研系统通常有各种私有协议。比如内部某个调度平台会把任务中间结果暴露成 WebSocket 接口、某个数字孪生系统使用 Protobuf 编码的二进制协议。要接进 Flink SQL只能在 Connector 层做协议适配因为这是官方留好的扩展点。如果不走 Connector你想在 SQL 任务里读这些数据只能通过 UDF 或者预先落库绕一大圈而且无法利用 Flink 的精确一次语义。1.3 自定义数据源的扩展点全景从 Flink 源码角度看一个完整的 Source Connector 涉及这些核心角色Factory负责解读 DDL 的WITH参数做合法性校验生成表描述相关的上下文对象。DynamicTableSource这是“扫描表”数据源的核心逻辑类似于 DataStream API 里的SourceFunction但它需要知道表结构ResolvedSchema、物理类型转换规则、是否支持流/批模式。ScanRuntimeProvider提供运行时实现比如生成SourceFunction、InputFormat或Source接口的实现。Flink 新版推荐使用基于Source接口的 FLIP-27 架构不过不少老项目还是SourceFunction居多。TableSource适配层旧版org.apache.flink.table.sources.TableSource用于兼容历史 API慢慢会被淘汰建议新代码不再直接实现它。除了 Source如果你希望自定义 Sink或者实现维表异步查询还有DynamicTableSink、LookupableTableSource等扩展点但本次以数据源为主重点放在读取侧。有一点容易被忽略Flink SQL 的 DDL 中connector属性就是工厂类的标识比如connector custom。这个custom在代码中并不是显式枚举的而是你在Factory的requiredContext()方法里声明的标识字符串。多个 Factory 之间靠这个标识做路由。如果两个 jar 都声称自己处理customFlink 会报冲突告诉你发现多个匹配的工厂实现。开发时要注意这个约束避免和其他扩展包里的连接器撞了标识。1.4 选型直接实现 SourceFunction 还是新的 Source 接口这个问题一开始容易纠结。如果你的 Flink 版本在 1.12 以上官方推荐使用支持批流一体的Source接口也就是常说的 FLIP-27 新架构。但实际开发中在 Table 环境里实现一个 Source 接口慢且细节繁多你需要处理SourceReader、SplitEnumerator、Checkpoint等复杂内容。而如果你是做一个内部小范围使用的数据源并且任务主要是流模式实现SourceFunction是更务实的方案因为ScanRuntimeProvider直接支持包装SourceFunction代码量小很多且容易控制。作为一个“自定义数据源扩展”的入门实践我更建议先用SourceFunction打通全链路确认表的 schema 映射、参数传递、watermark 策略都正常工作然后再根据需求迁移到新接口。这篇内容也会基于SourceFunction展开因为这是最直接、最快见效的一条路并且几乎所有 Flink 版本都兼容。2. 核心接口与关键实现细节2.1 需要 maven 引入的最小依赖组合先讨论依赖。开发 Connector 属于 Flink Table 模块的扩展所以必须引入 Table API 相关依赖因为工厂类和DynamicTableSource都在flink-table-api-java或flink-table-common中。如果只引入 Flink Streaming是无法看到这些接口的。以 Maven 为例最精简的一组 dependency 如下properties flink.version1.17.2/flink.version /properties dependencies dependency groupIdorg.apache.flink/groupId artifactIdflink-table-common/artifactId version${flink.version}/version scopeprovided/scope /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-table-api-java-bridge/artifactId version${flink.version}/version scopeprovided/scope /dependency !-- 如果运行时用IDE执行需要引入这些 -- dependency groupIdorg.apache.flink/groupId artifactIdflink-streaming-java/artifactId version${flink.version}/version scopeprovided/scope /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-clients/artifactId version${flink.version}/version scopeprovided/scope /dependency /dependency把 scope 设为 provided是因为最终运行在 Flink 集群中这些依赖由集群环境提供。IDE 里调试时需要加对应 module这个在 IDEA 运行配置里处理即可。开发过程中不建议直接把 flink 核心包打到 fat jar否则很容易出现因为覆盖内置类导致的冲突。2.2 一条 SQL DDL 的旅程在动手写代码之前建议先背熟一条 DDL 的解析过程。用户执行CREATE TABLE my_source ( id INT, name STRING, event_time TIMESTAMP_LTZ(3), watermark for event_time AS event_time - interval 5 second ) WITH ( connector custom, topic user_behavior, format json );此时 Flink SQL Planner 会扫描所有META-INF/services/org.apache.flink.table.factories.Factory中注册的工厂类。对每个工厂调用requiredContext()得到connector、format等上下文 key-value。用 DDL 的 WITH 配置做匹配找到connectorcustom的那个工厂。调用createDynamicTableSource(Context context)传入封装好的FactoryUtil上下文。之后框架拿着DynamicTableSource实例做优化、运行时生成代码最终变成SourceFunction或DataStream交给 JobGraph。因此你至少在 jar 包的resources/META-INF/services/org.apache.flink.table.factories.Factory这个文本文件中写上一行工厂类的全限定名。很多初学者的错误是忘了添加这个文件或者文件内容格式错误导致本地 run 时报Could not find any factories for connector custom。更隐蔽的坑是多个工厂文件的换行符不一致或者文件末尾多了不可见字符所以建议写完以后用cat -A看一眼。2.3 Factory 接口怎么实现才算合格Factory 接口的实现必须实现两个方法requiredContext()和factoryIdentifier()。从实际使用的角度我用一个自定义的模拟消息源来做示范。这个源叫custom它会从内存队列或者一个循环生成的测试数据中产生记录。写工厂类时最稳妥的方式是直接继承AbstractTableFactory但不同版本抽象类不同为了稳定起见直接实现DynamicTableSourceFactory更直接。核心代码如下import org.apache.flink.configuration.ConfigOption; import org.apache.flink.configuration.ConfigOptions; import org.apache.flink.table.api.TableSchema; import org.apache.flink.table.connector.source.DynamicTableSource; import org.apache.flink.table.factories.DynamicTableSourceFactory; import org.apache.flink.table.factories.FactoryUtil; import java.util.HashSet; import java.util.Set; public class CustomDynamicTableSourceFactory implements DynamicTableSourceFactory { public static final ConfigOptionString TOPIC ConfigOptions .key(topic) .stringType() .noDefaultValue(); public static final ConfigOptionInteger NUM_PARTITIONS ConfigOptions .key(num.partitions) .intType() .defaultValue(1); Override public String factoryIdentifier() { return custom; } Override public SetConfigOption? requiredOptions() { final SetConfigOption? options new HashSet(); options.add(TOPIC); return options; } Override public SetConfigOption? optionalOptions() { final SetConfigOption? options new HashSet(); options.add(NUM_PARTITIONS); return options; } Override public DynamicTableSource createDynamicTableSource(Context context) { final FactoryUtil.TableFactoryHelper helper FactoryUtil.createTableFactoryHelper(this, context); helper.validate(); final String topic helper.getOptions().get(TOPIC); final Integer partitions helper.getOptions().get(NUM_PARTITIONS); return new CustomDynamicTableSource(context.getCatalogTable().getResolvedSchema(), topic, partitions); } }这里有个重要点FactoryUtil.createTableFactoryHelper的validate()会帮你检查代码里requiredOptions()和optionalOptions()声明的选项。如果用户 DDL 里漏掉了必填参数比如topic在这里就会被拦截并抛出清晰异常。这是 Flink 1.11 之后推荐的雏形写法自己手动解析 Properties 虽然也可以但会丢失校验这种免费的保护。factoryIdentifier()返回值就是connector参数的值。有些老项目会直接把requiredContext()返回的 map 中加一个connectorkey那种做法在旧版可用新版本中已经被factoryIdentifier()取缔。如果你照着老文章写可能会在 1.17 上出现编译错误。2.4 DynamicTableSource 的实现细节CustomDynamicTableSource需要实现DynamicTableSource并至少实现getScanRuntimeProvider(ScanContext context)方法。在这里要完成最有技术含量的工作把 SQL 的表结构映射成一行行的数据再发到下游。表结构来自ResolvedSchema在运行时可以这样拿到物理字段、类型信息public class CustomDynamicTableSource implements DynamicTableSource { private final ResolvedSchema schema; private final String topic; private final int numPartitions; public CustomDynamicTableSource(ResolvedSchema schema, String topic, int numPartitions) { this.schema schema; this.topic topic; this.numPartitions numPartitions; } Override public ScanRuntimeProvider getScanRuntimeProvider(ScanContext context) { final DataType physicalDataType schema.toPhysicalRowDataType(); final TypeInformationRowData typeInfo context.createTypeInformation(physicalDataType); // 构造一个 SourceFunction SourceFunctionRowData sourceFunction new CustomSourceFunction(topic, numPartitions, physicalDataType); return SourceFunctionProvider.of(sourceFunction, true); } Override public DynamicTableSource copy() { return new CustomDynamicTableSource(schema, topic, numPartitions); } Override public String asSummaryString() { return Custom Data Source; } }这里特别强调两件事第一copy()方法必须实现且返回一个独立的新对象。Flink 在规划作业时会把 source 对象复制到多个地方如果浅拷贝引用相同状态会导致运行时并发问题。初学者很容易漏掉这个方法导致后续扫表时出现各种奇怪的对象共享错误。第二SourceFunctionProvider.of(sourceFunction, true)的第二个参数是“是否为有界流”。如果你的数据源是可结束的例如读取一个文件列表、读固定的测试记录应该传true这样 Flink 可以把作业作为流模式执行同时允许批流混跑。如果传false表示无限流任务永远不会正常结束。对于自定义模拟数据源建议实现一个可配置是否无限循环的选项方便测试。2.5 数据格式转换从业务对象到 RowData这一步是自定义数据源最“劝退”的地方。Flink 内部处理的数据结构是RowData不是传统的 Java Bean 或者 Map。如果直接让 SourceFunction 发一个普通对象Runtime 会在序列化和 codegen 时找不到类型经常报Cannot determine type for object或者class not registered之类的错误。比较适合新手的方式是使用GenericRowData。从业务数据到GenericRowData的过程本质是按你 DDL 里定义的字段顺序逐个赋值。假如 DDL 是id INT, name STRING, event_time TIMESTAMP_LTZ(3)那你在 source 中解析出一条完整数据后应该构造GenericRowData row new GenericRowData(3); row.setField(0, userId); row.setField(1, name); row.setField(2, TimestampData.fromEpochMillis(tsMillis));如果不处理时间戳字段而是直接放一个long那么下游就无法把它解析成 TIMESTAMP 类型运行时会报 “expect TIMESTAMP_LTZ but get BIGINT” 的转换错误。所以请牢记表 schema 的TIMESTAMP_LTZ对应TimestampData不是毫秒数或java.sql.Timestamp。如果 schema 中有DECIMAL那要放DecimalData。ARRAY、MAP等复杂类型也都有对应的 Internal 结构需要查文档。使用DataType的转换工具类可以减少一部分手工转化压力但理解映射规则依然是基本功。2.6 用 SourceFunction 发送数据时的并行度设计SourceFunction 本身有一个很重要的问题它不是分区概念而是整个 source 算子对应几个并行子任务取决于调用SourceFunctionProvider.of后 Flink 分配给它的并行度。默认我们会让用户在 DDL 的WITH里额外配一个source.parallelism然后创建 Table Source 时把这个参数传给运行时。在getScanRuntimeProvider里设置并行度不是直接设置 SourceFunction而是通过SourceFunctionProvider.of(sourceFunction, false, parallelism)这个重载方法传入。这个 parallelism 通常通过context.getConfiguration()读取不过最简单的是在 Factory 解析时把source.parallelism作为 optional option 保存下来。关于并行度的设计有一个非常常见的坑自定义 SourceFunction 里如果使用了ScheduledExecutorService之类的资源当并行度大于 1 时每个并行实例都会创建独立的线程池如果这个线程池没有在cancel()里释放任务取消时会留下孤儿线程。轻则报警告重则导致资源泄漏。因此凡是自定义 SourceFunction 里的资源都需要在cancel()方法中优雅停止在run()的循环里检查running标志保证线程可以被中断。3. 实操过程从零实现一个可运行的“模拟数据源连接器”3.1 先明确要做什么这里我用一个非常贴近日常开发痛点的例子做一个名为custom的模拟数据源它不连接任何外部系统而是直接从内部配置的常量数组里循环读数据。这个场景在测试 FLink SQL 任务、演示 Demo 时太常用了。官方缺一个“内置模拟数据源”除非你去下 flink-test-utils生产环境又不想引入测试包。自己写一个只要二十来行核心代码。我们的输入参数是topic自定义的标识用于区分不同的测试数据集合。num.partitions模拟分区数控制 source 并行度。records-per-second每秒发送多少条数据用于生成压力避免真实测试时瞬间冲完。loop是否无限循环发送。如果是批计算场景设置为 false。DDL 形如CREATE TABLE user_log ( uid BIGINT, name STRING, action STRING, event_time TIMESTAMP_LTZ(3) ) WITH ( connector custom, topic user-log, num.partitions 2, records-per-second 100, loop true );这样用户只需要在 SQL 客户端里建表就能跑select * from user_log where action buy完全不需要修改代码。3.2 开发目录结构建议Maven 工程下建议这样组织代码保持清晰src/main/java/ └── com/example/flink/connector ├── CustomDynamicTableSourceFactory.java ├── CustomDynamicTableSource.java ├── CustomSourceFunction.java └── RowDataGenerator.java src/main/resources/ └── META-INF/services/ └── org.apache.flink.table.factories.FactoryRowDataGenerator负责按 schema 生成模拟数据。CustomSourceFunction则把RowDataGenerator产出的数据包装成一个带背压的流。很多人会想直接在 SourceFunction 的run()里写一大段 while 循环生成数据搞得类非常臃肿。我们把“生成”和“发送”分开后面接真实数据源时替换也比较方便。3.3 最关键的一行注册服务文件META-INF/services/org.apache.flink.table.factories.Factory这个文件非常容易被忽视。它的内容就是一行com.example.flink.connector.CustomDynamicTableSourceFactory注意文件是 UTF-8 无 BOM 编码不要有前后空格换行符用 LF。如果项目里还实现了其他 Factory比如 Sink Factory那就换行继续加全限定名。不要吐槽这个机制老JDK 的 SPI 就这么设计的Flink 也沿用了。只要这个文件不对后面必然报No factory found for connector custom。曾经我见到一个同事把这个服务文件放到了src/main/java目录下面没有放到 resources结果 IDEA 里还能跑打包到集群后一直报找不到工厂。这就是因为 maven 默认不会把 java 目录里的.txt文件当资源打进 jar。编译后用jar tf查看包内路径确认存在META-INF/services/org.apache.flink.table.factories.Factory才是真的成功。3.4 实现 RowDataGenerator这个类很像一个“测试数据工厂”。实际工作中数据可能是从某个 WebSocket 收到的 JSON或从自研的消息中间件 poll 出来的 byte[]。我们把核心的编码逻辑放进这个类里。为了演示我直接生成 10 条固定样本然后循环。public class RowDataGenerator { private final int fieldCount; private final long initTimestamp; private final String[] actions {click, buy, addCart}; private final Random random new Random(); public RowDataGenerator(int fieldCount) { this.fieldCount fieldCount; this.initTimestamp System.currentTimeMillis(); } public RowData nextRow() { GenericRowData row new GenericRowData(fieldCount); // 仅针对 user_log 示例写死字段映射 row.setField(0, random.nextLong()); row.setField(1, user_ random.nextInt(1000)); row.setField(2, actions[random.nextInt(actions.length)]); row.setField(3, TimestampData.fromEpochMillis(initTimestamp random.nextInt(5000))); return row; } }在实际项目中fieldCount 应该来自ResolvedSchema的字段数量。这里为了讲解直接写死了字段。如果你是写通用 Connector建议把字段映射抽象成一个接口根据 schema 动态分配。否则换一张表结构就要改代码就不叫 Connector 了。3.5 实现 CustomSourceFunctionSourceFunction 的核心方法是run(SourceContext ctx)。我们在里面开一个 Async按每秒 rate 来控制发送节奏。最关键的细节是ctx.collect(row)是线程安全的但如果你通过collect方法发送它会直接进入下游不经过任何缓冲区所以你在run()方法里 sleep 并不影响其他并行子任务。实现如下public class CustomSourceFunction extends RichSourceFunctionRowData { private final int fieldCount; private final int recordsPerSecond; private final boolean loop; private volatile boolean running true; private transient RowDataGenerator generator; public CustomSourceFunction(int fieldCount, int recordsPerSecond, boolean loop) { this.fieldCount fieldCount; this.recordsPerSecond recordsPerSecond; this.loop loop; } Override public void open(Configuration parameters) { generator new RowDataGenerator(fieldCount); } Override public void run(SourceContextRowData ctx) throws Exception { long interval 1000 / Math.max(recordsPerSecond, 1); int count 0; while (running) { ctx.collect(generator.nextRow()); count; if (!loop count 10) { break; } Thread.sleep(interval); } } Override public void cancel() { running false; } }你可能发现这里的interval只是简单除以数量实际上这种控制流速的方式不够精确。如果想要精确 QPS建议记录startTime和expectedEmitCount用while (running expectedTime currentTime)去 sleep。更简单的方式是用一个RateLimiter或者Flink自带的WatermarkStrategy机制。不过初学者先跑通第一版就好。从原理上讲其实run()方法自己 sleep 会严重限制吞吐如果你想做高吞吐测试可以把 recordsPerSecond 设置得极大比如 1000000睡眠时间就会趋近于零然后整条 pipeline 的瓶颈就转移到下游算子和 Kafka 等 Sink。再说一个很实际的技巧RichSourceFunction的getRuntimeContext()能拿到当前并行子任务的编号。numPartitions参数设置为 2 时Source 并行度是 2那么每个子任务生成的测试数据应该尽量分散避免相同 id。你可以用getRuntimeContext().getIndexOfThisSubtask()拼接到name字段中这样下游透视时能看出数据来自哪个并行通道。3.6 如何在 SQL 客户端里跑通验证将工程打包为 jar不需要包含 flink-table 相关provided 依赖即可。启动本地 Flink SQL Client/opt/flink/bin/sql-client.sh embedded \ -j target/custom-connector-1.0-SNAPSHOT.jar \ -s default然后执行建表加查询CREATE TABLE user_log ( uid BIGINT, name STRING, action STRING, event_time TIMESTAMP_LTZ(3), WATERMARK FOR event_time AS event_time - INTERVAL 5 SECOND ) WITH ( connector custom, topic user-log, num.partitions 2, records-per-second 5, loop true ); SELECT action, COUNT(*) FROM user_log GROUP BY action;如果一切正常你会看到每秒大约有 5 条数据并行度 2 时其实是两个子任务合起来 10 条流进聚合算子。这里有个值得玩味的点当并行度大于 1 时records-per-second的总吞吐是单并行度数值乘以并行度。因为每个并行子任务都会按这个速率循环发送。如果你希望总量固定需要在 RowDataGenerator 中增加一个全局计数器机制例如AtomicLong由一个子任务分担生成或者接受近似即可。3.7 中途踩过的一个大坑ClassLoader 找不见工厂类本地用 IDEA 运行 SQL 客户端时一切正常但提交到测试集群上的 SQL Job 后一直报org.apache.flink.table.factories.Factory 找不到 custom connector检查发现 jar 包在lib/下依赖也完整。最后原因出在 Flink 的 SPI 加载逻辑使用的 ClassLoader 不是应用默认的而是与作业一起上传的动态 ClassLoader。由于我们打包时用了maven-shade-plugin并且 merge 了所有META-INF/services但是旧 Flink 版本的 SQL 作业提交方式并不会把产出的 jar 加载到与 Planner 相同的 loader 中。解决方案有几种确认你使用的是sql-client.sh标准提交而不是把连接器 jar 放到lib/后又进行 Flink run。使用FactoryUtil时加try“pom”无法自动解决最稳妥的方式是在with-plugins模式下把 jar 放在plugins/下。如果实在不行可启用classloader.resolve-order: parent-first并确保依赖幂等。不过新版更推荐child-first。这个坑现象和 SPI 文件缺失很像但实际原因是运行时上下文的 loader 不对。排查思路是先分别查看引擎日志、用户 jar 日志中Factory服务文件的 URL如果能打印出java.net.URL却仍说没有匹配那就是多模块下Factory接口类由不同类加载器加载导致“类型不相容”这属于比较恶心的类加载器隔离问题。必要时可以自定义FlinkExtensionsFilter或使用父类优先。4. 常见问题与排障技巧实录4.1 报错Could not find any factories for connector custom这是一个“经典必踩”问题。先检查三件事resources/META-INF/services/org.apache.flink.table.factories.Factory是否存在且内容包含工厂类全限定名注意不能有多余空格。jar 包内是否真的有这个文件。用jar tf查看而不是只看编译目录。工厂类是否可访问、没有依赖冲突。如果你在接口实现上漏加了implement方法也会导致NoSuchElementException。在本地 IDEA 运行时还需要确保 classpath 包含 target/classes 和 resources 目录。如果使用 fat jar 提交集群还要留意maven-shade-plugin会合并META-INF/services文件不会自动覆盖但如果你之前用了错误的IncludeResourceTransformer可能导致 File 内容丢失。所以一定要验证最终 jar。4.2 运行时报类别转换异常GenericType vs RowData出现这类问题多半是你 SourceFunction 中 collect 出去的不是RowData或者你使用的是GenericRow表格老类型而不是GenericRowData。Flink 1.11 之后 Table 运行时统一使用RowData作为内部数据格式你的 SourceFunction 泛型必须是SourceFunctionRowData如果你写成了SourceFunctionRow就会在 Planner 生成代码时发现类型不匹配。另一种可能性是你通过context.createTypeInformation(physicalDataType)创建了TypeInformationRowData但源函数实际发送的是自定义 POJO那也会出现类型不匹配。排查方式比较直接在 SourceFunction 的run()方法内部断言一下ctx.collect进来的对象类型最简单的是打日志输出对象的getClass()。如果看到 classorg.apache.flink.table.data.GenericRow而不是GenericRowData你的依赖很可能把flink-table-planner和flink-table-runtime版本混了。4.3 Schema 里定义的字段与实际生成数据不一致这可以说是自定义数据源里最高频的问题。比如 DDL 定义 4 个字段你 RowDataGenerator 只生成了前 3 个字段处理时不会立刻报错而是等到下游某个算子需要读取第 4 个字段的那一列时才抛 “field index out of range”。很多人在WHERE过滤条件中用不到第 4 个字段所以 Job 跑一天不出错一旦出现SELECT *就会挂掉。我的建议是在RowDataGenerator的构造函数中传入ResolvedSchema的fieldCount()并且在nextRow()开头就初始化为相应数量。每次组装时对缺失字段设null值宁可让它空着也不要让数组越界。批模式下一次作业因为一行越界而全量失败是很磨人的事情。4.4 Watermark 策略失效或 Event Time 无进展如果你的自定义数据源带有 event time 字段并且 DDL 里声明了 watermark但下游窗口从不触发很可能是 Source 里collect完数据后没有调用ctx.collectWithTimestamp(row, timestamp)方法。Flink SQL 中 watermark 的计算依赖记录的rowtime如果只collect(row)而不带时间戳watermark 会一直停留在初始值Long.MIN_VALUE。解决方法有两种在 SourceFunction 中调用ctx.collectWithTimestamp(row, eventTime)并配合ctx.emitWatermark(new Watermark(eventTime - 5000L))这样能把水的生成逻辑下推到 Source。在 SQL 层声明 watermark让 Flink 基于字段值自动计算。但这要求collectWithTimestamp或下游的 TimestampAssigner 能识别这一字段。通常更合理的方式是在 SourceFunction 里先把 eventTime 字段解析出来然后调用带时间戳的方法。如果发现窗口延迟触发也可能是你 sleep 时间太长导致 data 间隔较大而窗口触发的 watermark 还差一点。调小发送间隔或者调整 watermark idle 策略即可。4.5 并行度设置为什么没有生效有时你在 DDL 里配了source.parallelism 4但观察 Web UI 上 Source 并行度还是默认 1。这通常是因为在SourceFunctionProvider.of(sourceFunction, true)时没有把并行度参数传进去。如果你查过SourceFunctionProvider的源码会发现有三个参数和四个参数的重载。第四个参数就是并行度不传时使用默认值 1与用户在 env 中设置的并行度无关。正确姿势是在CustomDynamicTableSource里保存该值然后调用SourceFunctionProvider.of(sourceFunction, true, configuredParallelism);这个 configuredParallelism 来源就是解析 WITH 中的source.parallelism。对于 SQL 作业来说用户还可以在作业级设置parallelism.default但 SourceFunctionProvider 里的并行度优先级更高所以设计时要注意如果外部没配置就传 -1让 Flink 用全局并行度。4.6 打包遇到 “Transformer” 或依赖冲突自定义连接器最头疼的还是打包。因为不可避免要用到一些通讯协议的客户端比如 HTTP、WebSocket、Protobuf 等这些依赖很可能和 Flink 自带的冲突。更常见的是netty、guava、jackson版本冲突。建议直接使用maven-shade-plugin并且过滤掉META-INF/*.RSA、META-INF/*.SF等签名文件避免 SecurityException。最好只在 jar 中保留连接器自身和真实需要的第三方依赖不要把 flink-connector-* 相关的类打进去。如果你调用了官方连接器中的类那更要注意 graph 中的依赖关系通常采用compileexclude的方式处理。至于spi文件被覆盖的问题shade 插件会自动 merge只要你不主动跳过META-INF/services下的资源就没问题。给你一段常用的 shade 配置参考plugin groupIdorg.apache.maven.plugins/groupId artifactIdmaven-shade-plugin/artifactId version3.4.1/version executions execution phasepackage/phase goalsgoalshade/goal/goals configuration filters filter artifact*:*/artifact excludes excludeMETA-INF/*.SF/exclude excludeMETA-INF/*.DSA/exclude excludeMETA-INF/*.RSA/exclude /excludes /filter /filters transformers transformer implementationorg.apache.maven.plugins.shade.resource.ServicesResourceTransformer/ /transformers /configuration /execution /executions /pluginServicesResourceTransformer很关键没有它多个META-INF/services文件会互相覆盖最终只留下最后一个其他 Factory 全部失效。5. 扩展建议从 Source 走向完整 Connector5.1 如何进一步支持检查点与精确一次语义自定义 SourceFunction 最大的槽点是没有真正的“暂停/恢复-读位点”能力。如果你只是读取测试数据或一次性文件那无所谓。但如果接真实消息中间件为了在故障恢复时不丢数据就要实现CheckpointedFunction接口在snapshotState中记录当前消费到哪条消息在initializeState里从状态恢复。做法很简单在CustomSourceFunction中声明一个ListStateLong保存 offsetprivate transient ListStateLong offsetState; Override public void snapshotState(FunctionSnapshotContext context) throws Exception { offsetState.clear(); offsetState.add(currentOffset); } Override public void initializeState(FunctionInitializationContext context) throws Exception { ListStateDescriptorLong descriptor new ListStateDescriptor(offset, Long.class); offsetState context.getOperatorStateStore().getListState(descriptor); if (context.isRestored()) { for (Long value : offsetState.get()) { currentOffset value; } } }同时在run()里每次消费并发送数据后currentOffset。这种做法属于 at-least-once如果要 exactly-once 还是得上新 Source 接口加 SplitEnumerator 管理分片。但对于大多数自定义数据源at-least-once 已经够用因为下游配合幂等写效果等同于精确一次。5.2 批量读数据时建议包装成 InputFormat如果连接器是面向批模式的用 SourceFunction 并标记有界流也可以但性能和并发表现不如InputFormat。Flink 内部有InputFormatSourceFunction包装器在ScanRuntimeProvider中你可以直接返回InputFormatSourceFunction或者实现InputFormat。不过现在新架构更推荐实现Source接口然后由框架负责拆分和数据读取并发。实际项目中我更推荐先实现 SourceFunction 版本跑通业务逻辑后再渐进式替换为 FLIP-27。因为 SQL 层对 Connector 的接口封装屏蔽了大部分运行时的复杂性用户看到的是同一个动态表底层切换起来代价不大。如果你直接挑战 Source 接口在初期会陷入SourceReader.pollNext返回RecordsWithSplitIds的空转和检查点实现中调试效率很低。5.3 把自定义连接器发布到内部 Maven 仓库当连接器稳定后你不希望每个团队都重新编译。建议打成单独的 artifact发布到内部 Maven 仓库。pom 中保持provided依赖不变但建议把flink-table-common的版本号放到你使用版本的 profile 中。这样后续升级 Flink 时只需要重新适配一个小版本而不是全量重写。发布时还要注意写好LICENSE和NOTICE避免涉及第三方库的许可证问题。特别是使用了一些私有协议解析库时尽量选择宽松 license 的依赖。这个细节平时无所谓但你一旦跨部门或公司分发就会成为法务关卡。5.4 连接器开发的测试策略最后谈测试。我见过许多连接器代码看起来能工作但几乎没有单元测试。自定义连接器测试分三层单元测试测试工厂类的参数校验、requireOptions是否正确。集成测试用FlinkMiniCluster启动一个任务建表后执行简短的 SQL然后通过TableResult拉取前几条结果验证 Source 能否输出到 sink。端到端测试在sql-client下跑完整的建表与计算验证时序与 watermark 行为。写集成测试时注意不要把flink-test-utils的StreamTestEnvironment和 SQLTableEnvironment混淆一个是 DataStream 测试一个是 Table 测试二者可以共用 mini cluster 但上下文不同。如果你有时间建议加一个参数化测试把loopfalse和looptrue两种模式都覆盖到确保 check 与取消流程不会挂起。作为收尾我再分享一个个人经验如果你做真正意义上的生产级自定义连接器最好在 Factory 的asSummaryString()里返回包含所有配置项的可读字符串这样你在EXPLAIN中能看到单个 source 的属性。很多问题就是在EXPLAIN阶段发现并行度写错、source 被 planner 优化合并的。对于这个模拟数据源连接器我实际跑过一次持续 3 天的长稳测试重点观察了背压下 Source 的 sleep 是否有效、任务重启后状态是否干净。结论是只要 SourceFunction 里没有静态变量和未释放线程这个机制在 Flink SQL 生态里是相当可靠的。你也可以从这个最小实现开始逐步扩展成真正对接自研消息中间件的生产级组件这是接触 Flink 底层运行机制非常值得投入的一条路径。