Wormhole大数据流式处理平台五大功能

导读：在上一篇文章「Wormhole 大数据流式处理平台之设计思想」中，我们介绍了 Wormhole 的设计思想，并给出了 Stream、UMS、Flow、Namespace 等相关概念的具体定义，从文章中我们得知，Wormhole 作为实时流式处理平台，其设计思想最终是为流上处理数据而服务的。在本文中，我们主要从 Wormhole 的功能设计入手，重点介绍 Wormhole 所支持的几个基本功能。

Wormhole 支持的功能很多，如图 1 所示，除了流式数据处理，Wormhole 在管理和运维等方面也做的比较完善。下面我们从流式处理、平台管理、数据质量、数据安全以及运维监控五个维度来介绍 Wormhole 的具体功能。

一、流式处理

Wormhole 的核心是流式处理，并将流式处理抽象为 Flow（流式处理逻辑管道，具体参见：#Wormhole# 流式处理平台设计思想）。Flow 的引入，使得一个 Spark Streaming 上可以跑不同的处理逻辑，也就是多个 Flow 可以在一个 Spark Streaming 上同时执行而互不影响。这种异构逻辑的并行处理大大提高了资源利用率，也提高了流式处理的易用性。

如图 2 所示，Flow 从处理过程角度分为解析、转换、写入三个过程，具体如下：

1.1 解析

Flow 支持多种消息协议，UMS 和用户自定义 JSON 两种消息协议：

• UMS

UMS 是 Flow 支持的标准消息协议，在设计思想的文章中有介绍，这里不再介绍。（参见：#Wormhole# 流式处理平台设计思想）

• 自定义 JSON

开源后，为了适配用户已有系统的数据格式需求，Flow 开始支持用户自定义 JSON 消息协议，使用也比较方便简单，只要在页面贴一个 JSON 消息例子，就会自动解析，然后通过点击配置即可完成自定义 JSON 的 Schema 的定义。

1.2 转换

这里的转换主要指对流上指定的 Namespace 的数据进行处理，处理方式包括 Transform SQL（包含 Spark SQL、Lookup SQL、Stream Join SQL）和接口扩展等，并且所有操作都可以有多项，即一个 Flow 中可以有多个 Spark SQL，多个 Lookup SQL，多个接口扩展等，具体如下：

• Spark SQL

利用 Spark 天然支持的 SQL 对数据做一些 map 操作，用户指需要在页面编写 SQL 即可实现实时对流上数据的 Spark SQL 处理。

• Lookup SQL

Lookup SQL 是指将流上指定 Namespace 数据按某个或某几个字段 join 外部实体数据系统的数据，也就是将流上的数据加列处理，在页面编写 SQL 即可实现对流上数据的 Lookup 操作。目前支持多种 Lookup SQL 数据系统，包括 Mysql、Oracle、Postgresql、SQLServer、Cassandra、Mongodb、Phoenix、ElasticSearch、Vertical、KUDU、Redis、Hbase，除了 Redis 和 Hbase 写法是类 SQL 写法之外，其他都支持 SQL 写法。下面举例介绍 SQL 的编写：

✔ 单字段关联：

select col1, col2, … from tableName where colA in namespace.X;

✔ 多字段关联：

select col1, col2, … from tableName where (colA,colB) in (namespace.X,namespace.Y);

✔ Redis

因 Redis 不是结构化存储方式，所以只能模仿 SQL 写法：

Redis 的 value 是字符串时：select name:type as n1 from default(simple) joinby (key1+’_’+key2);

Redis 的 value 是 JSON 串时：select name:int,name:string,name:long from default(json) joinby (key1+’_’+key2);

✔ HBase

考虑到 HBase 的性能，只支持根据 Rowkey Lookup：

select h1:string as hx,h3:string from test_lookup(cf1) joinby mod(hash(sub(reverse(md5(id2)),6)),1000)/value(id2+’_’);

mod/hash/sub/reverse/md5 都是考虑数据倾斜问题对 rowkey 的数据进行的处理;

✔ Stream Join SQL

Stream Join SQL 是指将流上的两个 Namespace 的数据做 Join 操作，即将流上的数据 Namespace A 去 Join 流上的数据 Namespace B，得到一个宽表。

✔ Transform 扩展接口

虽然通过 SQL 已经可以解决大部分数据处理逻辑需求，但是为了满足一些个性化逻辑的应用，Flow 定义了标准的扩展接口，用户实现接口即可编写自定义逻辑，并且可以与 Transform SQL 在一个 Flow 里同时使用。

✔ Event Time Strategy

基于事件时间，根据数据状态做的一些策略，目前支持在一段时间后，数据某些字段不符合条件时，可以做一些处理的选择。主要针对的场景是当 Lookup 时，如果关联的数据不存在（延迟等原因），那么就可以将未 Lookup 到的数据缓存一段时间，直到超时。

✔ UDF 热加载

因 Spark SQL 支持 UDF，Wormhole 也支持了 UDF，并且支持热加载，即在不停 Spark Streaming 的情况下，加载 UDF 的 jar 包和类，并使用 UDF。

1.3 写入

写入是指将流上处理好的数据写入到指定的数据系统中。

• 多种 Sink 写入

