hadoop - 在 Pig 中过滤基于时间的数据

Question

我现在在本地模式下使用 Pig 0.11.1，从 CSV 加载数据。

到目前为止，我已经能够加载我们的数据集并对其执行所需的计算。下一步是从数据中抽取一些样本并执行相同的计算。为了复制现有流程，我们希望每十五分钟获取一个数据点。

这就是问题所在。我可以在 Pig 中编写一个过滤器，如果数据点恰好在 15 分钟间隔内，我将匹配该过滤器，但我将如何获取接近 15 分钟边界的数据点？

我需要查看十五分钟的标记并捕获那里的记录。如果该标记上没有记录(很可能)，那么我需要在标记之后抓取下一条记录。

我想我需要编写自己的过滤器 UDF，但似乎 UDF 需要是有状态的，以便它知道何时在时间间隔后找到第一个匹配项。我无法找到任何有状态 UDF 的示例，据我所知，这可能是一个坏主意，因为我们不知道最终在 Hadoop 上运行时如何映射/减少数据。

我可以通过存储键/时间戳值并编写一个解析这些值的 Python 脚本来分几步完成此操作。不过，我真的很想在 Pig 中尽可能多地保留这个过程。

编辑:最基本的数据是这样的:{id:long, timestamp:long} . timestamp以毫秒为单位。每组数据按 timestamp 排序.如果记录 X 正好落在最小值 timestamp 之后的 15 分钟边界上(开始时间)，捕获它。否则，只要有可能，在 15 分钟边界之后抓取下一个记录。我没有一个很好的例子来说明预期的结果是什么，因为我没有时间手动整理数据。

Answer 1

在 MapReduce 中满足条件“否则，在 15 分钟边界之后抓取下一条记录，无论何时”可能会很棘手，但是如果您将其稍微更改为“抓取该 15 分钟边界之前的上一条记录”比这可能很容易。想法是 15 分钟是 900000 毫秒，这样我们就可以将记录分组到涵盖 900000 毫秒的组中，对它们进行排序并取前一个。这是我脑海中的脚本示例:

inpt = LOAD '....' AS (id:long, timestamp:long);
intervals = FOREACH inpt GENERATE id, timestamp, timestamp / 900000 as interval;
grp = GROUP intervals BY interval;
result = FOREACH grp {
    sort = ORDER intervals BY timestamp DESC;
    top = LIMIT ord 1;
    GENERATE FLATTEN(top);
};