您现在的位置是:首页 > 正文

Flink1.14 Source概念入门讲解与源码解析(一) 最新发布

2024-02-01 01:37:10阅读 1

目录

Flink Source概念

Source

Source源码

getBoundedness()

createReader(SourceReaderContext readerContext)

createEnumerator(SplitEnumeratorContext enumContext)

SplitEnumerator restoreEnumerator(SplitEnumeratorContext enumContext, EnumChkT checkpoint) throws Exception,>

SimpleVersionedSerializer getSplitSerializer()

SimpleVersionedSerializer getEnumeratorCheckpointSerializer()

总结

参考


Flink Source概念

Flink的Source主要是由3个核心部分组成:Splits,SplitEnumerator,SourceReader。

  • Split:split是数据源的一部分切片数据,source端将数据进行切片分发,可以并行去读取数据,而split就是一个切片粒度,一般每一次每个slot读取一个split进行处理。
  • SplitEnumerator:SplitEnumerator是一个单例只产生在JobManager中,产生split切片并且分发给sourceReader(TM里面),主要负责负载均衡,维持等待中的split的积压平衡,并且分发split给source Reader。
  • SourceReader:请求split文件并且进行处理,sourceReader是并行运行在TM的source算子中,并且产出并行的时间流/记录流。

(放一下官网的图。。。)

Flink作为批流一体的架构,Data Source API支持数据文件是无界流或者是有界批文件。

对于有界批文件来说,enumerator会产生一系列的split文件,并且每一个split文件明确是有限大小的;而对于无界流来说,则有两种情况 1)splits文件是无限的 2)enumerator不断的产生新的split文件

具体说明一下:

有界文件

Source数据源存在一个URI/Path路径,并且有固定的format去明确如何解析文件。

  • 一个split切片是一个文件,或者是多个文件(一个区域内)。
  • SplitEnumerator会列出目录下所有的文件,当下一个reader需要split切片文件的时候,就会将下一个split发送过去,一旦所有的文件全部发送完成,那么就会发出一个 NoMoreSplits 的标志。
  • SourceReader请求一个split切片,然后读取解析得到的split文件,如果没有更多的split文件后,即收到了 NoMoreSplits 那么就会停止读取。

有界Kafka

同理,只不过每一个split是一个明确的topic分区的end offset。一旦sourceReader达到了end offset,就会完成这个split文件的读取。当所有的split文件完成后,sourceReader就会结束。

无界文件流

无界的情况下,将永远不会产生 NoMoreSplits 的标志,会周期性监控URI/Path路径下是否会产生新的文件。一旦产生了新文件则会生成新的split切片并分发给可用的sourcereaders。

无界Kafka

Source数据源是一个Kafka的Topic文件,或者是一系列Topic/Topic正则。

  • Split切片文件是一个Kafka的Topic分区。
  • SplitEnumerator会连接broker,列出所有订阅的topic分区。enumerator能有选择的重复去发现订阅了的topics新增的分区数据。
  • sourcereader读取分配的split文件(topics 分区)并不会有一个end标志,所以reader永远也不会有end的情况。

Source

