S3 事件触发器是否可扩展？

Question

将大约 3K 个对象（文件）加载到 S3。有一个事件会触发加载到该 S3 存储桶的每个文件。

Lambda 仅接收大约 300 个对象的事件触发器。如果我重试（从 S3 移回并将其放回 S3），它会为另外 400 个对象生成事件，其余事件甚至没有到达 lambda。

我在这里缺少什么，我如何为创建的任意数量的对象进行缩放？

var async = require('async');                                                                                                                                                                                
var aws = require('aws-sdk');                                                                                                                                                                                
var s3 = new aws.S3();                                                                                                                                                                                       
var kinesis = new aws.Kinesis();                                                                                                                                                                             
var sns = new aws.SNS();                                                                                                                                                                                     
var config = require('./config.js');                                                                                                                                                                         


var logError = function(errormsg) {                                                                                                                                                                          
    sns.publish({                                                                                                                                                                                            
        TopicArn: config.TopicArn,                                                                                                                                                                           
        Message: errormsg                                                                                                                                                                                    
    }, function(err, data) {                                                                                                                                                                                 
        if (err) {                                                                                                                                                                                           
            console.log(errormsg);                                                                                                                                                                           
        }                                                                                                                                                                                                    
    });                                                                                                                                                                                                      
};                                                                                                                                                                                                           


exports.handler = function(event, context, callback) {                                                                                                                                                       

    var readS3andSendtoKinesis = function(record, index, cb) {                                                                                                                                               
        var params = {                                                                                                                                                                                       
            Bucket: record.s3.bucket.name,                                                                                                                                                                   
            Key: record.s3.object.key                                                                                                                                                                        
        }; 
        console.log('Received File: ' +  record.s3.object.key);                                                                                                                                                                                                 
        s3.getObject(params, function(err, data) {                                                                                                                                                           
            if (!err) {                                                                                                                                                                                      
                var kinesisParams = {                                                                                                                                                                        
                    Data: data.Body.toString('utf8'),                                                                                                                                                        
                    PartitionKey: config.PartitionKey,                                                                                                                                                       
                    StreamName: config.StreamName                                                                                                                                                            
                };                                                                                                                                                                                           
                kinesis.putRecord(kinesisParams, function(err, data) {                                                                                                                                       
                    if (err) {                                                                                                                                                                               
                        // Handle Kinesis Failures                                                                                                                                                           
                        logError(JSON.stringify(err, null, 2));                                                                                                                                              
                    }                                                                                                                                                                                        
                    cb(null, 'done');                                                                                                                                                                        
                });                                                                                                                                                                                          
            } else {                                                                                                                                                                                         
                // Handle S3 Failures                                                                                                                                                                        
                logError(JSON.stringify(err, null, 2));                                                                                                                                                      
                cb(null, 'done');                                                                                                                                                                            
            }                                                                                                                                                                                                
        });                                                                                                                                                                                                  
    };                                                                                                                                                                                                       

    async.eachOfLimit(event.Records, 1, readS3andSendtoKinesis, function(err) {                                                                                                                              
        callback(null, 'Done');                                                                                                                                                                              
    });                                                                                                                                                                                                      
}; 

既然大家都推荐看cloudwatch，这里分享一下相关lambda的cloudwatch指标，

Answer 1

我们发现根本原因似乎是在资源的另一端出现故障。S3 触发器发生，无法扩展到它收到的巨大触发器。

为了解决，

尽快返回 S3 Lambda 触发器，延迟会导致问题。

如果您花费太多时间处理触发器内的业务逻辑，在我们的例子中，我们从 S3 读取数据并写入流。相反，我们只是写入 S3 的位置并在接收端从 S3 读取。

希望能帮助到你。