我在 PySpark 结构化流中遇到以下问题。
我的流数据中的每一行都有一个用户 ID 和一个时间戳。现在，对于每一行和每个用户，我想添加一个时间戳不同的列。
例如，假设我收到的第一行是：“用户 A，08:00:00”。如果第二行显示“用户 A，08:00:10”，那么我想在第二行添加一个名为“间隔”的列，表示“10 秒”。
有没有人知道如何做到这一点？我尝试使用结构化流文档的窗口函数示例，但没有用。
非常感谢
【问题讨论】：
标签： apache-spark pyspark spark-structured-streaming
因为我们谈论的是结构化流和“每一行和每个用户”，这告诉我您应该使用带有某种流聚合的流查询（groupBy 和 @987654324 @)。
对于流式聚合，您只能依靠结构化流中的微批量流执行。这使得单个用户的记录可以是两个不同微批次的一部分。这让你需要一个状态。
综上所述，您需要有状态的流式聚合。
有了这个，我想你想要Arbitrary Stateful Operations之一，即KeyValueGroupedDataset.mapGroupsWithState或KeyValueGroupedDataset.flatMapGroupsWithState（见KeyValueGroupedDataset）：
许多用例需要比聚合更高级的有状态操作。例如，在许多用例中，您必须从事件的数据流中跟踪会话。为了进行这种会话化，您必须将任意类型的数据保存为状态，并使用每个触发器中的数据流事件对状态执行任意操作。
从 Spark 2.2 开始，这可以使用操作
mapGroupsWithState和更强大的操作flatMapGroupsWithState来完成。这两种操作都允许您在分组数据集上应用用户定义的代码来更新用户定义的状态。
状态将是每个用户找到的最后一条记录。看起来可行。
我担心的是：
这个流式查询要处理多少用户？ （越多状态越大）
何时清理状态（不再期望在流中的用户）？ （这将保持合理大小的状态）
【讨论】：