目前支持主流的关系型数据库和 NoSQL 系统，包括 Mysql、Oracle、HBase、Kafka、ElasticSearch、MongoDB、Cassandra、KUDU。

• Sink 写入方式多样

可以根据用户配置确定数据的写入方式，目前支持追加和幂等写入。追加是指将所有数据 insert 到数据系统中，不区分数据状态；幂等是指 Wormhole 接收到的数据包括 insert/update/delete 状态，但能够保证与源数据一致状态的写入到数据系统中（如果 Kafka 中数据能保证顺序则支持强一致性，否则支持最终一致性）。

• Sink 接口扩展

Wormhole 虽然已经支持了主流的存储系统，但为了更好的兼容性，Flow 定义了标准的写入接口，用户可以根据自己的需求实现写入逻辑。

二、平台管理

Wormhole 提供了一个可视化操作的 web 系统—Rider，用来对各项配置和流程进行统一管理。同时也可以对外提供 Restful 方式操作 Wormhole Stream 和 Wormhole Flow。并且通过 Rider 来管理和配置多租户等，具体功能可以参考我们的《Wormhole 用户手册》。

https://edp963.github.io/worm…

三、数据质量

互联网公司中存在着大量数据，并且数据依然以很快的速度增长。其中，金融数据的质量异常重要，这一点与互联网其他数据有很大的不同。Wormhole 在这方面做了很多工作。

3.1 数据一致性

Wormhole 可以保证数据的最终一致性，这一点主要是通过幂等、数据备份和回灌等方式来保证。

3.2 异常反馈

当在计算过程中，如果出现异常，则 Wormhole 就会把相关的 Flow、起止 offset、event time 等信息反馈给监控系统，然后可以手动对错误进行处理。

3.3 端到端比对

为了验证数据的一致性，实现了一个端到端实时比对的插件，在 Flow 进行中即可数据比对，并且不影响 Flow 执行。

3.4 数据回灌

Wormhole 提供了数据备份的能力，将数据以文本形式写入到 HDFS 上。同时 Wormhole 也支持将备份的数据按一定条件（起止时间等）将数据回灌到对应的 topic 中，然后让 Wormhole 再消费一次。比如有异常反馈时，可以手动的将对应数据重新回灌到对应 topic 中，然后 Wormhole 可以幂等的将数据写入到各个数据系统，保证数据最终一致性。

四、数据安全

金融数据是非常敏感的，那么平台化就要保证数据的安全，在这方面，Wormhole 通过权限控制、数据隔离和脱敏加密等方式保证了数据安全：

4.1 权限控制

Wormhole 定义了三种类型用户，分别为管理员用户（admin）、普通用户（user）和第三方系统用户（app）。admin 用户负责管理数据资源的连接地址，UDF jar 包，其他用户等信息。user 用户负责管理流式执行引擎和业务逻辑。app 用户代表通过第三方系统与 wormhole 集成的用户，具有部分 user 用户的能力。通过权限的控制，实现了功能的约束，进而保证数据安全。

4.2 数据隔离

所有数据都是通过 Namespace 定义的，user 用户可以使用哪些数据（Namespace）资源是由 admin 分配的，user 用户登录到系统后，只能使用 admin 用户为其授权的数据（Namespace），也就实现了数据隔离。

4.3 脱敏加密

金融数据的一些信息需要进行加密才能对其他项目提供，那就可以在流上直接处理，通过 UDF 对某些字段进行加密、加盐等等，保证使用方看到的数据是脱敏的，进而保证敏感信息不外泄。

五、运维监控

5.1 实时动态管理

实时动态管理包括两方面，一方面是可以对 Flow 的相关配置进行实时管理，并且实时生效，这一点主要是使用了 Zookeeper 的能力；另一方面主要是针对 Spark 不支持的一些功能进行了扩展，包括不停 Spark Streaming 时，动态加载与注册 UDF、和动态管理接入的 Topic。

5.2 生命周期管理

Wormhole 的 Stream 和 Flow 分别设计了一套有限状态机，也就是为两者分配了生命周期，保证操作的正确性。

5.3 性能监控

通过每个 batch 的每个 Flow 处理情况的 Feedback 信息，可以对每个 batch 的延迟情况、吞吐量、数据量等实时监控。

5.4 硬件资源隔离

主要从两个层次实现，一个是每个项目可以使用多少资源（CPU/ 内存）都是分配的，超过时无法启动新的 Stream；另一个是每个 Spark Streaming 应用的资源是指定的，并且由 Yarn 分配，就是说 Spark Streaming 应用本身已经是资源隔离的。

流式处理支持异构逻辑的并行处理，提高了资源利用率；可视化操作的 web 系统统一管理各项配置和流程；数据质量通过“异常反馈”、“数据回灌”等方式得到了极大保障；数据安全也因“权限控制”、“数据隔离”、“脱敏加密”等方式得以实现。在介绍 Wormhole 的功能篇中，我们了解了 Wormhole 所支持的几个重要功能。那么 Wormhole 的设计细节具体是怎样来实现的呢？在 Wormhole 系列的第三篇文章中，我们会讲解其设计细节中的几个关键点，敬请大家期待~

作者：赵平

来源：宜信技术学院