PaaStorm如何将数据从源实时转换到目标？

本文主要介绍“PaaStorm如何进行从源到目标的实时数据转换”。在日常操作中，我相信很多人对PaaStorm如何进行从源到目的地的实时数据转换有疑问。边肖查阅了各种资料，整理出简单易用的操作方法，希望能帮助大家解答“PaaStorm如何进行从源到目的的实时数据转换”的疑惑！接下来，请和边肖一起学习！

这名字中有什么含义？

PaaStorm的名字其实是PaaSTA和Storm的组合。那么PaaStorm到底是做什么的呢？要回答这个问题，我们先来看看数据管道的基本结构：

看看“Transformer”这一步，我们就知道大部分存储在Kafka中的消息无法直接导入到目标系统中。想象一个红移集群来存储广告推送数据。推送集群只想存储上游系统的某个字段(比如某个业务的平均权重)，否则会保存原始数据并进行聚合。如果红移广告推送集群想要存储所有上游数据，会浪费存储空间，降低系统性能。

过去，每个服务都会编写复杂的MapReduce任务，并在将数据写入目标数据存储之前对其进行处理。然而，这些MapReduce任务都遇到了上面提到的性能和扩展问题。数据管道的好处之一是，无论上游数据是什么，消费者程序都可以获得所需的数据形式。

减少示例代码

我们可以让每个消费者程序根据自己的需要进行数据转换。比如广告推送系统可以自己编写转换服务，从Kafka中的业务数据中提取收视统计，自己维护转换服务。这种方法起初运行良好，但最终当系统扩大规模时，我们遇到了问题。

我们希望提供一个基于以下考虑的转换框架：

许多转换逻辑是通用的，可以在多个团队之间共享。例如将标志位转换成有意义的字段。

这样的转换逻辑通常需要大量的样本代码。如连接数据源或数据用途、保存状态、监控吞吐量、故障恢复等。这种代码不需要从一个服务复制到另一个服务。

为了保证数据的实时处理，数据转换操作应该尽可能快，并基于流。

减少示例代码最自然的方法是提供转换接口。每个人的服务在接口中实现转换操作的特定逻辑，然后，剩下的工作由我们的流处理框架完成。

把Kafka作为消息总线

PaaStorm最初是Kafka到Kafka的转换框架，但逐渐演变为支持其他类型的终端节点。把Kafka作为PaaStorm的终端节点，简化了很多事情：每一个对数据感兴趣的服务都可以在Topic上注册，关注任何转换后的数据或者原始数据，新消息到达时再处理就行了，根本不在乎是谁创建了Topic。转换后的数据将根据卡夫卡的保留策略进行保存。因为卡夫卡是一个发布-订阅系统，所以下游系统也可以随时消费数据。

用Storm处理一切

采用了PaaStorm之后，我们如何将我们的卡夫卡话题之间的关系形象化？因为某些主题中的数据会从源端到端地流向其他主题，所以我们可以将我们的拓扑视为一个有向无环图：

每个节点都是一个Kafka Topic，箭头表示PaaStorm提供的转换操作。这时，“paastom”这个名字就变得更有意义了：和Storm一样，paastom通过转换模块(像Bolt一样)提供了到数据流源(像Spout)的实时转换。

PaaStorm内部机制

PaaStorm的核心抽象被称为Spout(Spolt(Spout和Bolt)的组合)。顾名思义，Spolt接口还定义了两件重要的事情：输入数据源和对该数据源的消息数据的一些处理。

下面例子定义了一个最简单的Spolt：

此Spolt将处理“refresh _ primary . business . ABC 123 EFG 456”主题中的每条消息，并在原始消息中添加一个保存lsquo的字段；姓名。字段的大写值，然后发送处理后的新版本消息。

>

值得一提的是数据管道中的所有消息都是不可修改的。要得到一条修改过的消息，就要创建一个新的对象。而且，因为我们在为消息体中增加一个新字段(就是那个增加的“大写字母的name”字段)，新消息的模式已经改变了。在生产环境中，消息的模式ID是从来都不能写死的。我们要依靠Schematizer服务来为一条修改过的消息注册并提供合适的模式。

