Pravega应用实战：为什么云原生特性对流处理很重要？

云原生（Cloud Native）与Pravega

随着容器技术和云服务的发展，Kubernetes 和云原生运动已大规模地重定义了应用设计和开发的一些方面。云原生是一种基于微服务架构思想、以容器技术为载体，产品研发运营的模式。通过这样的软件开发模式才能真正的发挥云的弹性、动态调度、自动伸缩等云所特有的能力。

Pravega从设计之初就是云原生（Cloud Native）的应用，可以在各大公有/私有云平台上进行部署和运行。

它的组件都是以低耦合的微服务形式存在，通过运行多个服务实例保证高可用性。
每个服务实例运行于单独的容器中，使用容器实现服务的相互隔离。
可以使用容器编排工具（如Kubernetes）进行统一的服务发现、治理和编排，提高资源利用率，降低运营成本。

Kubernetes是由Google在2014年开源的一个允许自动化部署、管理和伸缩容器的工具，目前已经成为容器编排调度的实际标准。它提供了一些强大的功能，例如容器之间的负载均衡，重启失败的容器以及编排容器使用的存储等。Kubernetes通过将云应用进行抽象简化出各种概念对象，如Pod、Deployment、Job、StatefulSet等，形成了云原生应用的通用可移植的模型，让云应用可以在云之间无缝迁移。

Pravega团队通过第三方资源机制扩展了Kubernetes的API，开发了能够使得Pravega集群的创建、配置和管理更高效和自动化的Operator，包括Pravega Operator 和Zookeeper Operator，通过他们可以使得Pravega在Kubernetes环境中快速创建集群和动态扩展。这些Operator以及其他相关的容器镜像会上传至Pravega在DockerHub官方的镜像仓库：https://hub.docker.com/u/pravega 中，用户也可以直接拉取使用，源代码也在GitHub网站：https://github.com/pravega 上公开。

Pravega核心组件及交互

Pravega能够以一致的方式灵活地存储不断变化的流数据，主要得益于控制面（Control Plane）和数据面（Data Plane）的有机结合。两者都由低耦合的分布式微服务组件管理，前者主要由控制器（Controller）实现，后者主要由段存储器（Segment Store）实现，它们通常以多实例的形式运行在集群中。除此以外，一个完整的Pravage集群还包括一组Zookeeper实例，一组Bookkeeper实例，以及用于提供第二层的存储的服务或接口。它们的关系如图：

控制器（Controller）是Pravega的控制中心，对外提供JAVA和REST接口，接收客户端对于Stream的创建、删除、读写等请求；对内负责Segment的管理和集群的管理。客户端对于Stream的读写请求在控制器中被分拆为对Segment的请求，控制器确定需要使用哪些Segment，从而分发给相应的段存储器来操作。

段存储器（Segment Store）提供了Segment的管理入口，实现了Segment的创建，删除，修改和读取功能。数据被存储在一层和二层存储上，由段存储器负责数据的存储和降层操作。其中一层存储由低延迟的Bookkeeper担任，通常运行于集群的内部；二层存储由容量大且成本较低存储的担任，一般运行于集群的外部。

Zookeeper做为集群的协调者, 它维护可用的控制器和段存储器列表，控制器会监听它们的变化。当一个控制器从集群中删除时，它的工作会被其他的控制器自动接管。当段存储器发生变化时，控制器也会将段容器重新映射以保证系统的正常运行。

Pravega的部署

了解Pravega最好方法就是自己动手部署一个，然后跑一下Pravega示例程序：https://github.com/pravega/pravega-samples

单机版部署

单机版部署是最快捷的方式，你只需要从Pravega Release Github：https://github.com/pravega/pravega/releases 下载一个Pravega发行版，解压后运行：

bin/pravega-standalone

单机版部署只能用来学习和测试，不能用于生产环境中，程序一旦关闭所有的数据也会丢失。

集群部署

Pravega可以运行于多个主机所组成的集群中，也可以运行于云平台中。这里我们只介绍Kubernetes环境下的部署，其他的方式参考http://pravega.io/docs/latest/deployment/deployment/

