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 数据源，我们这样设计：

1. Split ：表示一个独立的数据分片，包含：

   public class HttpSourceSplit implements SourceSplit {
       private final String splitId;
       private final String endpoint;
       private final int page;
       private final int pageSize;
       // 分页参数、认证信息等
   }
   

2. SplitEnumerator ：负责：

   • 初始分片生成（如根据日期范围分片）
   • 动态分片发现（如检测新数据）
   • 分片再平衡（当 TaskManager 增减时）

3. SourceReader ：使用 SplitReader API 实现：

   • 多线程并发获取数据
   • 背压处理
   • 分片级水位线管理

2.2 并发模型选择

Flink 提供了两种线程模型：

模型	特点	适用场景
FixedSizeFetcher	固定数量线程，均匀分配分片	分片处理耗时均匀的场景
DynamicFetcher	动态调整线程，优先分配未处理分片	分片处理耗时差异大的场景

对于 HTTP API 分页场景，推荐使用 FixedSizeSplitFetcherManager ，因为：

• 每个分页请求耗时相对稳定
• 避免动态调整带来的开销

3. 完整实现步骤

3.1 基础结构搭建

首先定义 Source 入口类：

public class HttpSource implements Source<String, HttpSourceSplit, Void> {
    private final String baseUrl;
    private final int pageSize;
    
    @Override
    public Boundedness getBoundedness() {
        return Boundedness.BOUNDED; // 或 UNBOUNDED
    }
    
    @Override
    public SplitEnumerator<HttpSourceSplit, Void> createEnumerator(
        SplitEnumeratorContext<HttpSourceSplit> enumContext) {
        return new HttpSplitEnumerator(enumContext, baseUrl, pageSize);
    }
    
    // 其他必要方法...
}

3.2 实现 SplitEnumerator

核心是分片生成策略。假设我们按时间范围分页：

public class HttpSplitEnumerator implements SplitEnumerator<HttpSourceSplit, Void> {
    @Override
    public void start() {
        // 初始分片生成
        List<HttpSourceSplit> splits = generateInitialSplits();
        // 分配给Reader
        enumContext.assignSplits(new SplitsAssignment<>(assignToReaders(splits)));
    }
    
    private List<HttpSourceSplit> generateInitialSplits() {
        // 示例：按天生成分片
        return IntStream.range(0, totalPages)
            .mapToObj(i -> new HttpSourceSplit(
                "split-" + i,
                baseUrl,
                i,
                pageSize
            )).collect(Collectors.toList());
    }
}

3.3 实现 SplitReader

这是性能关键部分，需要处理：

1. 分页获取 ：实现 HTTP 客户端逻辑
2. 背压处理 ：合理控制请求速率
3. 错误重试 ：对临时故障的容错

public class HttpSplitReader implements SplitReader<String, HttpSourceSplit> {
    private final HttpClient client;
    private HttpSourceSplit currentSplit;
    private int currentRecordIndex;
    private List<String> currentPage;
    
    @Override
    public RecordsWithSplitIds<String> fetch() throws IOException {
        if (needFetchNextPage()) {
            currentPage = fetchPage(currentSplit);
            currentRecordIndex = 0;
        }
        
        String record = currentPage.get(currentRecordIndex++);
        return new RecordsWithSplitIds<String>() {
            @Override
            public String nextSplit() {
                return currentSplit.splitId();
            }
            
            @Override
            public String nextRecordFromSplit() {
                return record;
            }
            
            // 其他必要方法...
        };
    }
    
    private List<String> fetchPage(HttpSourceSplit split) {
        // 实现HTTP请求和响应解析
        HttpRequest request = HttpRequest.newBuilder()
            .uri(URI.create(buildPageUrl(split)))
            .build();
        
        HttpResponse<String> response = client.send(
            request, HttpResponse.BodyHandlers.ofString());
        
        return parseResponse(response.body());
    }
}

3.4 配置并发度

通过 FixedSizeSplitFetcherManager 配置并发线程数：

public class HttpSourceReader extends SourceReaderBase<String, 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);
DataStream<String> stream = env.fromSource(
    source,
    WatermarkStrategy.noWatermarks(),
    "HTTP Source"
).setParallelism(4); // 控制Reader并行度

4. 高级优化技巧

4.1 背压处理策略

当 HTTP 服务响应变慢时，我们需要：

1. 动态调整请求速率 ：

   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);
       }
   }
   

2. 在 SplitReader 中应用 ：

   @Override
   public RecordsWithSplitIds<String> fetch() {
       rateLimiter.acquirePermit();
       // ...执行请求
   }
   

4.2 分片级水位线

对于时间敏感数据，实现分片级水位线对齐：

public class HttpRecordEmitter implements RecordEmitter<String, String, HttpSplitState> {
    @Override
    public void emitRecord(
        String record,
        SourceOutput<String> 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(List<HttpSourceSplit> splits, int subtaskId) {
    // 将未确认的分片重新分配
    pendingSplits.addAll(splits);
}

5. 性能调优实战

5.1 关键配置参数

参数	建议值	说明
num-fetchers	CPU核心数×2	获取线程数
http.connection.timeout	30s	连接超时
http.read.timeout	60s	读取超时
fetch.queue.capacity	1000	获取队列大小
max.retries	3	最大重试次数

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 List<String> fetchPage(HttpSourceSplit split) {
        long start = System.currentTimeMillis();
        try {
            List<String> result = doFetch(split);
            latencyHistogram.update(System.currentTimeMillis() - start);
            fetchSuccess.inc();
            return result;
        } catch (Exception e) {
            fetchFailures.inc();
            throw e;
        }
    }
}

5.3 与内置连接器对比

特性	自定义HTTP Source	Flink Kafka Connector
吞吐量	中等（受HTTP限制）	高
延迟	较高	低
精确一次语义	可支持	原生支持
动态分片	灵活支持	固定分区
适用场景	私有API集成	消息队列集成

6. 生产环境注意事项

1. 认证安全 ：

   public class SecureHttpClient {
       private final SSLContext sslContext;
       
       public SecureHttpClient(String certPath) {
           this.sslContext = createSSLContext(certPath);
       }
       
       private SSLContext createSSLContext(String certPath) {
           // 加载证书等安全配置
       }
   }
   

2. 限流熔断 ：

   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;
           }
       }
   }
   

3. 日志与调试 ：

   • 为每个分片分配唯一ID便于追踪
   • 记录分片分配和完成情况
   • 使用MDC实现请求链路追踪

7. 扩展应用场景

7.1 增量同步模式

通过保存分片状态实现增量获取：

public class HttpSplitEnumerator {
    private final Map<String, 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 Source<String, HybridSplit, Void> {
    private final List<Source> subSources;
    
    @Override
    public SplitEnumerator<HybridSplit, Void> createEnumerator(...) {
        return new HybridSplitEnumerator(
            subSources.stream()
                .map(src -> src.createEnumerator(...))
                .collect(Collectors.toList())
        );
    }
}

7.3 数据预处理

在 Source 端进行初步过滤：

public class FilteringHttpSplitReader extends HttpSplitReader {
    @Override
    public RecordsWithSplitIds<String> fetch() {
        RecordsWithSplitIds<String> records = super.fetch();
        return new FilteredRecords(records, this::filterRecord);
    }
    
    private boolean filterRecord(String record) {
        // 实现过滤逻辑
        return record.contains("important");
    }
}