71、Flink 的 Hybrid Source 详解

Hybrid Source

1.概述

Hybrid Source 解决了从异构数据源顺序读取输入以生成单个输入流的问题。

示例：从 S3 读取前几天的有界输入，然后使用 Kafka 的最新无界输入，当有界文件输入完成而不中断应用程序时 Hybrid Source 会从 FileSource 切换到 KafkaSource。

在 Hybrid Source 出现之前，需要创建一个具有多个源的拓扑结构，并由用户定义切换机制；使用 HybridSource 之后，从 DataStream API 的角度看，多个源在 Flink 作业图中显示为单个源。

需要依赖如下：

<dependency>     <groupId>org.apache.flink</groupId>     <artifactId>flink-connector-base</artifactId>     <version>1.19.0</version> </dependency> 

2.下一个源的起始位置

要在一个 Hybrid Source 中排列多个源，除最后一个源外的所有源都需要有界；因此通常需要为源分配一个开始和结束位置。

a）固定起始位置

示例：从文件中读取到预先确定的切换时间，然后继续从 Kafka 中读取，每个源都覆盖了预先已知的范围，可以像直接使用一样预先创建包含的源。

long switchTimestamp = ...; // derive from file input paths  FileSource<String> fileSource =   FileSource.forRecordStreamFormat(new TextLineInputFormat(), Path.fromLocalFile(testDir)).build();  KafkaSource<String> kafkaSource =           KafkaSource.<String>builder()                   .setStartingOffsets(OffsetsInitializer.timestamp(switchTimestamp + 1))                   .build();  HybridSource<String> hybridSource =           HybridSource.builder(fileSource)                   .addSource(kafkaSource)                   .build(); 

b）动态其实位置

示例：文件源需要读取的数据量很大，可能比下一个源可用的保留时间更长，切换需要在 “当前时间-X” 发生。

因此要将下一个源的启动时间设置为切换时间，需要从以前的文件枚举器中转移结束位置，以便通过实现 SourceFactory 来延迟构建KafkaSource。

注意：枚举器需要支持获取结束时间戳。

FileSource<String> fileSource = CustomFileSource.readTillOneDayFromLatest();  HybridSource<String> hybridSource =     HybridSource.<String, CustomFileSplitEnumerator>builder(fileSource)         .addSource(             switchContext -> {               CustomFileSplitEnumerator previousEnumerator =                   switchContext.getPreviousEnumerator();                              // how to get timestamp depends on specific enumerator               long switchTimestamp = previousEnumerator.getEndTimestamp();                              KafkaSource<String> kafkaSource =                   KafkaSource.<String>builder()                       .setStartingOffsets(OffsetsInitializer.timestamp(switchTimestamp + 1))                       .build();                              return kafkaSource;             },             Boundedness.CONTINUOUS_UNBOUNDED)         .build();