Lambdaを使用したDynamoDBのパラレルクエリ活用法

    response = table.scan(
        FilterExpression=Key('category').eq(category) & Key(
            'date').between(start_date, end_date),
        Segment=segment,
        TotalSegments=total_segments
    )

import boto3
from boto3.dynamodb.conditions import Key

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table("ParallelQuery")
start_date = '2022/01/01'
end_date = '2022/03/28'
category = 'books'


def seq_query_with_index():
    def query(ExclusiveStartKey=None):
        params = {
            "IndexName": 'category-date-idx',
            "KeyConditionExpression": Key('category').eq(category) & Key(
                'date').between(start_date, end_date),
            "ReturnConsumedCapacity": "TOTAL"
        }
        if ExclusiveStartKey is not None:
            params['ExclusiveStartKey'] = ExclusiveStartKey

        return table.query(**params)

    response = query()
    data = response['Items']
    consumed_rcu = response['ConsumedCapacity']['CapacityUnits']

    while 'LastEvaluatedKey' in response:
        response = query(response['LastEvaluatedKey'])
        data.extend(response['Items'])
        consumed_rcu += response['ConsumedCapacity']['CapacityUnits']

    return (data, consumed_rcu)

def lambda_handler(event, context):
    data, rcu = seq_query_with_index()

    print("RCU: "+ str(rcu))

    return {
        'statusCode': 200,
        'body': len(data)
    }

import boto3
import multiprocessing
from boto3.dynamodb.conditions import Key

query_date = ['2022/01/0', '2022/01/1', '2022/01/2', '2022/02/0',
              '2022/02/1', '2022/02/2', '2022/03/0', '2022/03/1', '2022/03/2']


dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table("ParallelQuery")

category = 'books'
total_segments = len(query_date)
datas = []

def para_query(conn, segment):
    def query(segment,ExclusiveStartKey=None):
        params = {
            "IndexName": 'category-date-idx',
            "KeyConditionExpression": Key('category').eq(category) & Key(
                'date').begins_with(query_date[segment]),
            "ReturnConsumedCapacity": "TOTAL"
        }
        if ExclusiveStartKey is not None:
            params['ExclusiveStartKey'] = ExclusiveStartKey

        return table.query(**params)
        
    response = query(segment)
    data = response['Items']

    consumed_rcu = response['ConsumedCapacity']['CapacityUnits']
    while 'LastEvaluatedKey' in response:
        response = query(segment,response['LastEvaluatedKey'])
        data.extend(response['Items'])
        consumed_rcu += response['ConsumedCapacity']['CapacityUnits']
    print('Segment ' + str(segment) + ' returned ' + str(len(data)) + ' items')

    conn.send((data, consumed_rcu))
    conn.close()

def lambda_handler(event, context):
    process_list = []
    parent_connection_list = []
    for i in range(total_segments):
        parent_conn, child_conn = multiprocessing.Pipe()
        parent_connection_list.append(parent_conn)
        process_list.append(multiprocessing.Process(
            target=para_query, args=(child_conn, i)))
    
    for process in process_list:
        process.start()
    
    data = []
    rcu = 0
    for parent_connection in parent_connection_list:
        d, consumed_rcu = parent_connection.recv()
        data.extend(d)
        rcu += consumed_rcu

    sorted_data = sorted(data, key=lambda x:x['date'])
    print("RCU: " + str(rcu))

    return {
        'statusCode': 200,
        'body': len(sorted_data)
    }

Segment 0 returned 4383 items
Segment 6 returned 4488 items
Segment 3 returned 4506 items
Segment 1 returned 4982 items
Segment 4 returned 4945 items
Segment 2 returned 4554 items
Segment 7 returned 4886 items
Segment 5 returned 4445 items
Segment 8 returned 4442 items