Apache Flink流分区器剖析

本文涉及的产品
实时计算 Flink 版,5000CU*H 3个月
简介: 这篇文章介绍Flink的分区器,在流进行转换操作后,Flink通过分区器来精确得控制数据流向。 StreamPartitioner StreamPartitioner是Flink流分区器的基类,它只定义了一个抽象方法: public abstract StreamPartition...

这篇文章介绍Flink的分区器,在流进行转换操作后,Flink通过分区器来精确得控制数据流向。

StreamPartitioner

StreamPartitioner是Flink流分区器的基类,它只定义了一个抽象方法:

    public abstract StreamPartitioner<T> copy();

但这个方法并不是各个分区器之间互相区别的地方,定义不同的分区器的核心在于——各个分区器需要实现channel选择的接口方法:

    int[] selectChannels(T record, int numChannels);

该方法针对当前的record以及所有的channel数目,返回一个针对当前这条记录采用的output channel的索引数组。(注意这里返回的是数组,说明一个记录可能会输出到多个channel这点我们后面会谈到)。

该接口方法来自于StreamPartitioner实现的接口ChannelSelector

分区器整体类图:

flink-data-stream-partitioner_class-diagram

GlobalPartitioner

全局分区器,其实现很简单——默认选择了索引为0的channel进行输出。

private int[] returnArray = new int[] { 0 };

@Override
public int[] selectChannels(SerializationDelegate<StreamRecord<T>> record,int numberOfOutputChannels) {
    return returnArray;
}

ForwardPartitioner

该分区器将记录转发给在本地运行的下游的(归属于subtask)的operattion。其实现跟上面的GlobalPartitioner一致,就不贴代码了。

ShufflePartitioner

混洗分区器,该分区器会在所有output channel中选择一个随机的进行输出。

private int[] returnArray = new int[1];

@Override
public int[] selectChannels(SerializationDelegate<StreamRecord<T>> record,int numberOfOutputChannels) {
    returnArray[0] = random.nextInt(numberOfOutputChannels);
    return returnArray;
}

HashPartitioner

hash分区器,该分区器对key进行hash后计算得到channel索引。它通过构造器获得KeySelector的实例(该实例用来获取当前记录的key)。

获得key后,通过其hashcodenumberOfOutputChannels取模后计算得出最终输出的channel的索引。

    public int[] selectChannels(SerializationDelegate<StreamRecord<T>> record,
            int numberOfOutputChannels) {
        Object key;
        try {
            key = keySelector.getKey(record.getInstance().getValue());
        } catch (Exception e) {
            throw new RuntimeException("Could not extract key from " + record.getInstance().getValue(), e);
        }
        returnArray[0] = MathUtils.murmurHash(key.hashCode()) % numberOfOutputChannels;

        return returnArray;
    }

BroadcastPartitioner

广播分区器,用于将该记录广播给下游的所有的subtask。这里采用了两个标记:

  • set
  • setNumber
    public int[] selectChannels(SerializationDelegate<StreamRecord<T>> record,
            int numberOfOutputChannels) {
        if (set && setNumber == numberOfOutputChannels) {
            return returnArray;
        } else {
            this.returnArray = new int[numberOfOutputChannels];
            for (int i = 0; i < numberOfOutputChannels; i++) {
                returnArray[i] = i;
            }
            set = true;
            setNumber = numberOfOutputChannels;
            return returnArray;
        }
    }

从上面的实现可见,它返回了一个跟numberOfOutputChannels相等的数组(数组的大小就是即将输出到channel的个数)。

RebalancePartitioner

重平衡分区器,用于实现类似于round-robin这样的轮转模式的分区器。通过累加、取模的形式来实现对输出channel的切换。

    public int[] selectChannels(SerializationDelegate<StreamRecord<T>> record,
            int numberOfOutputChannels) {
        this.returnArray[0] = (this.returnArray[0] + 1) % numberOfOutputChannels;
        return this.returnArray;
    }

RescalePartitioner

也是以round-robin的形式将元素分区到下游subtask的子集中。

上游操作所发送的元素被分区到下游操作的哪些子集,依赖于上游和下游操作的并行度。例如,如果上游操作的并行度为2,而下游操作的并行度为4,那么一个上游操作会分发元素给两个下游操作,同时另一个上游操作会分发给另两个下游操作。相反的,如果下游操作的并行度为2,而上游操作的并行度为4,那么两个上游操作会分发数据给一个下游操作,同时另两个上游操作会分发数据给另一个下游操作。

在上下游的并行度不是呈倍数关系的情况下,下游操作会有数量不同的来自上游操作的输入。具体的实现代码同RebalancePartitioner

CustomPartitionerWrapper

自定义分区器包装器,该包装器封装了对于自定义的分区器的实现。自定义的分区测量依赖于Partitioner接口。它提供了自定义分区器的契约。核心接口方法是:

    /**
     * Computes the partition for the given key.
     * 
     * @param key The key.
     * @param numPartitions The number of partitions to partition into.
     * @return The partition index.
     */
    int partition(K key, int numPartitions);

