Storm中的分布式缓存答案

【问题标题】：Distributed caching in stormStorm中的分布式缓存
【发布时间】：2023-03-29 15:08:01
【问题描述】：

Apachestorm中如何存储临时数据？

在storm topology中，bolt需要访问之前处理过的数据。

Eg: if the bolt processes varaiable1 with result as 20 at 10:00 AM.

在10:15 AM 再次收到varaiable1 作为50 然后结果应该是30 (50-20)

稍后如果 varaiable1 收到 70，那么结果应该是 20 (70-50) 10:30。

如何实现这个功能。

【问题讨论】：

标签： apache-storm

【解决方案1】：

简而言之，您想使用 Storm 的运行元组进行微批量计算。首先，您需要在元组集中定义/查找键。使用该键在螺栓之间进行字段分组（不要使用随机分组）。这将保证相关元组将始终发送到相同键的下游螺栓的相同任务。定义类级集合 List/Map 以维护旧值并在其中添加新值进行计算，不用担心它们在同一 bolt 的不同 executor 实例之间是线程安全的。

【讨论】：

【解决方案2】：

恐怕目前还没有这样的内置功能。但是您可以使用任何类型的分布式缓存，例如 memcached 或 Redis。这些缓存解决方案真的很容易使用。

【讨论】：

【解决方案3】：

有几种方法可以做到这一点，但这取决于您的系统要求、团队技能和基础架构。

您可以使用 Apache Cassandra 来存储事件，并在元组中传递行的键，以便下一个 Bolt 可以检索它。

如果您的数据本质上是时间序列，那么也许您想看看OpenTSDB 或InfluxDB。

您当然可以回退到软件事务内存之类的东西，但我认为这需要大量的精心制作。

【讨论】：

我们有这个想法，但是我们需要存储的数据非常少而且是临时的。所以我们不能得到一个专用的数据库。

【解决方案4】：

你可以使用 CacheBuilder 在你的扩展 BaseRichBolt 中记住你的数据（把它放在 prepare 方法中）：

// init your cache.
this.cache = CacheBuilder.newBuilder()
                         .maximumSize(maximumCacheSize)
                         .expireAfterWrite(expireAfterWrite, TimeUnit.SECONDS)
                         .build();

然后在执行中，您可以使用缓存来查看您是否已经看到了该键条目。从那里你可以添加你的业务逻辑：

// if we haven't seen it before, we can emit it.
if(this.cache.getIfPresent(key) == null) {
    cache.put(key, nearlyEmptyList);
    this.collector.emit(input, input.getValues());
}

this.collector.ack(input);

【讨论】：

我怎样才能得到钥匙？因为当另一个元组到来时会再次调用bolt，并且这个元组现在有其他值。
嗨，您从元组中获得了密钥。在你的螺栓中，你有方法执行（公共无效执行（元组元组）{..）。从中，您可以提取您的密钥。

【解决方案5】：

这个问题非常适合演示 Apache Spark 在微批次上的内存计算。但是，您的用例在 Storm 中实现起来很简单。

确保螺栓使用字段分组。它会始终将传入的元组散列到同一个bolt，这样我们就不会丢失任何元组。
在螺栓的本地缓存中维护一个 Map。此映射将保留“变量”的最后一个已知值。

    class CumulativeDiffBolt extends InstrumentedBolt{

    Map<String, Integer> lastKnownVariableValue;

    @Override
    public void prepare(){
         this.lastKnownVariableValue = new HashMap<>();
         ....

    @Override
    public void instrumentedNextTuple(Tuple tuple, Collector collector){
         .... extract variable from tuple
         .... extract current value from tuple
         Integer lastValue = lastKnownVariableValue.getOrDefault(variable, 0)
         Integer newValue = currValue - lastValue

         lastKnownVariableValue.put(variable, newValue)
         emit(new Fields(variable, newValue));
       ...
    }

【讨论】：