Flink 1.19 自定义 Source 开发实战基于 SplitReader API 实现高吞吐 HTTP 数据源1. 为什么需要自定义 Source在现代数据架构中企业往往需要从各种异构系统中获取数据。虽然 Flink 提供了丰富的内置连接器如 Kafka、文件系统等但在实际业务场景中我们经常遇到以下需求需要从私有协议或非标准 REST API 获取数据要求对数据拉取过程进行细粒度控制如分页、限流需要实现特殊的分片Split策略以提高并行度要求处理复杂的认证和错误恢复机制这正是自定义 Source 的用武之地。Flink 1.19 的 SplitReader API 提供了一套高效抽象让我们能够专注于业务逻辑而非底层并发控制。2. HTTP 数据源架构设计2.1 核心组件交互一个完整的自定义 Source 包含三个关键组件[SplitEnumerator] ←协调→ [SourceReader] ↑ ↑ 生成和管理Split 通过SplitReader消费数据对于 HTTP 数据源我们这样设计Split表示一个独立的数据分片包含public class HttpSourceSplit implements SourceSplit { private final String splitId; private final String endpoint; private final int page; private final int pageSize; // 分页参数、认证信息等 }SplitEnumerator负责初始分片生成如根据日期范围分片动态分片发现如检测新数据分片再平衡当 TaskManager 增减时SourceReader使用 SplitReader API 实现多线程并发获取数据背压处理分片级水位线管理2.2 并发模型选择Flink 提供了两种线程模型模型特点适用场景FixedSizeFetcher固定数量线程均匀分配分片分片处理耗时均匀的场景DynamicFetcher动态调整线程优先分配未处理分片分片处理耗时差异大的场景对于 HTTP API 分页场景推荐使用FixedSizeSplitFetcherManager因为每个分页请求耗时相对稳定避免动态调整带来的开销3. 完整实现步骤3.1 基础结构搭建首先定义 Source 入口类public class HttpSource implements SourceString, HttpSourceSplit, Void { private final String baseUrl; private final int pageSize; Override public Boundedness getBoundedness() { return Boundedness.BOUNDED; // 或 UNBOUNDED } Override public SplitEnumeratorHttpSourceSplit, Void createEnumerator( SplitEnumeratorContextHttpSourceSplit enumContext) { return new HttpSplitEnumerator(enumContext, baseUrl, pageSize); } // 其他必要方法... }3.2 实现 SplitEnumerator核心是分片生成策略。假设我们按时间范围分页public class HttpSplitEnumerator implements SplitEnumeratorHttpSourceSplit, Void { Override public void start() { // 初始分片生成 ListHttpSourceSplit splits generateInitialSplits(); // 分配给Reader enumContext.assignSplits(new SplitsAssignment(assignToReaders(splits))); } private ListHttpSourceSplit generateInitialSplits() { // 示例按天生成分片 return IntStream.range(0, totalPages) .mapToObj(i - new HttpSourceSplit( split- i, baseUrl, i, pageSize )).collect(Collectors.toList()); } }3.3 实现 SplitReader这是性能关键部分需要处理分页获取实现 HTTP 客户端逻辑背压处理合理控制请求速率错误重试对临时故障的容错public class HttpSplitReader implements SplitReaderString, HttpSourceSplit { private final HttpClient client; private HttpSourceSplit currentSplit; private int currentRecordIndex; private ListString currentPage; Override public RecordsWithSplitIdsString fetch() throws IOException { if (needFetchNextPage()) { currentPage fetchPage(currentSplit); currentRecordIndex 0; } String record currentPage.get(currentRecordIndex); return new RecordsWithSplitIdsString() { Override public String nextSplit() { return currentSplit.splitId(); } Override public String nextRecordFromSplit() { return record; } // 其他必要方法... }; } private ListString fetchPage(HttpSourceSplit split) { // 实现HTTP请求和响应解析 HttpRequest request HttpRequest.newBuilder() .uri(URI.create(buildPageUrl(split))) .build(); HttpResponseString response client.send( request, HttpResponse.BodyHandlers.ofString()); return parseResponse(response.body()); } }3.4 配置并发度通过FixedSizeSplitFetcherManager配置并发线程数public class HttpSourceReader extends SourceReaderBaseString, HttpSourceSplit { public HttpSourceReader(SourceReaderContext context) { super( () - new HttpSplitReader(), new FixedSizeSplitFetcherManager( context.getConfiguration().getInteger( SourceConfig.NUM_FETCHERS), new LinkedBlockingQueue(), HttpSplitReader::new), new HttpRecordEmitter(), context.getConfiguration(), context); } // 实现必要方法... }在作业中使用时HttpSource source new HttpSource(https://api.example.com/data, 100); DataStreamString stream env.fromSource( source, WatermarkStrategy.noWatermarks(), HTTP Source ).setParallelism(4); // 控制Reader并行度4. 高级优化技巧4.1 背压处理策略当 HTTP 服务响应变慢时我们需要动态调整请求速率public class AdaptiveRateLimiter { private volatile double currentRate; public void onSuccess(long latency) { currentRate Math.min(maxRate, currentRate * 1.1); } public void onError() { currentRate Math.max(minRate, currentRate * 0.5); } }在 SplitReader 中应用Override public RecordsWithSplitIdsString fetch() { rateLimiter.acquirePermit(); // ...执行请求 }4.2 分片级水位线对于时间敏感数据实现分片级水位线对齐public class HttpRecordEmitter implements RecordEmitterString, String, HttpSplitState { Override public void emitRecord( String record, SourceOutputString output, HttpSplitState splitState) { long timestamp extractTimestamp(record); output.collect(record, timestamp); // 更新分片水位线 splitState.updateWatermark(timestamp); } }4.3 检查点与恢复确保分片状态可序列化public class HttpSplitState implements Serializable { private final HttpSourceSplit split; private long watermark; private int currentPage; // 实现序列化方法... }在 SplitEnumerator 中处理检查点Override public void addSplitsBack(ListHttpSourceSplit splits, int subtaskId) { // 将未确认的分片重新分配 pendingSplits.addAll(splits); }5. 性能调优实战5.1 关键配置参数参数建议值说明num-fetchersCPU核心数×2获取线程数http.connection.timeout30s连接超时http.read.timeout60s读取超时fetch.queue.capacity1000获取队列大小max.retries3最大重试次数5.2 监控指标通过 Flink Metrics 暴露关键指标public class HttpSplitReader { private final Counter fetchSuccess; private final Counter fetchFailures; private final Histogram latencyHistogram; public HttpSplitReader(SourceReaderContext context) { this.fetchSuccess context.metricGroup() .counter(fetchSuccess); // 初始化其他指标... } private ListString fetchPage(HttpSourceSplit split) { long start System.currentTimeMillis(); try { ListString result doFetch(split); latencyHistogram.update(System.currentTimeMillis() - start); fetchSuccess.inc(); return result; } catch (Exception e) { fetchFailures.inc(); throw e; } } }5.3 与内置连接器对比特性自定义HTTP SourceFlink Kafka Connector吞吐量中等受HTTP限制高延迟较高低精确一次语义可支持原生支持动态分片灵活支持固定分区适用场景私有API集成消息队列集成6. 生产环境注意事项认证安全public class SecureHttpClient { private final SSLContext sslContext; public SecureHttpClient(String certPath) { this.sslContext createSSLContext(certPath); } private SSLContext createSSLContext(String certPath) { // 加载证书等安全配置 } }限流熔断public class CircuitBreaker { private final int failureThreshold; private int consecutiveFailures; public void execute(Runnable operation) { if (consecutiveFailures failureThreshold) { throw new CircuitBreakerOpenException(); } try { operation.run(); consecutiveFailures 0; } catch (Exception e) { consecutiveFailures; throw e; } } }日志与调试为每个分片分配唯一ID便于追踪记录分片分配和完成情况使用MDC实现请求链路追踪7. 扩展应用场景7.1 增量同步模式通过保存分片状态实现增量获取public class HttpSplitEnumerator { private final MapString, Long splitWatermarks new HashMap(); Override public Void snapshotState(long checkpointId) { return serializeState(splitWatermarks); } Override public void notifyCheckpointComplete(long checkpointId) { // 持久化水位线位置 persistWatermarks(); } }7.2 混合数据源结合多个API端点public class HybridSource extends SourceString, HybridSplit, Void { private final ListSource subSources; Override public SplitEnumeratorHybridSplit, Void createEnumerator(...) { return new HybridSplitEnumerator( subSources.stream() .map(src - src.createEnumerator(...)) .collect(Collectors.toList()) ); } }7.3 数据预处理在 Source 端进行初步过滤public class FilteringHttpSplitReader extends HttpSplitReader { Override public RecordsWithSplitIdsString fetch() { RecordsWithSplitIdsString records super.fetch(); return new FilteredRecords(records, this::filterRecord); } private boolean filterRecord(String record) { // 实现过滤逻辑 return record.contains(important); } }