在1.14-1.15版本的时候source api是一个工厂模式的接口,用于创建以下的组件。

  • Split Enumerator
  • Source Reader (在1.16版本之后变为通过SourceReaderFactory接口实现
  • Split Serializer
  • Enumerator Checkpoint Serializer

除此之外,Source 还提供了 Boundedness 的特性,从而使得 Flink 可以选择合适的模式来运行 Flink 任务。

Source 实现应该是可序列化的,因为 Source 实例会在运行时被序列化并上传到 Flink 集群。

Source源码

接下来看看source的源码

import org.apache.flink.annotation.Public;
import org.apache.flink.core.io.SimpleVersionedSerializer;

import java.io.Serializable;

/**
 * The interface for Source. It acts like a factory class that helps construct the {@link
 * SplitEnumerator} and {@link SourceReader} and corresponding serializers.
 *
 * @param <T> The type of records produced by the source.
 * @param <SplitT> The type of splits handled by the source.
 * @param <EnumChkT> The type of the enumerator checkpoints.
 */
// 在flink1.16之后,source的接口变为public interface Source<T, SplitT extends SourceSplit, EnumChkT> extends SourceReaderFactory<T, SplitT>
@Public
public interface Source<T, SplitT extends SourceSplit, EnumChkT> extends Serializable {

    /**
     * Get the boundedness of this source.
     *
     * @return the boundedness of this source.
     */
    Boundedness getBoundedness();

    /**
     * Creates a new reader to read data from the splits it gets assigned. The reader starts fresh
     * and does not have any state to resume.
     *
     * @param readerContext The {@link SourceReaderContext context} for the source reader.
     * @return A new SourceReader.
     * @throws Exception The implementor is free to forward all exceptions directly. Exceptions
     *     thrown from this method cause task failure/recovery.
     */
    SourceReader<T, SplitT> createReader(SourceReaderContext readerContext) throws Exception;

    /**
     * Creates a new SplitEnumerator for this source, starting a new input.
     *
     * @param enumContext The {@link SplitEnumeratorContext context} for the split enumerator.
     * @return A new SplitEnumerator.
     * @throws Exception The implementor is free to forward all exceptions directly. * Exceptions
     *     thrown from this method cause JobManager failure/recovery.
     */
    SplitEnumerator<SplitT, EnumChkT> createEnumerator(SplitEnumeratorContext<SplitT> enumContext)
            throws Exception;

    /**
     * Restores an enumerator from a checkpoint.
     *
     * @param enumContext The {@link SplitEnumeratorContext context} for the restored split
     *     enumerator.
     * @param checkpoint The checkpoint to restore the SplitEnumerator from.
     * @return A SplitEnumerator restored from the given checkpoint.
     * @throws Exception The implementor is free to forward all exceptions directly. * Exceptions
     *     thrown from this method cause JobManager failure/recovery.
     */
    SplitEnumerator<SplitT, EnumChkT> restoreEnumerator(
            SplitEnumeratorContext<SplitT> enumContext, EnumChkT checkpoint) throws Exception;

    // ------------------------------------------------------------------------
    //  serializers for the metadata
    // ------------------------------------------------------------------------

    /**
     * Creates a serializer for the source splits. Splits are serialized when sending them from
     * enumerator to reader, and when checkpointing the reader's current state.
     *
     * @return The serializer for the split type.
     */
    SimpleVersionedSerializer<SplitT> getSplitSerializer();

    /**
     * Creates the serializer for the {@link SplitEnumerator} checkpoint. The serializer is used for
     * the result of the {@link SplitEnumerator#snapshotState()} method.
     *
     * @return The serializer for the SplitEnumerator checkpoint.
     */
    SimpleVersionedSerializer<EnumChkT> getEnumeratorCheckpointSerializer();
}

我们一个一个函数来看,毕竟一堆看上去确实感觉挺头疼的。。。。

getBoundedness()

主要是返回数据源是否有界,返回类型是Boundedness的枚举类,值只有两个BOUNDED 和 CONTINUOUS_UNBOUNDED。

具体的接口实现有四类(后面的实现都是有四类,这边只讲fileSource相关的,就不会过多介绍了。。。)

public abstract class AbstractFileSource<T, SplitT extends FileSourceSplit>
        implements Source<T, SplitT, PendingSplitsCheckpoint<SplitT>>, ResultTypeQueryable<T> {
    @Override
    public Boundedness getBoundedness() {
        return continuousEnumerationSettings == null
                ? Boundedness.BOUNDED // 有界
                : Boundedness.CONTINUOUS_UNBOUNDED; // 无界
    }
}

public class DorisSource<OUT> implements Source<OUT, DorisSourceSplit, PendingSplitsCheckpoint>, ResultTypeQueryable<OUT> {
    public Boundedness getBoundedness() {
        return this.boundedness;
    }
}

public class HybridSource<T> implements Source<T, HybridSourceSplit, HybridSourceEnumeratorState> {
    public Boundedness getBoundedness() {
        return ((HybridSource.SourceListEntry)this.sources.get(this.sources.size() - 1)).boundedness;
    }
}

public class NumberSequenceSource
        implements Source<
                        Long,
                        NumberSequenceSource.NumberSequenceSplit,
                        Collection<NumberSequenceSource.NumberSequenceSplit>>,
                ResultTypeQueryable<Long> {

    @Override
    public Boundedness getBoundedness() {
        return Boundedness.BOUNDED;
    }
}

其中,continuousEnumerationSettings主要的作用是设置轮询时间,多久去对于无界的文件进行扫描。

createReader(SourceReaderContext readerContext)

创建一个全新的source reader去读取分配给到它的splits文件,不包含任何状态恢复,返回接口SourceReader。在flink1.16的版本中已经放在了SourceReaderFactory接口中实现。

    // abstractFileSource中的实现 
    @Override
    public SourceReader<T, SplitT> createReader(SourceReaderContext readerContext) {
        // fileSourceReader是一种读取方式,从FileSourceSplit中读取记录
        return new FileSourceReader<>(
                readerContext, readerFormat, readerContext.getConfiguration());
    }

其中,readerContext是Flink运行时source的上下文;readerFormat是BulkFormat<T, SplitT>类型(BulkFormat一次读取一批次的数据并且解析),对于reader而言,BulkFormat类主要是一个工厂以及一个配置的持有者,真正读取文件的其实是 BulkFormat.Reader,这个方法是在BulkFormat类中的 createReader(Configuration, FileSourceSplit)方法创建。

createEnumerator(SplitEnumeratorContext<SplitT> enumContext)

为这个source创建新的SplitEnumerator,开始一个新的input。

    @Override
    public SplitEnumerator<SplitT, PendingSplitsCheckpoint<SplitT>> createEnumerator(
            SplitEnumeratorContext<SplitT> enumContext) {

        final FileEnumerator enumerator = enumeratorFactory.create();

        // read the initial set of splits (which is also the total set of splits for bounded
        // sources)
        final Collection<FileSourceSplit> splits;
        try {
            // TODO - in the next cleanup pass, we should try to remove the need to "wrap unchecked"
            // here
            splits = enumerator.enumerateSplits(inputPaths, enumContext.currentParallelism());
        } catch (IOException e) {
            throw new FlinkRuntimeException("Could not enumerate file splits", e);
        }

        return createSplitEnumerator(enumContext, enumerator, splits, null);
    }

其中,enumerator是由FileEnumerator工厂类产生的,这个类主要任务是找到所有需要读取的文件,切分它们成为FileSourceSplit。并且遍历路径的同时会过滤文件(如果有文件不想要读取可以通过名称进行过滤),决定是否切分文件为多个split,如何去切分的。

splits = enumerator.enumerateSplits(inputPaths, enumContext.currentParallelism());

这里则是进行切分split,里面的函数实现主要是通过递归进行遍历path。顺便提一嘴,具体实现是接口FileEnumerator的具体实现NonSplittingRecursiveEnumerator类。

    @Override
    public Collection<FileSourceSplit> enumerateSplits(Path[] paths, int minDesiredSplits)
            throws IOException {
        final ArrayList<FileSourceSplit> splits = new ArrayList<>();

        for (Path path : paths) {
            final FileSystem fs = path.getFileSystem();
            final FileStatus status = fs.getFileStatus(path);
            addSplitsForPath(status, fs, splits);
        }

        return splits;
    }

    private void addSplitsForPath(
            FileStatus fileStatus, FileSystem fs, ArrayList<FileSourceSplit> target)
            throws IOException {
        if (!fileFilter.test(fileStatus.getPath())) {
            return;
        }
        // 判断是文件还是目录,如果是文件则转化为source split去读取。
        // 比如hdfs的话,就会去获取datanode的host
        if (!fileStatus.isDir()) {
            convertToSourceSplits(fileStatus, fs, target);
            return;
        }

        final FileStatus[] containedFiles = fs.listStatus(fileStatus.getPath());
        for (FileStatus containedStatus : containedFiles) {
            // 递归遍历文件目录
            addSplitsForPath(containedStatus, fs, target);
        }
    }

最后createSplitEnumerator这个函数则是去根据是有界数据还是无界数据进行划分,如果无界数据存在alreadyProcessedPaths也会直接去划分split,如果alreadyProcessedPaths为空,才会去周期性的监控路径是否产生新文件。(后续再讲。。。)

    private SplitEnumerator<SplitT, PendingSplitsCheckpoint<SplitT>> createSplitEnumerator(
            SplitEnumeratorContext<SplitT> context,
            FileEnumerator enumerator,
            Collection<FileSourceSplit> splits,
            @Nullable Collection<Path> alreadyProcessedPaths) {

        // cast this to a collection of FileSourceSplit because the enumerator code work
        // non-generically just on that base split type
        @SuppressWarnings("unchecked")
        final SplitEnumeratorContext<FileSourceSplit> fileSplitContext =
                (SplitEnumeratorContext<FileSourceSplit>) context;

        final FileSplitAssigner splitAssigner = assignerFactory.create(splits);

        if (continuousEnumerationSettings == null) {
            // bounded case
            return castGeneric(new StaticFileSplitEnumerator(fileSplitContext, splitAssigner));
        } else {
            // unbounded case
            if (alreadyProcessedPaths == null) {
                alreadyProcessedPaths = splitsToPaths(splits);
            }

            return castGeneric(
                    new ContinuousFileSplitEnumerator(
                            fileSplitContext,
                            enumerator,
                            splitAssigner,
                            inputPaths,
                            alreadyProcessedPaths,
                            continuousEnumerationSettings.getDiscoveryInterval().toMillis()));
        }
    }

    @SuppressWarnings("unchecked")
    private SplitEnumerator<SplitT, PendingSplitsCheckpoint<SplitT>> castGeneric(
            final SplitEnumerator<FileSourceSplit, PendingSplitsCheckpoint<FileSourceSplit>>
                    enumerator) {

        // cast arguments away then cast them back. Java Generics Hell :-/
        return (SplitEnumerator<SplitT, PendingSplitsCheckpoint<SplitT>>)
                (SplitEnumerator<?, ?>) enumerator;
    }

    private static Collection<Path> splitsToPaths(Collection<FileSourceSplit> splits) {
        return splits.stream()
                .map(FileSourceSplit::path)
                .collect(Collectors.toCollection(HashSet::new));
    }

SplitEnumerator<SplitT, EnumChkT> restoreEnumerator(SplitEnumeratorContext<SplitT> enumContext, EnumChkT checkpoint) throws Exception

主要是通过一个checkpoint去恢复一个枚举器。最后调用的函数与createEnumerator只是多了一个checkpoint.getAlreadyProcessedPaths()参数传递。

    @Override
    public SplitEnumerator<SplitT, PendingSplitsCheckpoint<SplitT>> restoreEnumerator(
            SplitEnumeratorContext<SplitT> enumContext,
            PendingSplitsCheckpoint<SplitT> checkpoint) {

        final FileEnumerator enumerator = enumeratorFactory.create();

        // cast this to a collection of FileSourceSplit because the enumerator code work
        // non-generically just on that base split type
        @SuppressWarnings("unchecked")
        final Collection<FileSourceSplit> splits =
                (Collection<FileSourceSplit>) checkpoint.getSplits();

        return createSplitEnumerator(
                enumContext, enumerator, splits, checkpoint.getAlreadyProcessedPaths());
    }


SimpleVersionedSerializer<SplitT> getSplitSerializer()

主要是为source splits创建一个序列化器,在splits从enumerator到reader的时候或者是当reader进行checkpoint的时候执行。

    @Override
    public SimpleVersionedSerializer<FileSourceSplit> getSplitSerializer() {
        return FileSourceSplitSerializer.INSTANCE;
    }


@PublicEvolving
public final class FileSourceSplitSerializer implements SimpleVersionedSerializer<FileSourceSplit> {

    public static final FileSourceSplitSerializer INSTANCE = new FileSourceSplitSerializer();

SimpleVersionedSerializer<EnumChkT> getEnumeratorCheckpointSerializer()

获取SplitEnumerator checkpoint的序列化器,用于处理SplitEnumerator#snapshotState()方法返回的结果

    @Override
    public SimpleVersionedSerializer<PendingSplitsCheckpoint<SplitT>>
            getEnumeratorCheckpointSerializer() {
        return new PendingSplitsCheckpointSerializer<>(getSplitSerializer());
    }

以上就是Source的接口的所有方法,主要包含创建 SourceReaderSplitEnumerator 和对应get序列化器的方法。

总结

目前可以看出,Souce接口的更新,其实是因为Flink在1.12之前将批处理任务与流处理任务分为两种实现模式。

在底层实现中

DataSet API中Source对应的核心借口是InputFormat,功能上主要有三点:

  1. 描述输入的数据如何被划分为不同的InputSplit,继承于 INputSplitSource
  2. 描述如何从单个InputSplit读取记录,具体包括如何打开一个分配到的InputSplit,如何从这个INputSplit读取一条记录,如何得知记录已经读完和如何关闭这个Inputsplit
  3. 描述如何获取输入数据的统计信息(比如文件的大小、记录的数目)

1、3两点主要会被JobManager/JobMaster在调度Exection时使用,而第2点读取数据功能则会在运行时被TaskManager使用。

DataStream API中 Source 对应的核心接口为 SourceFunction 以及 SourceContext。前者直接继承 Function 接口与 Operator 交互,负责通用的状态管理(比如初始化或取消);后者代表运行时的上下文,负责与单条记录级别的数据的交互。此外还有其他一些辅助类型的类或接口。

运行时,Source 主要通过 SourceContext 来控制数据的输出。从 SourceContext 接口的方法即可以看出,Source 在接受到数据后的主要工作有以下几点:

  1. 从外部摄入数据或者生成数据,输出到下游
  2. 为数据生成 Event Time Timestamp(仅在 Time Characteristic 为 Event Time 时有用)
  3. 计算 Watermark 并输出(仅在 Time Characteristic 为 Event Time 时有用)
  4. 当暂时不会有新数据时将自己标记为 Idle ,以避免下游一直等待自己的 Watermark

综上所述,之前的 Source 接口并不能很好的满足批流一体的发展,所以在 FLIP-27中选择重构Source接口,新接口的核心是通过 SplitEnumerator 和 SplitReader,前者负责发现和分配 Split、触发 Checkpoint 等管理工作,后者负责 Split 的实际读取处理。此外,新增 Operator 间的通信机制(复用大部分现有的 RPC 机制),让 Source Subtask 之间可以协调完成 Event Time 对齐等新特性。最后, SplitReader 底层封装了通用的线程模型,相比之前的 SourceFunction 大大简化了 Source 的实现。

参考

漫谈 Flink Source 接口重构 | 时间与精神的小屋

Flink 源码之新 Source 架构 - 简书

数据源 | Apache Flink

网站文章

  • HTML5新特性(基本)

    HTML5的基本新特性 视频音频的加入 表单的自动验证

    2024-02-01 01:36:42
  • 卡顿掉帧问题分析之工具篇

    卡顿掉帧问题分析之工具篇

    从上面的分析可以看到,谷歌原生在Android系统的很多系统关键流程上都添加了相关的,以观察这些流程执行的耗时情况。但是这些信息覆盖的范围毕竟是有限的,有时候并不能满足我们的分析需求。比如我们开发者想...

    2024-02-01 01:36:29
  • java变c_C、Java代码如何变成计算机指令

    java - 字节码 - 虚拟机 - 机器码 - CPU执行C 语言代码 ―编译―》汇编代码 ―翻译―》机器码机器码是 CPU 能识别的语言,由 0、1 组成。一条条机器码就是一条条计算机指令。汇编代...

    2024-02-01 01:36:02
  • FairMOT理解与实现

    FairMOT理解与实现

    理解:(1)概述:多目标跟踪,单纯跟踪能力不足以完成任务,所以,不同于单目标跟踪,这里加入了检测任务,可以将多目标跟踪任务看成为目标检测+重识别任务。(2)论文网络结构:文中网络结构分3大部分:网络结...

    2024-02-01 01:35:53
  • 删除字符串中重复字符。

    题目:删除字符串中重复字符。如果可以,优先删除重复字符中排在比他小字符前面的字符。 比如,输入:bbcacdww;输出:bacdw 分析:如果根本不允许开设数组,则只能就地进行字符串去重,那么可以依次访问字符串中的字符,并删除从该字符串开始到结尾的所有相同字符。时间复杂度为O(n^2 )。void removeDuplicate(char s[]){ int i = 0; whil

    2024-02-01 01:35:46
  • VM下的Centos7安装ftp服务

    VM下的Centos7安装ftp服务

    Linux安装ftp组件1  安装vsftpd组件 [root@bogon~]# yum -y install vsftpd 2  配置vsftpd组件l  打开vsftpd配置文件/etc/vsftpd/vftpd.confl  配置文件的内容如下anonymous_enable=NO //设定不允许匿名访问local_enable=YES //设定本地用

    2024-02-01 01:35:39
  • matlab练习程序(广度优先搜索BFS、深度优先搜索DFS)

    matlab练习程序(广度优先搜索BFS、深度优先搜索DFS)

    如此经典的算法竟一直没有单独的实现过,真是遗憾啊。广度优先搜索在过去实现的二值图像连通区域标记和prim最小生成树算法时已经无意识的用到了,深度优先搜索倒是没用过。这次单独的将两个算法实现出来,因为算法本身和图像没什么关系,所以更纯粹些。广度优先搜索是从某一节点开始,搜索与其线连接的所有节点,按照广度方向像外扩展,直到不重复遍历所有节点。深度优先搜索是从某一节点开始,沿着其搜索到的...

    2024-02-01 01:34:59
  • win10 安装配置Git

    win10 安装配置Git

    3.继续,执行ssh-keygen-trsa,(注意ssh-keygen无空格),生成SSH(你的电脑与Gitee通信的安全连接)百度地址链接https//pan.baidu.com/s/1Y_P_e...

    2024-02-01 01:34:43
  • 机器学习概述

    简介什么是机器学习?机器学习就是从【数据】中自动分析,获得【规律(模型)】,并利用规律对未知数进行【预测】。样本数据(数据集)的载体- 通常情况下历史数据都不会存储在数据库中,而是存储在文件中(.cs...

    2024-02-01 01:34:37
  • Mysql 基于GTID的主从复制及切换

    参考http://imysql.com/tag/gtidhttp://mysqllover.com/?p=594Mysql 基于GTID的主从复制及切换一、主从复制配置两个mysql服务的my.cnf 中相关内容配置[mysqld]#从复制数据库表设置replicate-wild-ignore-table = mysql.%,information_schema.%,inno...

    2024-02-01 01:34:02