Apache Beam 中的缓存：静态变量与有状态处理答案

【问题标题】：Caching in Apache Beam: Static variable vs Stateful processingApache Beam 中的缓存：静态变量与有状态处理
【发布时间】：2018-02-12 18:38:37
【问题描述】：

我正在向 Java 数据流管道内的一个 DoFns 添加缓存功能。 DoFn 目前正在使用 REST 客户端向每个输入元素的端点发送请求（根据请求数量收费，并且响应大约每小时更改一次），我想要实现的是缓存来自的响应端点，并让它每 15 分钟到期一次。

我找到了两种方法来做到这一点：一种来自类似的stackoverflow question，它建议使用静态变量来托管缓存服务（我使用番石榴进行缓存）。但是我不确定如何从 DoFn 之外更新到期时间。

我发现的另一种方法是使用stateful processing 存储跟踪请求和响应的哈希，并使用 TimerSpec 每 15 分钟清除一次“缓存”。虽然看起来没有办法为缓存中的每个元素设置一个计时器。

我还没有尝试过第二种方法。虽然我要实现它，但我想知道是否有人遇到过类似的情况，有什么建议，或者有更好的方法。

【问题讨论】：

【解决方案1】：

但是我不确定如何从 DoFn 之外更新到期时间。

你需要吗？ DoFn 首先进行查找，如果条目已经过期，它会发出请求并更新缓存。缓存读写需要是线程安全的。

虽然看起来没有办法为缓存中的每个元素设置一个计时器。

您可能可以设置一个计时器，每 X 分钟扫描一次整个缓存并刷新过期条目。但是，如果您的状态未使用像这样的全局状态进行键控，则会限制管道的并行性。

【讨论】：

对于你的第一个问题，它是可选的，但我想根据DataflowPipelineOptions 设置的每个配置设置不同的到期时间。对于您的第二点，这是一个有趣的想法，它并不完全是我的想法，感觉有点矫枉过正，但可能会起作用。我想知道您是否曾经尝试过并且知道这种方法会如何执行，因为每次更新本质上都是在重写 hashmap
1.您可以将lastUpdateTimestamp 与缓存数据一起存储。 2. 不用扫描整个缓存也可以这样做。
不完全确定“不扫描整个缓存”。我假设使用ValueState<HashMap<String, Object>> 来存储缓存，所以每个ValueState.write() 都会覆盖整个HashMap
有MapState，虽然不是所有跑者都支持。