运行之前，需要保证你拥有一套Kubernetes环境，可以是公有云上的Kubernetes服务（如GKE，Amazon EKS），或者是分布式集群上自建的Kubernetes环境（如通过Kubeadm），以及命令行工具kubectl，helm。

首先，在你的Kubernetes环境中创建一个Zookeeper集群。

Zookeeper集群可以使用Zookeeper Operator来创建，你可以直接使用deploy文件夹中的资源描述文件来部署。

git clone https://github.com/pravega/zookeeper-operator \u0026amp;\u0026amp; cd zookeeper-operator# 创建名为ZookeeperCluster的自定义资源定义（custom resource definition）kubectl create -f deploy/crds/zookeeper_v1beta1_zookeepercluster_crd.yaml# 创建Zookeeper Operator的服务账号、角色和角色绑定，并部署Zookeeper Operatorkubectl create -f deploy/default_nsall_ns/rbac.yamlkubectl create -f deploy/default_nsall_ns/operator.yaml# 部署Zookeeper集群，根据该资源描述文件，将会创建有三个节点的Zookeeper集群kubectl create -f deploy/crds/zookeeper_v1beta1_zookeepercluster_cr.yaml

然后，为Pravega第二层存储创建单独的持久化存储卷（PV）及持久化存储卷声明（PVC）。

这里我们使用 NFS Server Provisioner：https://github.com/kubernetes/charts/tree/master/stable/nfs-server-provisioner ，其他的方式请参考Pravega Operator的自述文件。

NFS Server Provisioner是一个开源工具，它提供一个内置的NFS服务器，可以根据PVC声明动态地创建基于NFS的持久化存储卷。

通过helm chart创建nfs-server-provisioner，执行helm install stable/nfs-server-provisioner 将会创建一个名为nfs的存储类（StorageClass）、nfs-server-provisioner服务与实例、以及相应的服务账户和角色绑定。
新建一个持久化存储卷声明文件pvc.yaml，这里storageClassName指定为nfs。当它被创建时，NFS Server Provisioner会自动创建相应的持久化存储卷。pvc.yaml内容如下：

  kind: PersistentVolumeClaim  apiVersion: v1  metadata:    name: pravega-tier2  spec:    storageClassName: \u0026quot;nfs\u0026quot;    accessModes:      - ReadWriteMany    resources:      requests:        storage: 50Gi

通过kubectl create -f pvc.yaml创建该持久化存储卷声明，你会发现相应的持久化存储卷也被创建。

接着，部署一个Pravega Operator。

你可以直接使用deploy文件夹中的资源描述文件部署：

git clone https://github.com/pravega/pravega-operator \u0026amp;\u0026amp; cd pravega-operatorkubectl create -f pravega-operator/deploy

这里会创建一个名为PravegaCluster自定义资源定义（Custom Resource Definition）、服务账号、角色、角色绑定，并把Pravega Operator部署到Kubernetes集群中。

最后，修改资源描述文件并创建Pravega集群。

资源描述文件cr.yaml指定了Zookeeper地址、各组件的实例数和存储空间。完整文件可以从这里获得：https://github.com/pravega/pravega-operator/tree/master/example

apiVersion: \u0026quot;pravega.pravega.io/v1alpha1\u0026quot;kind: \u0026quot;PravegaCluster\u0026quot;metadata:name: \u0026quot;pravega\u0026quot;spec: # 配置zookeeper集群的地址zookeeperUri: example-zookeepercluster-client:2181 # 配置bookkeeper，建议至少三个实例bookkeeper:  replicas: 3     ...pravega:   # 配置控制器实例，建议至少两个实例  controllerReplicas: 2   # 配置段存储器实例，建议至少三个实例  segmentStoreReplicas: 3   # 配置第二层存储，使用之前创建的持久化存储卷声明  tier2:    filesystem:      persistentVolumeClaim:        claimName: pravega-tier2  ...

根据描述文件（cr.yaml）创建一个Pravega集群：

