Apache Storm 可以用来处理具有一组动态属性的元组吗？

Posted 2023-04-18

【中文标题】Apache Storm 可以用来处理具有一组动态属性的元组吗？

【英文标题】：Can Apache Storm be used to process tuples with a dynamic set of properties?

【发布时间】：2015-11-12 18:31:53

【问题描述】：

我目前正在评估 Apache Storm 以处理来自多个数据源的异构数据。虽然所有数据可能有一些共同的属性（即“类型”属性），但我希望能够使用许多不同的“类”元组，并且能够处理新的数据类型，而对拓扑的更改最少。举个例子，这些数据类型可能是什么样子：

type=LogTransaction,timestamp=...,user=...,duration=...
type=LogEvent,timestamp=...,user=...,message=...

Storm 页面上的示例主要处理简单的元组，这些元组是预先定义好的，以便 spout / bolts 可以静态声明输出字段。

我最初的想法是声明类型并将所有其他属性存储在 Map<String,Object> 中，然后可以声明：

public void declareOutputFields(OutputFieldsDeclarer ofd) 
    ofd.declare(new Fields("type", "attributes"));

但是，我相信 Storm 的许多更高级的功能将不再正常工作。例如，据我了解，我不能再使用 Trident 对任何属性执行groupBy。

有没有更好的方法来处理我在 Apache Storm 中遗漏的这类数据？我确实发现 this post 描述了一个类似的问题，但是我想避免为每种数据类型创建一个 Java 类。

【问题讨论】：

暂且搁置特定于 Storm 的代码，如果每个元组没有特定的类，您将如何处理多个类的元组？

我的一个要求是能够根据一个或多个属性对数据进行分组并将其流式传输到不同的目的地。这方面的一个例子是收集与某个用户相关联的所有元组（通过“用户”属性）。我知道任何组件最终会使用该数据都需要了解每个“类”来处理数据，但理想情况下流式管道应该处理任何类型的元组。

您似乎在说您的要求基本上是，“无论我们决定向您发送什么，您都必须正确且动态地处理。”这似乎尽可能模糊。这些元组中的任何内容是否相关，以便您可以将某些内容抽象为接口或抽象类？

不幸的是，这确实非常符合我的要求......除了每个元组中都存在的 type 属性之外，没有任何单一的数据属性可以保证存在。但是，即使有共享属性，据我了解，我仍然必须将每个子类的属性声明为输出字段才能在 Apache Storm 中使用它们 - 还是我遗漏了什么？

你知道每种类型的所有可能属性吗？

【参考方案1】：

您可以使用自己的自定义字段，只要该字段是可序列化的，它会在风暴中与多个主管一起工作。

因为storm是一种分布式数据处理工具，当存在多个supervisor时，基于grouping，某些bolt会发出fields到不同supervisor上运行的bolt。在这种情况下，输出字段将被序列化并通过网络发送。这种序列化可以是常规的 java 序列化或 Kryo 序列化（以避免网络延迟）。

因此，如果您的 jvm 无法序列化您的输出字段，您可能会遇到异常。

【讨论】：

如果我错了，请纠正我，但 Java 序列化意味着必须在编译时知道这些字段，这意味着这不能解决我对必须为每种类型实现 Java 类的担忧数据。