并行执行DynamoDB查询(全局二级索引的BatchGetItems)
Posted
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了并行执行DynamoDB查询(全局二级索引的BatchGetItems)相关的知识,希望对你有一定的参考价值。
这里的想法是在GSI上运行查询时并行运行多个DynamoDB查询。截至目前BatchGetItems doesn't support querying over Indexes和推荐的方法是并行查询数据。我正在使用wg的例程来并行处理例程的执行。
函数的输入是带有ID的字符串数组,输出是Ids的属性。
当函数在本地运行时,没有问题,但是,当在AWS-Lambda上运行该函数时,返回的数据会不断增长;
即;输入2项应输出2项。如果在AWS-Lambda上测试该功能,
- 第一次该函数返回2项
- 第二次返回4个项目(相同的项目重复2次)
- 第3次返回6个项目(相同的项目重复4次)
等等。这是代码的片段。有没有正确处理的东西,每次运行lambda时,lambda输出都会有额外的数据集?
package main
import (
"context"
"fmt"
"os"
"sync"
"github.com/aws/aws-lambda-go/lambda"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/aws/session"
"github.com/aws/aws-sdk-go/service/dynamodb"
"github.com/aws/aws-sdk-go/service/dynamodb/dynamodbattribute"
)
//Final Output Interface
var bulkOutput []interface{}
func exitWithError(err error) {
fmt.Fprintln(os.Stderr, err)
os.Exit(1)
}
//LambdaInputJSON input for the lambda handler
type LambdaInputJSON struct {
Ids []string `json:"ids,omitempty"`
}
//HandleRequest : Lambda entry point
func HandleRequest(ctx context.Context, data LambdaInputJSON) ([]interface{}, error) {
return DynamoDBBatchGetRecords(data), nil
}
func main() {
lambda.Start(HandleRequest)
}
func DynamoDBBatchGetRecords(a LambdaInputJSON) []interface{} {
var wg sync.WaitGroup
var mutex = &sync.Mutex{}
iterations := len(a.Ids)
wg.Add(iterations)
for i := 0; i < iterations; i++ {
go QueryOutput(a.Ids[i], &wg, mutex)
}
wg.Wait()
return bulkOutput
}
//QueryOutput GoRoutine
func QueryOutput(data string, wg *sync.WaitGroup, mtx *sync.Mutex) {
var outputData []interface{}
defer wg.Done()
sess, err := session.NewSession(&aws.Config{
Region: aws.String("aws-region"),
})
if err != nil {
exitWithError(fmt.Errorf("failed to make Query API call, %v", err))
}
ddb := dynamodb.New(sess)
queryInput := &dynamodb.QueryInput{
Limit: aws.Int64(1),
TableName: aws.String("table-name"),
IndexName: aws.String("gsi-index"),
ScanIndexForward: aws.Bool(false),
ConsistentRead: aws.Bool(false),
KeyConditions: map[string]*dynamodb.Condition{
"column_name": {
ComparisonOperator: aws.String("EQ"),
AttributeValueList: []*dynamodb.AttributeValue{
{
S: aws.String(data),
},
},
},
},
}
output, err := ddb.Query(queryInput)
if err != nil {
exitWithError(fmt.Errorf("Failed to make Query API call, %v", err))
}
err = dynamodbattribute.UnmarshalListOfMaps(output.Items, &outputData)
if err != nil {
exitWithError(fmt.Errorf("Failed to unmarshal Query result items, %v", err))
}
mtx.Lock()
bulkOutput = append(bulkOutput, outputData)
mtx.Unlock()
}
答案
根据documentation,全局变量独立于Lambda函数的处理程序代码。这导致缓冲区随着时间的推移而积累。
纠正参考贴在下面。
package main
import (
"context"
"fmt"
"os"
"sync"
"github.com/aws/aws-lambda-go/lambda"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/aws/session"
"github.com/aws/aws-sdk-go/service/dynamodb"
"github.com/aws/aws-sdk-go/service/dynamodb/dynamodbattribute"
)
func exitWithError(err error) {
fmt.Fprintln(os.Stderr, err)
os.Exit(1)
}
//HandleRequest : Lambda entry point
func HandleRequest(ctx context.Context, data LambdaInputJSON) ([]interface{}, error) {
output := DynamoDBBatchGetRecords(data)
return output, nil
}
func main() {
lambda.Start(HandleRequest)
}
func DynamoDBBatchGetRecords(a LambdaInputJSON) []interface{} {
var dataOut []interface{}
var wg = &sync.WaitGroup{}
var mtx = &sync.Mutex{}
iterations := len(a.Ids)
wg.Add(iterations)
for i := 0; i < i; i++ {
go func(i int) {
defer wg.Done()
var outputData []interface{}
sess, err := session.NewSession(&aws.Config{
Region: aws.String("aws-region"),
})
if err != nil {
exitWithError(fmt.Errorf("failed to make Query API call, %v", err))
}
ddb := dynamodb.New(sess)
queryInput := &dynamodb.QueryInput{
Limit: aws.Int64(1),
TableName: aws.String("table"),
IndexName: aws.String("index"),
ScanIndexForward: aws.Bool(false),
ConsistentRead: aws.Bool(false),
KeyConditions: map[string]*dynamodb.Condition{
"index-column": {
ComparisonOperator: aws.String("EQ"),
AttributeValueList: []*dynamodb.AttributeValue{
{
S: aws.String(a.Ids[i]),
},
},
},
},
}
output, err := ddb.Query(queryInput)
if err != nil {
exitWithError(fmt.Errorf("E1 failed to make Query API call, %v", err))
}
err = dynamodbattribute.UnmarshalListOfMaps(output.Items, &outputData)
if err != nil {
exitWithError(fmt.Errorf("E2 failed to unmarshal Query result items, %v", err))
}
mtx.Lock()
dataOut = append(dataOut, outputData[0])
mtx.Unlock()
}(i)
}
wg.Wait()
return dataOut
}
以上是关于并行执行DynamoDB查询(全局二级索引的BatchGetItems)的主要内容,如果未能解决你的问题,请参考以下文章
.net 通过 DynamoDBContext 查询 DynamoDB 中的全局二级索引