使用自定义逻辑处理来自 Redshift 的数十亿条记录答案

【问题标题】：Process several billion records from Redshift using custom logic使用自定义逻辑处理来自 Redshift 的数十亿条记录
【发布时间】：2020-01-22 18:13:47
【问题描述】：

我想对 Redshift 中的数据集应用自定义逻辑。输入数据示例：

userid, event,  fileid, timestamp, ....
100000, start,  120,    2018-09-17 19:11:40
100000, done,   120,    2018-09-17 19:12:40
100000, done,   120,    2018-09-17 19:13:40
100000, start,  500,    2018-09-17 19:13:50
100000, done,   120,    2018-09-17 19:14:40
100000, done,   500,    2018-09-17 19:14:50
100000, done,   120,    2018-09-17 19:15:40

这意味着类似：

file 120:  start-----done-----done-----done-----done 
file 150:                      start-----done   
time    :  11:40----12:40----13:40-----14:40-----15:40

但它应该看起来像

file 120:  start-----done-----done 
file 150:                      start-----done   
time    :  11:40----12:40----13:40-----14:40-----15:40

一旦文件 150 启动，文件 120 就被中断

请记住，这里有很多不同的用户和许多不同的文件。

清理后的数据应该是：

userid, event,  fileid, timestamp, ....
100000, start,  120,    2018-09-17 19:11:40
100000, done,   120,    2018-09-17 19:12:40
100000, done,   120,    2018-09-17 19:13:40
100000, start,  500,    2018-09-17 19:13:50
100000, done,   500,    2018-09-17 19:14:50

不应为同一用户同时拥有多个并发文件。因此，在第二个开始后，不应从当前数据集中删除来自第一个的事件。

代码很简单，但在 python 上，它很容易为 Google Dataflow 扩展，例如，将 100GB+ 从 AWS 转移到 GC 不是一个好主意。

问题 #1： 是否可以在 SQL 上执行（使用 postgres/redshift 特定功能）或更好地使用 Spark？（但不确定如何在那里实现）

问题 #2： 任何关于使用 AWS Batch 或其他可能更好的建议，因为使用 apache Beam - 这很容易而且非常明显，但是 AWS Batch 如何工作以及如何将数据集划分为块（如每个用户的组） - 这是一个大问题。我的建议是以某种方式将数据从 redshift 卸载到 S3 存储桶中，但以单独的文件 = 用户的方式将其划分，然后如果 aws 批处理支持这一点 - 只需提供存储桶，并且每个文件都应该在已创建的实例上同时处理。不确定这是否有意义。

【问题讨论】：

。请从样本数据中显示您想要的结果。目前尚不清楚您要生产什么。
@GordonLinoff 我已经编辑了内容并添加了结果。谢谢。

标签： sql apache-spark bigdata amazon-redshift google-cloud-dataflow

【解决方案1】：

如果要删除fileid 与用户最近的start 不匹配的行，可以使用lag(ignore nulls)：

select t.*
from (select t.*,
             lag(case when event = 'start' then file_id end ignore nulls) over (partition by userid order by timestamp) as start_fileid
      from t
     ) t
where event = 'start' or start_fileid = fileid;

【讨论】：