2019/04/17

IoT前沿 | 流数据处理难？一切都在计划之中

戴尔官方企采中心

各位读者，《IoT前沿》栏目已经进行了三期啦~

之前的内容里，我们已经对Pravega的特性和优势进行了介绍，相信大家对它已经有了一个完整的了解。

现在问题来了，既然Pravega歪瑞酷，能够解决未来流数据处理的难题，那么我要什么时候才能用上这项技术呢？

咳咳，虽然正式推出市场还在戴尔易安信的计划之中...

但是，Pravega的发行版已经开放，可以供大家下载使用了！

俗话说“眼过百遍不如动手一遍”，Pravega说得再好，那也是王婆卖瓜，不如让各位读者亲自动手部署一番，因此在今天的这篇文章里，我们将为大家详细介绍Pravega的部署方法。

Pravega从入门到精通，从这里开始~

作者简介

滕昱

滕昱：就职于Dell EMC中国研发集团，非结构化数据存储部门团队并担任软件开发总监。2007年加入Dell EMC以后一直专注于分布式存储领域。参加并领导了中国研发团队参与两代Dell EMC对象存储产品的研发工作并取得商业上成功。从2017年开始，兼任Streaming存储和实时计算系统的设计开发与领导工作。

刘晶晶

刘晶晶：现就职于DellEMC，10 年+分布式、搜索和推荐系统开发以及架构设计经验，现从事流存储相关的设计与开发工作。

周煜敏

周煜敏：复旦大学计算机专业研究生，从本科起就参与Dell EMC分布式对象存储的实习工作。现参与Flink相关领域研发工作。

Pravega属于戴尔科技集团IoT战略下的一个子项目。该项目是从0开始构建，用于存储和分析来自各种物联网终端的大量数据，旨在实现实时决策。其结合了戴尔易安信PowerEdge服务器，并无缝集成到非结构化数据产品组合Isilon和Elastic Cloud Storage（ECS）中，同时拥抱Flink生态，以此为用户提供IoT所需的关键平台。

戴尔科技集团IoT解决方案集合了戴尔科技家族的力量，覆盖从边缘到核心再到云端

1云原生与Pravega

随着容器技术和云服务的发展，Kubernetes和云原生已经重定义了应用设计和开发的一些方面。

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

• 它的组件都是以低耦合的微服务形式存在，通过运行多个服务实例保证高可用性。

• 每个服务实例运行于单独的容器中，使用容器实现服务的相互隔离。

• 可以使用容器编排工具（如Kubernetes）进行统一的服务发现、治理和编排，提高资源利用率，降低运营成本。

同时，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上公开。

2Pravega核心组件及交互

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

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

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

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

3Pravega的部署

了解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。

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

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

git clone https://github.com/pravega/zookeeper-operator && cd zookeeper-operator

# 创建名为 ZookeeperCluster 的自定义资源定义（custom resource definition）

kubectl create -f deploy/crds/zookeeper_v1beta1_zookeepercluster_crd.yaml

# 创建 Zookeeper Operator 的服务账号、角色和角色绑定，并部署 Zookeeper Operator

kubectl create -f deploy/default_nsall_ns/rbac.yaml

kubectl create -f deploy/default_nsall_ns/operator.yaml

# 部署 Zookeeper 集群，根据该资源描述文件，将会创建有三个节点的 Zookeeper 集群

kubectl create -f deploy/crds/zookeeper_v1beta1_zookeepercluster_cr.yaml

2然后，为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: "nfs"

accessModes:

- ReadWriteMany

resources:

requests:

storage: 50Gi

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

3接着，部署一个Pravega Operator。

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

git clone https://github.com/pravega/pravega-operator && cd pravega-operator

kubectl create -f pravega-operator/deploy

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

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

资源描述文件cr.yaml指定了Zookeeper地址、各组件的实例数和存储空间。完整文件可以从这里获得：

https://github.com/pravega/pravega-operator/tree/master/example

apiVersion: "pravega.pravega.io/v1alpha1"

kind: "PravegaCluster"

metadata:

name: "pravega"

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

4创建一个简单的应用

让我们来看看如何构建一个简单的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的尾部追加，读数据可以从指定位置读。

5使用Writer向流中写数据

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

❶ 使用StreamManager创建一个Scope。

StreamManager streamManager = StreamManager.create(controllerURI);

final boolean scopeCreation = streamManager.createScope(scope);

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

❷ 使用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动态弹性伸缩特性》中具体介绍。降层策略已经在上一篇中介绍过。

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

try (ClientFactory clientFactory = ClientFactory.withScope(scope, controllerURI);

EventStreamWriter<String>writer = clientFactory.createEventWriter(streamName,

new JavaSerializer(), EventWriterConfig.builder().build())) {

final CompletableFuture<Void>writeFuture = writer.writeEvent(routingKey, message);

}

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

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

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

6使用Reader从流中数据

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

❶ 使用ReaderGroupManager创建一个ReaderGroup。

final ReaderGroupConfig readerGroupConfig = ReaderGroupConfig.builder()

.stream(Stream.of(scope, streamName))

.build();

try (ReaderGroupManager readerGroupManager = ReaderGroupManager.withScope(scope, controllerURI)) {

readerGroupManager.createReaderGroup(readerGroup, readerGroupConfig);

}

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

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

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

try (ClientFactory clientFactory = ClientFactory.withScope(scope, controllerURI);

EventStreamReader<String> reader = clientFactory.createReader("reader",

readerGroup,

new JavaSerializer<String>(),

ReaderConfig.builder().build())) {

EventRead<String> event = null;

do {

event = reader.readNextEvent(READER_TIMEOUT_MS);

} 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的云原生特性、核心组件、安装部署实践以及Reader/Writer的基本应用实践。

截至目前，我们已经花了4个篇幅（第一期、第二期、第三期）详细了Pravega，相信你对它已经有了全面的了解，在下一期的内容里，我们将对Pravega的仅一次语义及事务支持进行介绍。欢迎大家持续关注，如何你有疑问，可在下方留言或知乎号上（见下方二维码）找到我们。下一期见~