Kinesis lambda DynamoDB

我正在学习 AWS 服务的一个用例。在浏览完文档后，我想出了一个简单的流程。我想使用 Streams API 和 KPL 将数据提取到 Kinesis 流中。我使用示例 putRecord 方法将数据提取到流中。我正在将此 JSON 摄取到流中 -

{"userid":1234,"username":"jDoe","firstname":"John","lastname":"Doe"}

数据被摄取后，我会在 putRecordResult 中得到以下响应 -

Put Result :{ShardId: shardId-000000000000,SequenceNumber: 49563097246355834103398973318638512162631666140828401666}
Put Result :{ShardId: shardId-000000000000,SequenceNumber: 49563097246355834103398973318645765717549353915876638722}
Put Result :{ShardId: shardId-000000000000,SequenceNumber: 49563097246355834103398973318649392495008197803400757250}

现在，我编写一个 Lambda 函数来获取这些数据并将其推送到 DynamoDB 表中。这是我的 Lambda 函数 -

console.log('Loading function');
var AWS = require('aws-sdk');
var tableName = "sampleTable";
var doc = require('dynamodb-doc');
var db = new doc.DynamoDB();

exports.handler = (event, context, callback) => {
    //console.log('Received event:', JSON.stringify(event, null, 2));
    event.Records.forEach((record) => {
        // Kinesis data is base64 encoded so decode here
        const payload = new Buffer(record.kinesis.data, 'base64').toString('ascii');
        console.log('Decoded payload:', payload);
        var userid = event.userid;
        var username = event.username;
        var firstname = event.firstname;
        console.log(userid + "," + username +","+ firstname);

        var item = {
            "userid" : userid,
            "username" : username,
            "firstname" : firstname
        };

        var params = {
            TableName : tableName,
            Item : item
        };
        console.log(params);

        db.putItem(params, function(err, data){
            if(err) console.log(err);
            else console.log(data);
        });

    });
    callback(null, `Successfully processed ${event.Records.length} records.`);
};

不知何故，我无法在 lambda 函数执行中看到 console.logs。我在流页面中看到，已将 putRecord 写入流并获取，但不知何故，我在 Lambdafunction 页面和 DynamoDB 表中看不到任何内容。

我有一个针对 Java 代码的 IAM 策略，用于将数据提取到 Kinesis 中，另一个针对 Lambda 函数（即 lambda-kinesis-execution-role），还有一个针对 DynamoDB 的策略，用于将数据提取到表中。

有没有任何教程可以展示如何以正确的方式完成它？我感觉我在这个过程中遗漏了很多要点，例如如何链接所有这些 IAM 策略并使它们同步，以便当数据放入流中时由 Lambda 处理并最终进入 Dynamo？

非常感谢任何指示和帮助。

如果您上面的代码是您正在使用的代码的直接副本，那么您正在引用event.userid但你应该使用payload.userid。您已将 Kinesis 记录解码到有效负载变量中。