ホームページ >バックエンド開発 >Python チュートリアル >Lambda 関数を使用した So DynamoDB からのデータの解析とロード

Lambda 関数を使用した So DynamoDB からのデータの解析とロード

Susan Sarandonオリジナル: 2025-01-06 06:24:47231ブラウズ

多くのシナリオでは、JSON としてフォーマットされたデータを操作する必要があり、データを抽出して処理し、将来使用できるようにテーブルに保存したいと考えています。

この記事では、Lambda 関数を使用して JSON 形式のデータを S3 バケットから DynamoDB テーブルにロードする方法について説明します

前提条件

オブジェクトを S3 にアップロードする権限を持つ IAM ユーザー
S3 および DynamoDB へのアクセス許可を持つ Lambda 実行ロール

アーキテクチャとコンポーネント

以下のアーキテクチャは、3 つの AWS サービスを使用していることを示しています

S3 バケット
ラムダ関数
DynamoDB テーブル

Parsing & Loading Data from So DynamoDB with Lambda Function

リフレッシュとして以下のサービスの簡単な説明:

S3 バケット: データのストレージサービスとして、拡張性、セキュリティ、高パフォーマンスのサービスを備えたオブジェクトストレージサービスが使用されます
Lambda 関数: インフラストラクチャを気にせずにコードを実行できるサーバーレスコンピューティングサービス。セットアップが簡単で、多くのプログラミング言語をサポートします。これを利用してコードを実行し、ロジックをデプロイします。
DynamoDB: データをテーブルに保存するために使用されるサーバーレス NoSQL データベース。Lambda 関数によって処理されたデータを保存するために使用します

流れ

ユーザーは、舞台裏で PutObject API を実行するコンソールまたは CLI を介して JSON ファイルを S3 バケットにアップロードします
オブジェクトが正常にアップロードされ、S3 イベントがトリガーされてラムダ関数が呼び出され、ファイルをロードして処理されます
Lambda はデータを処理し、DynamoDB テーブルにロードします

実装手順

上記の図をデプロイするための手順と構成を順に見ていきます

1- 以下の構成で Lambda 関数を作成します

ゼロから作成
関数名: ParserDemo
ランタイム: Python 3.1x

残りはデフォルトのままにしておきます
Lambda を作成した後、以下のようにタイムアウト設定と実行ロールを変更する必要があります:

Parsing & Loading Data from So DynamoDB with Lambda Function

ロジックを実行するためにこの Python コードを書きました

import json
import boto3

s3_client = boto3.client('s3')
dynamodb = boto3.resource('dynamodb')

def lambda_handler(event, context):



    bucket_name = event['Records'][0]['s3']['bucket']['name'] # Getting the bucket name from the event triggered by S3
    object_key = event['Records'][0]['s3']['object']['key'] # Getting the Key of the item when the data is uploaded to S3
    print(f"Bucket: {bucket_name}, Key: {object_key}")


    response = s3_client.get_object(
    Bucket=bucket_name,
    Key=object_key
)


    # We will convert the streamed data into bytes
    json_data = response['Body'].read()
    string_formatted = json_data.decode('UTF-8') #Converting data into string

    dict_format_data = json.loads(string_formatted) #Converting Data into Dictionary 


    # Inserting Data Into DynamoDB

    table = dynamodb.Table('DemoTable')
    if isinstance(dict_format_data, list): #check if the file contains single record
        for record in dict_format_data:
            table.put_item(Item=record)

    elif isinstance(dict_format_data, dict): # check if the file contains multiple records 
        table.put_item(Item=data)

    else:  
        raise ValueError("Not Supported Format") # Raise error if nothing matched

2- S3 バケットを作成します

BucketName: 一意の名前を使用します

残りの構成はデフォルトのままにしておきます

以下のように、作成した S3 バケットをトリガーとしてラムダ関数に追加します。

Parsing & Loading Data from So DynamoDB with Lambda Function