kubectl create -f pravega-operator/example/cr.yaml

集群创建成功后，你可以通过以下命令查看集群的运行状态：

kubectl get all -l pravega_cluster=pravega

创建一个简单的应用

让我们来看看如何构建一个简单的Pravega应用程序。最基本的Pravega应用就是使用读客户端（Reader）从Stream中读取数据或使用写客户端（Writer）向Stream中写入数据。两个简单的例子都可以在Pravega示例中的 gettingstarted 应用程序中找到：https://github.com/pravega/pravega-samples/tree/master/pravega-client-examples/src/main/java/io/pravega/example/gettingstarted

要正确实现这些应用，首先回顾一下Pravega是如何高效并发地读写Stream：

为了实现并发地读写，Stream被分为一个或多个Segment，系统可以根据I/O负载动态调整Segment的数目。
写数据时，多个Writer可以同时向多个Segment追加数据而无需知道它们的变化，由路由键（Routing key）保证顺序的一致性。
- 路由键是一个字符串，控制器会根据它的哈希值而决定该事件将会被派发到哪个Segment中。具有同样路由键的事件会被派发到同一个Segment，这样可以保证它们能以一致的顺序被访问。
- 如果Segment发生了变化，具有相同路由键的事件也会一致的被映射到新的Segment中。
读数据时，读者组（ReaderGroup）中的一组Reader可以同时从不同的Segment中读数据。
- 一个ReaderGroup包含一个或多个Reader，每个Reader从一个或多个Segment中读数据。
- 为了保证每个事件只被读取一次，一个Segment只能被当前ReaderGroup中的一个Reader读。
- 一个ReaderGroup可以从一个或多个Stream中读数据，不同的ReaderGroup是相互独立的。
写数据只能向Stream的尾部追加，读数据可以从指定位置读。

使用Writer向流中写数据

示例HelloWorldWriter举例说明了如何使用EventStreamWriter向Stream中写一个事件。我们来看一下其中最关键的run()方法：

1) 使用StreamManager创建一个Scope。

StreamManager streamManager = StreamManager.create(controllerURI);final boolean scopeCreation = streamManager.createScope(scope);

StreamManager是创建、删除和管理stream及scope的接口，通过指定一个控制器地址与控制器通信。

2) 使用StreamManager创建一个Stream。

StreamConfiguration streamConfig = StreamConfiguration.builder()        .scalingPolicy(ScalingPolicy.fixed(1))        .build();final boolean streamCreation = streamManager.createStream(scope, streamName, streamConfig);

创建stream的时候需要指定scope，名称和配置项。

其中，流配置项包括流的伸缩策略（Scaling Policy）和降层策略（Retention Policy）。Pravega支持三种伸缩策略，将会在下一篇《Pravega动态弹性伸缩特性》中具体介绍。降层策略已经在上一篇中介绍过。

3) 使用ClientFactory创建一个Writer，并向Stream中写数据。

try (ClientFactory clientFactory = ClientFactory.withScope(scope, controllerURI);         EventStreamWriter\u0026lt;String\u0026gt; writer = clientFactory.createEventWriter(streamName,                                                          new JavaSerializer\u0026lt;String\u0026gt;(),                                                EventWriterConfig.builder().build())) {    final CompletableFuture\u0026lt;Void\u0026gt; writeFuture = writer.writeEvent(routingKey, message);}

ClientFactory是用于创建Readers，Writers和其它类型的客户端对象的工具，它是在Scope的上下文中创建的。ClientFactory以及由它创建的对象会消耗Pravega的资源，所以在示例中用try-with-resources来创建这些对象，以保证程序结束时这些对象会被正确的关闭。如果你使用其他的方式创建对象，请确保在使用结束后正确的调用这些对象的close()方法。

在创建Writer的时候还需要指定一个序列化器，它负责把Java对象转化为字节码。事件在Pravega中是以字节码的形式存储的，Pravega并不需要知道事件的具体类型，这使得Pravega可以存储任意类型的对象，由客户端负责提供序列化/反序列化的方法。