该接口方法的描述很清晰,通过给定的key以及numPartitions返回partition的index.

CustomPartitionerWrapper通过构造器注入Partitioner的实例,然后在selectChannels方法中通过partition接口来获得最终的channel索引。

    public int[] selectChannels(SerializationDelegate<StreamRecord<T>> record, int numberOfOutputChannels) {

        K key = null;
        try {
            key = keySelector.getKey(record.getInstance().getValue());
        } catch (Exception e) {
            throw new RuntimeException("Could not extract key from " + record.getInstance(), e);
        }

        returnArray[0] = partitioner.partition(key,
                numberOfOutputChannels);

        return returnArray;
    }

小结

以上的这些分区器,最终会体现在DataStream的API中用来对数据流进行物理分区。




原文发布时间为:2016-06-22


本文作者:vinoYang


本文来自云栖社区合作伙伴CSDN博客,了解相关信息可以关注CSDN博客。

相关实践学习
基于Hologres轻松玩转一站式实时仓库
本场景介绍如何利用阿里云MaxCompute、实时计算Flink和交互式分析服务Hologres开发离线、实时数据融合分析的数据大屏应用。
Linux入门到精通
本套课程是从入门开始的Linux学习课程,适合初学者阅读。由浅入深案例丰富,通俗易懂。主要涉及基础的系统操作以及工作中常用的各种服务软件的应用、部署和优化。即使是零基础的学员,只要能够坚持把所有章节都学完,也一定会受益匪浅。
目录
相关文章
|
21天前
|
SQL Java API
Apache Flink 2.0-preview released
Apache Flink 社区正积极筹备 Flink 2.0 的发布,这是自 Flink 1.0 发布以来的首个重大更新。Flink 2.0 将引入多项激动人心的功能和改进,包括存算分离状态管理、物化表、批作业自适应执行等,同时也包含了一些不兼容的变更。目前提供的预览版旨在让用户提前尝试新功能并收集反馈,但不建议在生产环境中使用。
505 13
Apache Flink 2.0-preview released
|
25天前
|
存储 缓存 算法
分布式锁服务深度解析:以Apache Flink的Checkpointing机制为例
【10月更文挑战第7天】在分布式系统中,多个进程或节点可能需要同时访问和操作共享资源。为了确保数据的一致性和系统的稳定性,我们需要一种机制来协调这些进程或节点的访问,避免并发冲突和竞态条件。分布式锁服务正是为此而生的一种解决方案。它通过在网络环境中实现锁机制,确保同一时间只有一个进程或节点能够访问和操作共享资源。
59 3
|
30天前
|
分布式计算 监控 大数据
大数据-148 Apache Kudu 从 Flink 下沉数据到 Kudu
大数据-148 Apache Kudu 从 Flink 下沉数据到 Kudu
52 1
|
29天前
|
数据挖掘 物联网 数据处理
深入探讨Apache Flink:实时数据流处理的强大框架
在数据驱动时代,企业需高效处理实时数据流。Apache Flink作为开源流处理框架,以其高性能和灵活性成为首选平台。本文详细介绍Flink的核心特性和应用场景,包括实时流处理、强大的状态管理、灵活的窗口机制及批处理兼容性。无论在实时数据分析、金融服务、物联网还是广告技术领域,Flink均展现出巨大潜力,是企业实时数据处理的理想选择。随着大数据需求增长,Flink将继续在数据处理领域发挥重要作用。
|
30天前
|
存储 分布式计算 大数据
大数据-145 Apache Kudu 架构解读 Master Table 分区 读写
大数据-145 Apache Kudu 架构解读 Master Table 分区 读写
40 0
|
1月前
|
消息中间件 druid Kafka
从Apache Flink到Kafka再到Druid的实时数据传输,用于分析/决策
从Apache Flink到Kafka再到Druid的实时数据传输,用于分析/决策
67 0
|
3月前
|
存储 消息中间件 Java
Apache Flink 实践问题之原生TM UI日志问题如何解决
Apache Flink 实践问题之原生TM UI日志问题如何解决
44 1
|
2月前
|
SQL 消息中间件 关系型数据库
Apache Doris Flink Connector 24.0.0 版本正式发布
该版本新增了对 Flink 1.20 的支持,并支持通过 Arrow Flight SQL 高速读取 Doris 中数据。
|
3月前
|
消息中间件 监控 数据挖掘
基于RabbitMQ与Apache Flink构建实时分析系统
【8月更文第28天】本文将介绍如何利用RabbitMQ作为数据源,结合Apache Flink进行实时数据分析。我们将构建一个简单的实时分析系统,该系统能够接收来自不同来源的数据,对数据进行实时处理,并将结果输出到另一个队列或存储系统中。
205 2
|
3月前
|
消息中间件 分布式计算 Hadoop
Apache Flink 实践问题之Flume与Hadoop之间的物理墙问题如何解决
Apache Flink 实践问题之Flume与Hadoop之间的物理墙问题如何解决
46 3

推荐镜像

更多