共计 7382 个字符，预计需要花费 19 分钟才能阅读完成。
简介： 如何通过 Apache Pulsar 原生的存储计算拆散的架构提供批流交融的根底，以及 Apache Pulsar 如何与 Flink 联合，实现批流一体的计算。
简介：StreamNative 联结创始人翟佳在本次演讲中介绍了下一代云原生音讯流平台 Apache Pulsar，并解说如何通过 Apache Pulsar 原生的存储计算拆散的架构提供批流交融的根底，以及 Apache Pulsar 如何与 Flink 联合，实现批流一体的计算。
GitHub 地址
https://github.com/apache/flink
欢送大家给 Flink 点赞送 star~
Apache Pulsar 绝对比拟新，它于 2017 年退出 Apache 软件基金会，2018 年才从 Apache 软件基金会毕业并成为一个顶级我的项目。Pulsar 因为原生采纳了存储计算拆散的架构，并且有专门为音讯和流设计的存储引擎 BookKeeper，联合 Pulsar 自身的企业级个性，失去了越来越多开发者的关注。明天的分享分为 3 个局部：
- Apache Pulsar 是什么；
- Pulsar 的数据视图；
- Pulsar 与 Flink 的批流交融。
一、Apache Pulsar 是什么
下图是属于音讯畛域的开源工具，从事音讯或者基础设施的开发者对这些肯定不会生疏。尽管 Pulsar 在 2012 年开始开发，直到 2016 年才开源，但它在跟大家见面之前曾经在雅虎的线上运行了很长时间。这也是为什么它一开源就失去了很多开发者关注的起因，它曾经是一个通过线上测验的零碎。
Pulsar 跟其余音讯零碎最基本的不同在于两个方面：
- 一方面，Pulsar 采纳存储计算拆散的云原生架构；
- 另一方面，Pulsar 有专门为音讯而设计的存储引擎，Apache BookKeeper。
架构
下图展现了 Pulsar 存储计算拆散的架构：
- 首先在计算层，Pulsar Broker 不保留任何状态数据、不做任何数据存储，咱们也称之为服务层。
- 其次，Pulsar 领有一个专门为音讯和流设计的存储引擎 BookKeeper，咱们也称之为数据层。
这个分层的架构对用户的集群扩大非常不便：
- 如果想要反对更多的 Producer 和 Consumer，能够裁减下面无状态的 Broker 层；
- 如果要做更多的数据存储，能够独自裁减底层存储层。
这个云原生的架构有两个次要特点：
- 第一个是存储计算的拆散；
- 另外一个特点是每一层都是一个节点对等的架构。
从节点对等来说，Broker 层不存储数据，所以很容易实现节点对等。然而 Pulsar 在底层的存储也是节点对等状态：在存储层，BookKeeper 没有采纳 master/slave 这种主从同步的形式，而是通过 Quorum 的形式。
如果是要放弃多个数据备份，用户通过一个 broker 并发地写三个存储节点，每一份数据都是一个对等状态，这样在底层的节点也是一个对等的状态，用户要做底层节点的扩容和治理就会很容易。有这样节点对等的根底，会给用户带来很大的云原生的便捷，不便用户在每一层独自扩容，也会进步用户的线上零碎的可用性和维护性。
同时，这种分层的架构为咱们在 Flink 做批流交融打好了根底。因为它原生分成了两层，能够依据用户的应用场景和批流的不同拜访模式，来提供两套不同的 API。
- 如果是实时数据的拜访，能够通过下层 Broker 提供的 Consumer 接口；
- 如果是历史数据的拜访，能够跳过 Broker，用存储层的 reader 接口，间接拜访底层存储层。
存储 BookKeeper
Pulsar 另一个劣势是有专门为流和音讯设计的存储引擎 Apache BookKeeper。它是一个简略的 write-ahead-log 形象。Log 形象和流的形象相似，所有的数据都是源源不断地从尾部间接追加。
它给用户带来的益处就是写入模式比较简单，能够带来比拟高的吞吐。在一致性方面，BookKeeper 联合了 PAXOS 和 ZooKeeper ZAB 这两种协定。BookKeeper 裸露给大家的就是一个 log 形象。你能够简略认为它的一致性很高，能够实现相似 Raft 的 log 层存储。BookKeeper 的诞生是为了服务咱们在 HDFS naming node 的 HA，这种场景对一致性要求特地高。这也是为什么在很多关键性的场景里，大家会抉择 Pulsar 和 BookKeeper 做存储的起因。
BookKeeper 的设计中，有专门的读写隔离，简略了解就是，读和写是产生在不同的磁盘。这样的益处是在批流交融的场景能够缩小与历史数据读取的互相烦扰，很多时候用户读最新的实时数据时，不可避免会读到历史数据，如果有一个专门为历史数据而筹备的独自的磁盘，历史数据和实时数据的读写不会有 IO 的争抢，会对批流交融的 IO 服务带来更好的体验。
利用场景
Pulsar 场景利用宽泛。上面是 Pulsar 常见的几种利用场景：
- 第一，因为 Pulsar 有 BookKeeper，数据一致性特地高，Pulsar 能够用在计费平台、领取平台和交易系统等，对数据服务质量，一致性和可用性要求很高的场景。
- 第二种利用场景是 Worker Queue / Push Notifications / Task Queue，次要是为了实现零碎之间的互相解耦。
- 第三种场景，与 Pulsar 对音讯和队列两种场景的反对比拟相干。Pulsar 反对 Queue 生产模式，也反对 Kafka 高带宽的生产模型。前面我会专门解说 Queue 生产模型与 Flink 联合的劣势。
- 第四个场景是 IoT 利用，因为 Pulsar 在服务端有 MQTT 协定的解析，以及轻量级的计算 Pulsar Functions。
- 第五个方面是 unified data processing，把 Pulsar 作为一个批流交融的存储的根底。
咱们在 2020 年 11 月底的 Pulsar Summit 亚洲峰会，邀请 40 多位讲师来分享他们的 Pulsar 落地案例。如果大家对 Pulsar 利用场景比拟感兴趣，能够关注 B 站上 StreamNative 的账号，观看相干视频。
二、Pulsar 的数据视图
在这些利用场景中，Unified Data Processing 尤为重要。对于批流交融，很多国内用户的第一反馈是抉择 Flink。咱们来看 Pulsar 和 Flink 联合有什么样的劣势？为什么用户会抉择 Pulsar 和 Flink 做批流交融。
首先，咱们先从 Pulsar 的数据视图来开展。跟其余的音讯零碎一样，Pulsar 也是以音讯为主体，以 Topic 为核心。所有的数据都是 producer 交给 topic，而后 consumer 从 topic 订阅生产音讯。
Partition 分区
为了不便扩大，Pulsar 在 topic 外部也有分区的概念，这跟很多音讯零碎都相似。下面提到 Pulsar 是一个分层的架构，它采纳分区把 topic 裸露给用户，然而在外部，实际上每一个分区又能够依照用户指定的工夫或者大小切成一个分片。一个 Topic 最开始创立的时候只有一个 active 分片，随着用户指定的工夫达到当前，会再切一个新的分片。在新开一个分片的过程中，存储层能够依据各个节点的容量，抉择容量最多的节点来存储这个新的分片。
这样的益处是，topic 的每一个分片都会平均地分布在存储层的各个节点上，实现数据存储的平衡。如果用户违心，就能够用整个存储集群来存储分区，不再被单个节点容量所限度。如下图所示，该 Topic 有 4 个分区，每一个分区被拆成多个分片，用户能够依照工夫（比方 10 分钟或者一个小时），也能够依照大小（比方 1G 或者 2G）切一个分片。分片自身有程序性，依照 ID 逐步递增，分片外部所有音讯依照 ID 枯燥递增，这样很容易保障程序性。
Stream 流存储
咱们再从单个分片来看一下，在常见流（stream）数据处理的概念。用户所有的数据都是从流的尾部一直追加，跟流的概念类似，Pulsar 中 Topic 的新数据一直的增加在 Topic 的最尾部。不同的是，Pulsar 的 Topic 形象提供了一些劣势：
- 首先，它采纳了存储和计算拆散的架构。在计算层，它更多的是一个音讯服务层，能够疾速地通过 consumer 接口，把最新的数据返回给用户，用户能够实时的获取最新的数据；
- 另外一个益处是，它分成多个分片，如果用户指定工夫，从元数据能够找到对应的分片，用户能够绕过实时的流间接读取存储层的分片；
- 还有一个劣势是，Pulsar 能够提供有限的流存储。
做基础设施的同学，如果看到依照工夫分片的架构，很容易想到把老的分片搬到二级存储外面去，在 Pulsar 里也是这样做的。用户能够依据 topic 的生产热度，设置把老的，或者超过时限或大小的数据主动搬到二级存储中。用户能够抉择应用 Google，微软的 Azure 或者 AWS 来存储老的分片，同时也反对 HDFS 存储。
这样的益处是：对最新的数据能够通过 BookKeeper 做疾速返回，对于老的冷数据能够利用网络存储云资源做一个有限的流存储。这就是 Pulsar 能够反对有限流存储的起因，也是批流交融的一个根底。
总体来说，Pulsar 通过存储计算拆散，为大家提供了实时数据和历史数据两套不同的拜访接口。用户能够根据外部不同的分片地位，依据 metadata 来抉择应用哪种接口来拜访数据。同时依据分片机制能够把老的分片放到二级存储中，这样能够撑持有限的流存储。
Pulsar 的统一体当初对分片元数据管理的方面。每个分片能够依照工夫寄存成不同的存储介质或格局，但 Pulsar 通过对每个分片的 metadata 治理，来对外提供一个分区的逻辑概念。在拜访分区中的一个分片的时候我能够拿到它的元数据，晓得它的在分区中的程序，数据的寄存地位和保留类型 Pulsar 对每一个分片的 metadata 的治理，提供了对立的 topic 的形象。
三、Pulsar 和 Flink 的批流交融
在 Flink 中，流是一个根底的概念，Pulsar 能够作为流的载体来存储数据。如果用户做一个批的计算，能够认为它是一个有界的流。对 Pulsar 来说，这就是一个 Topic 有界范畴内的分片。
在图中咱们能够看到，topic 有很多的分片，如果确定了起止的工夫，用户就能够依据这个工夫来确定要读取的分片范畴。对实时的数据，对应的是一个间断的查问或拜访。对 Pulsar 的场景来说就是不停的去生产 Topic 的尾部数据。这样，Pulsar 的 Topic 的模型就能够和 Flink 流的概念很好的联合，Pulsar 能够作为 Flink 流计算的载体。
- 有界的计算能够视为一个有界的流，对应 Pulsar 一些限定的分片；
- 实时的计算就是一个无界的流，对 Topic 中最新的数据做查问和拜访。
对有界的流和无界的流，Pulsar 采取不同的响应模式：
第一种是对历史数据的响应。如下图所示，左下角是用户的 query，给定起止的工夫限定流的范畴。对 Pulsar 的响应分为几步：
- 第一步，找到 Topic，依据咱们对立治理的 metadata，能够获取这个 topic 外面所有分片的 metadata 的列表；
- 第二步，依据工夫限定在 metadata 列表中，通过两分查找的形式来获取起始分片和终止的分片，抉择须要扫的分片；
- 第三步，找到这些分片当前通过底层存储层的接口拜访须要拜访的这些分片，实现一次历史数据的查找。
- 对实时数据的查找，Pulsar 也提供和 Kafka 雷同的接口，能够通过 consumer 的形式来读取最尾端分片（也就是最新的数据），通过 consumer 接口对数据进行实时拜访。它不停地查找最新的数据，实现之后再进行下一次查找。这种状况下，应用 Pulsar Pub/Sub 接口是一种最间接最无效的形式。
简略来说，Flink 提供了对立的视图让用户能够用对立的 API 来解决 streaming 和历史数据。以前，数据科学家可能须要编写两套利用别离用来解决实时数据和历史数据，当初只须要一套模型就可能解决这种问题。
Pulsar 次要提供一个数据的载体，通过基于分区分片的架构为下面的计算层提供流的存储载体。因为 Pulsar 采纳了分层分片的架构，它有针对流的最新数据拜访接口，也有针对批的对并发有更高要求的存储层拜访接口。同时它提供有限的流存储和对立的生产模型。
四、Pulsar 现有能力和停顿
最初咱们额定说一下 Pulsar 当初有怎么的能力和最近的一些停顿。
现有能力
schema
在大数据中，schema 是一个特地重要的形象。在音讯畛域外面也是一样，在 Pulsar 中，如果 producer 和 consumer 能够通过 schema 来签订一套协定，那就不须要生产端和生产端的用户再线下沟通数据的发送和接管的格局。在计算引擎中咱们也须要同样的反对。
在 Pulsar-Flink connector 中，咱们借用 Flink schema 的 interface，对接 Pulsar 自带的 Schema，Flink 可能间接解析存储在 Pulsar 数据的 schema。这个 schema 包含两种：
- 第一种是咱们常见的对每一个音讯的元数据（meatdata）包含音讯的 key、音讯产生工夫、或是其余元数据的信息。
- 另一种是对音讯的内容的数据结构的形容，常见的是 Avro 格局，在用户拜访的时候就能够通过 Schema 晓得每个音讯对应的数据结构。
同时咱们联合 Flip-107，整合 Flink metadata schema 和 Avro 的 metadata，能够将两种 Schema 联合在一起做更简单的查问。
source
有了这个 schema，用户能够很容易地把它作为一个 source，因为它能够从 schema 的信息了解每个音讯。
Pulsar Sink
咱们也能够把在 Flink 中的计算结果返回给 Pulsar 把它做为 Sink。
Streaming Tables
有了 Sink 和 Source 的反对，咱们就能够把 Flink table 间接裸露给用户。用户能够很简略的把 Pulsar 作为 Flink 的一个 table，查找数据。
write to straming tables
下图展现如何把计算结果或数据写到 Pulsar 的 Topic 中去。
Pulsar Catalog
Pulsar 自带了很多企业流的个性。Pulsar 的 topic（e.g. persistent://tenant\_name/namespace\_name/topic\_name）不是一个平铺的概念，而是分很多级别。有 tenant 级别，还有 namespace 级别。这样能够很容易得与 Flink 罕用的 Catalog 概念联合。
如下图所示，定义了一个 Pulsar Catalog，database 是 tn/ns，这是一个门路表白，先是 tenant，而后是 namespace，最初再挂一个 topic。这样就能够把 Pulsar 的 namespace 当作 Flink 的 Catalog，namespace 上面会有很多 topic，每个 topic 都能够是 Catalog 的 table。这就能够很容易地跟 Flink Cataglog 做很好的对应。在下图中，上方的是 Catalog 的定义，下方则演示如何应用这个 Catalog。不过，这里还须要进一步欠缺，后边也有打算做 partition 的反对。
FLIP-27
FLIP-27 是 Pulsar – Flink 批流交融的一个代表。后面介绍了 Pulsar 提供对立的视图，治理所有 topic 的 metadata。在这个视图中，依据 metadata 标记每个分片的信息，再依附 FLIP-27 的 framework 达到批流交融的目标。FLIP-27 中有两个概念：Splitter 和 reader。
它的工作原理是这样的，首先会有一个 splitter 把数据源做切割，之后交给 reader 读取数据。对 Pulsar 来说，splitter 解决的还是 Pulsar 的一个 topic。抓到 Pulsar topic 的 metadata 之后，依据每个分片的元数据来判断这个分片存储在什么地位，再选最合适的 reader 进行拜访。Pulsar 提供对立的存储层，Flink 依据 splitter 对每个分区的不同地位和格局的信息，抉择不同的 reader 读取 Pulsar 中的数据。
Source 高并发
另一个和 Pulsar 生产模式严密相干的是。很多 Flink 用户面临的问题是如何让 Flink 更快地执行工作。例如，用户给了 10 个并发度，它会有 10 个 job 并发，但如果一个 Kafka 的 topic 只有 5 个分区，因为每个分区只能被一个 job 生产，就会有 5 个 Flink job 是闲暇的。如果想要放慢生产的并发度，只能跟业务方协调多开几个分区。这样的话，从生产端到生产端和后边的运维方都会感觉特地简单。并且它很难做到实时的按需更新。
而 Pulsar 不仅反对 Kafka 这种每个分区只能被一个 active 的 consumer 生产的状况，也反对 Key-Shared 的模式，多个 consumer 能够独特对一个分区进行生产，同时保障每个 key 的音讯只发给一个 consumer，这样就保障了 consumer 的并发，又同时保障了音讯的有序。
对后面的场景，咱们在 Pulsar Flink 里做了 Key-shared 生产模式的反对。同样是 5 个分区，10 个并发 Flink job。然而我能够把 key 的范畴拆成 10 个。每一个 Flink 的子工作生产在 10 个 key 范畴中的一个。这样从用户生产端来说，就能够很好解耦分区的数量和 Flink 并发度之间的关系，也能够更好提供数据的并发。
主动 Reader 抉择
另外一个方向是上文提到的 Pulsar 曾经有对立的存储根底。咱们能够在这个根底上依据用户不同的 segment metadata 抉择不同的 reader。目前，咱们曾经实现该性能。
近期工作
最近，咱们也在做和 Flink 1.12 整合相干的工作。Pulsar-Flink 我的项目也在不停地做迭代，比方咱们减少了对 Pulsar 2.7 中事务的反对，并且把端到端的 Exactly-Once 整合到 Pulsar Flink repo 中；另外的工作是如何读取 Parquet 格局的二级存储的列数据；以及应用 Pulsar 存储层做 Flink 的 state 存储等。
版权申明： 本文内容由阿里云实名注册用户自发奉献，版权归原作者所有，阿里云开发者社区不领有其著作权，亦不承当相应法律责任。具体规定请查看《阿里云开发者社区用户服务协定》和《阿里云开发者社区知识产权爱护指引》。如果您发现本社区中有涉嫌剽窃的内容，填写侵权投诉表单进行举报，一经查实，本社区将立即删除涉嫌侵权内容。
