语境

考虑有一个 tuples 流，(string, timestamp)目标是让一个 bin 包含为每个字符串收到Map的最小时间戳。另一个限制是此 Map 的更新将是原子的。

对于此输入示例：

("a", 1)
("b", 2)
("a", 3)

预期输出：Map("a" -> 1, "b" -> 2)没有最大时间戳3。

我选择的当前实现是使用列表的 CDT 来保存时间戳，所以我的结果Map("a" -> ArrayList(1), "b" -> ArrayList(2))如下：

private val boundedAndOrderedListPolicy = new ListPolicy(ListOrder.ORDERED, ListWriteFlags.INSERT_BOUNDED)

private def bundleContext(str: String) =
      CTX.mapKeyCreate(Value.get(str), MapOrder.UNORDERED)

private def buildInitTimestampOp(tuple: (String, Long)): List[Operation] = {
      // having these two ops together assure that the size of the list is always 1
      val str = tuple._1
      val timestamp = tuple._2
      val bundleCtx: CTX = bundleContext(str)
      List(
        ListOperation.append(boundedAndOrderedListPolicy, initBin, Value.get(timestamp), bundleCtx),
        ListOperation.removeByRankRange(initBin, 1, ListReturnType.NONE, bundleCtx), // keep the first element of the order list - earliest time
      )
    }

这按预期工作。但是，如果您有更好的方法在没有列表的情况下以原子方式实现这一目标 - 我很乐意听到它。

我的问题

对我不起作用的是保留每个输入 str 收到的最大时间戳。对于上面的示例，所需的结果应该是Map("a" -> ArrayList(3), "b" -> ArrayList(2))。我的实现尝试是：

private def buildLastSeenTimestampOp(tuple: (String, Long)): List[Operation] = {
      // having these two ops together assure that the size of the list is always 1
      val str = tuple._1
      val timestamp = tuple._2
      val bundleCtx: CTX = bundleContext(str)
      List(
        ListOperation.append(boundedAndOrderedListPolicy, lastSeenBin, Value.get(timestamp), bundleCtx),
        ListOperation.removeByRankRange(lastSeenBin, 1, ListReturnType.NONE | ListReturnType.REVERSE_RANK, bundleCtx), // keep the last element of the ordered list - last time
      )
    }

知道为什么它不起作用吗？

回答

private def buildLastSeenTimestampOp(tuple: (String, Long)): List[Operation] = {
      // having these two ops together assure that the size of the list is always 1
      val str = tuple._1
      val timestamp = tuple._2
      val bundleCtx: CTX = bundleContext(str)
      List(
        ListOperation.append(boundedAndOrderedListPolicy, lastSeenBin, Value.get(command.timestamp.getMillis), bundleCtx),
        ListOperation.removeByRankRange(lastSeenBin, -1, ListReturnType.NONE | ListReturnType.INVERTED, bundleCtx), // keep the last element of the ordered list - last time
      )
    }