有没有一种方法可以使用Lambda进行S3文件拼接?
我正在使用Firehose将数据流式传输到S3,采用最长的时间间隔(15分钟或128mb),因此每天会生成96个数据文件,但我希望将所有数据聚合到单个每日数据文件中,以便在稍后使用Spark(EMR)读取数据时获得最快的性能。
我创建了一个解决方案,其中当Firehose将新文件流式传输到S3时,会调用Lambda函数。然后该函数从源存储桶中读取(s3.GetObject)新文件和目标存储桶中的已连接每日数据文件(如果已存在,则连接之前的每日数据文件,否则创建一个新文件)。接着,将两个响应正文解码为字符串并将它们相加,然后使用s3.PutObject将其写入目标存储桶(覆盖之前的聚合文件)。
问题是,当聚合文件达到150+ MB时,Lambda函数在读取两个文件时就达到了其约1500MB的内存限制,然后失败了。
目前我只有少量数据,每天只有几百MB,但这个数量将来会呈指数级增长。对于我来说,Lambda的限制非常低,并且它们已经在处理如此小的文件时达到了极限,这很奇怪。
还有什么其他的方法可以拼接S3数据,最好是由S3对象创建事件调用或通过某种定期任务(例如每日定时)来调用?
我正在使用Firehose将数据流式传输到S3,采用最长的时间间隔(15分钟或128mb),因此每天会生成96个数据文件,但我希望将所有数据聚合到单个每日数据文件中,以便在稍后使用Spark(EMR)读取数据时获得最快的性能。
我创建了一个解决方案,其中当Firehose将新文件流式传输到S3时,会调用Lambda函数。然后该函数从源存储桶中读取(s3.GetObject)新文件和目标存储桶中的已连接每日数据文件(如果已存在,则连接之前的每日数据文件,否则创建一个新文件)。接着,将两个响应正文解码为字符串并将它们相加,然后使用s3.PutObject将其写入目标存储桶(覆盖之前的聚合文件)。
问题是,当聚合文件达到150+ MB时,Lambda函数在读取两个文件时就达到了其约1500MB的内存限制,然后失败了。
目前我只有少量数据,每天只有几百MB,但这个数量将来会呈指数级增长。对于我来说,Lambda的限制非常低,并且它们已经在处理如此小的文件时达到了极限,这很奇怪。
还有什么其他的方法可以拼接S3数据,最好是由S3对象创建事件调用或通过某种定期任务(例如每日定时)来调用?
read.json
,对于1个文件需要约15秒,对于小文件需要约10分钟(我现在不记得实例类型了)。我想使用Spark来完成两件事:统计和最终机器学习。对于统计,我想过滤和分组数据并获得不同的计数。我认为保持Spark EMR集群24/7运行会产生比S3更高的成本。 - V. Samma