***提一句，数据管道的客户端库提供了好几种非常相似的用名字空间、Topic名、源名和模式ID的组合来生成“spolt_source”的方法。这样就可以很容易地让某个Spolt去找到它需要的所有源并从中读取数据。要了解更多信息，请参考Schematizer的文章。

与Kafka相关的处理是怎样的?

也许你已经发现上面的Spolt中没有什么代码是与Kafka  Topic相交互的。这是因为在PaaStorm中，所有真正的Kafka接口相关处理都是由一个内部实例(恰好也叫PaaStorm)完成的。PaaStorm实例会把一个特定的Spolt与对应的源和目的关联起来，并把消息送给Spolt处理，再把Spolt输出的消息发布到正确的Topic上去。

每个PaaStorm实例都用一个Spolt初始化。比如，下面的命令就用上文中定义的UppercaseNameSpolt开启了一次处理：

PaaStorm(UppercaseNameSpolt()).start()

这就意味着所有有意写一个新转换器的人都可以简单地定义一个新的Spolt子类，压根不用修改任何PaaStorm运行体相关的东西。

从内部来看，PaaStorm运行体的主方法也是惊人的简单，伪码如下：

这个运行体先做了一些设置：初始化了生产者和消费者，以及消息计数器。然后，它一直等待上游Topic中的新数据。如果有新数据到来，就用Spolt处理它。Spolt处理之后会输出一条或多条消息，生产者再把它发布到下游的Topic。

另外简单提一下，PaaStorm运行体也提供了比如消费者注册、心跳机制(名叫“tick”)等。比如某个Spolt要经常性地清空它的内容，那就可以用tick来触发。

关于状态保存

PaaStorm保证可以可靠地从故障中恢复。万一发生了崩溃，我们就该从正确的偏移位置开始重新消费。但不幸的是，这个正确的偏移量一般情况下都并不是我们从上游的Topic中消费的***那一条消息。原因是虽然我们已经消费了它，但事实上我们还没来得及把转换后的版本发布出去。

所以重新启动时正确的位置应该是上游Topic与已经成功发布到下游的***一条消息对应的位置。在知道发到下游的***一条消息的情况之后，我们需要知道它对应的上游的消息是哪一条，这样就可以从那里恢复了。

为了方便实现这个功能，PaaStorm的Spolt在处理一条原始消息时，会把与这条原始消息相对应的在上游Topic中的Kafka偏移量也加到转换后的包里。转换后的消息随后会在生产者的回调函数中把这个偏移量传回来。这样，我们就可以知道与下游Topic中***一条消息对应的上游Topic的偏移量了。因为回调函数只有在生产者成功地把转换后的消息发布出去之后才会调用，也就意味着原始消息已经被成功处理了，在这种情况下，消费者就可以很放心的在那个回调函数中提交这个偏移量了。万一发生崩溃，我们可以直接从还没有被完全处理的上游消息那里开始继续处理。

从上面的伪码中可以看到，PaaStorm也会统计消费掉的消息数和发布的消息数。这样，感兴趣的用户可以检查上游和下游Topic中的吞吐量。这让我们很轻松地有了对任意转换操作的监控和性能检查功能。在Yelp，我们是把我们的统计信息发给SignalFX的：

SignalFX图可以显示出在一个PaaStorm实例中生产者和消费者的吞吐量。在这个例子中，输入输出消息量并不匹配。

在PaaStorm中对生产者和消费者分开做统计的好处之一是我们可以把这两个吞吐量放在一起，看看瓶颈是在哪里。如果到不了这个粒度，是很难发现管道中的性能问题的。

PaaStorm的未来

PaaStorm提供了两个东西：一个接口，并实现了一套框架来支持这个接口。尽管我们并不希望PaaStorm的接口很快就被改动，但已经有一些孵化项目在计划解决“转换并连接”的问题了。在将来，我们希望能把PaaStorm的内部换成Kafka  Stream或者Apache  Beam，主要的障碍是对Python的支持程度如何，我们尤其看重的是对终端节点的支持。总之，在有开源的Python流处理项目成熟之前，我们会一直把PaaStorm用下去。

到此，关于“PaaStorm是如何从源到目的做数据的实时转换”的学习就结束了，希望能够解决大家的疑惑。理论与实践的搭配能更好的帮助大家学习，快去试试吧！若想继续学习更多相关知识，请继续关注网站，小编会继续努力为大家带来更多实用的文章！