用writeEvent方法将事件写入流，需要指定一个路由键（Routing key）。

使用Reader从流中数据

示例HelloWorldReader举例说明了如何使用EventStreamReader从Stream中读取事件，其关键部分也是在run()方法中。

1) 使用ReaderGroupManager创建一个ReaderGroup。

final ReaderGroupConfig readerGroupConfig = ReaderGroupConfig.builder()\t\t.stream(Stream.of(scope, streamName))\t\t.build();try (ReaderGroupManager readerGroupManager = \t\t\t\t\t\t\t   ReaderGroupManager.withScope(scope, controllerURI)) {\treaderGroupManager.createReaderGroup(readerGroup, readerGroupConfig);}

ReaderGroupManager类似于ClientFactory，也是在scope的上下文中创建的。

创建ReaderGroup需要指定名称和配置项，其中配置项规定了该ReaderGroup从哪些Stream中读数据，以及所要读取的Stream的起止位置。Pravega具有Position的概念，它表示Reader当前所在的Stream中的位置。应用保留Reader最后成功读取的位置，Position的信息可以用于Checkpoint恢复机制，如果读失败了就从这个保存的检查点重新开始读。

2) 创建一个Reader并从流中读数据。

try (ClientFactory clientFactory = ClientFactory.withScope(scope, controllerURI);\tEventStreamReader\u0026lt;String\u0026gt; reader = clientFactory.createReader(\u0026quot;reader\u0026quot;, \t\t\t\t\t\t\t\t\t\treaderGroup, \t\t\t\t\t\t\t\t\t\tnew JavaSerializer\u0026lt;String\u0026gt;(), \t\t\t\t\t\t\t\t\t\tReaderConfig.builder().build())) {\tEventRead\u0026lt;String\u0026gt; event = null;\tdo {   \t\tevent = reader.readNextEvent(READER_TIMEOUT_MS);     \t\t   } while (event.getEvent() != null);}

Reader也是由ClientFactory创建的。一个新建的Reader会被加入到相应的ReadGroup中，系统根据当前ReaderGroup的工作负载自动分配相应的段给新创建的Reader。Reader可以通过readNextEvent读取事件。

由于Pravega的自动伸缩功能，Segment的数量会随着负载的变化而变化，当ReaderGroup管理的Segment总数发生变化时，会触发段通知（SegmentNotification），ReaderGroup可以监听该事件并适时地调整Reader的数量。如果当前的Segment比较多，为了保证读的并发性，建议增加Reader；反之，如果当前的Segment比较少，建议减少Reader。由于Reader和Segment是一对多的关系，Reader的数量大于Segment的数量是没有意义的。

Pravega系列文章计划

Pravega根据Apache 2.0许可证开源，我们欢迎对流式存储感兴趣的大咖们加入Pravega社区，与Pravega共同成长。下一篇文章将着重介绍Pravega的动态弹性伸缩特性。本篇文章为Pravega系列第三篇，前两篇回顾如下：

实时流处理(Streaming)统一批处理(Batch)的最后一块拼图:Pravega
Pravega设计原理与基本架构介绍

作者简介

滕昱：就职于 DellEMC 非结构化数据存储部门 (Unstructured Data Storage) 团队并担任软件开发总监。2007 年加入 DellEMC 以后一直专注于分布式存储领域。参加并领导了中国研发团队参与两代 DellEMC 对象存储产品的研发工作并取得商业上成功。从 2017 年开始，兼任 Streaming 存储和实时计算系统的设计开发与领导工作。
刘晶晶，现就职于DellEMC，10年+分布式、搜索和推荐系统开发以及架构设计经验，现从事流存储和流搜索相关的设计与开发工作；
周煜敏，复旦大学计算机专业研究生，从本科起就参与 DellEMC 分布式对象存储的实习工作。现参与 Flink 相关领域研发工作。

参考：

https://www.pravega.io
https://github.com/pravega/zookeeper-operator
https://github.com/pravega/pravega-operator