自動化CSV匯入DynamoDB與Lambda VPC存取

– 6 分鐘閱讀

在雲端原生架構中，利用 AWS Lambda 和 DynamoDB 可以實作高擴充套件性的資料函式庫解決方案。本文將示範如何使用 Lambda 自動將 S3 上的 CSV 檔案匯入 DynamoDB，並解決 Lambda 函式存取 VPC 內資源的常見問題。首先，建立具備 S3 讀取、DynamoDB 完全存取以及 Lambda 基本執行許可權的 IAM 角色。接著，撰寫 Python Lambda 函式，從 S3 事件觸發下載 CSV 檔案並將資料寫入 DynamoDB。同時，設定 S3 通知以在檔案上傳時自動觸發 Lambda 函式。此外，為了減少冷啟動延遲，我們會使用佈建並發功能預先初始化執行環境。最後，示範如何設定 Lambda 函式的 VPC 存取許可權，使其能夠連線到 VPC 內的 ElastiCache 叢集，並提供相關的 IAM 策略設定、Redis 套件安裝，以及 VPC 設定步驟。

自動化 CSV 匯入至 DynamoDB 從 S3 使用 Lambda

在現代雲端原生應用程式的開發中，使用 AWS Lambda 和 Amazon DynamoDB 可以實作大規模可擴充套件的資料函式庫持久化，同時最小化所需的維運負擔。軟體架構師和開發人員可以利用這兩個服務來構建應用程式，而無需擔心伺服器基礎設施。

步驟一：建立 IAM 角色並附加必要策略

首先，建立一個 IAM 角色供 Lambda 函式使用，並附加必要的 IAM 策略。

aws iam create-role --role-name AWSCookbook507Lambda \
--assume-role-policy-document '{"Version": "2012-10-17", "Statement": [{"Effect": "Allow", "Principal": {"Service": "lambda.amazonaws.com"}, "Action": "sts:AssumeRole"}]}'

內容解密：

此步驟建立了一個名為 AWSCookbook507Lambda 的 IAM 角色，並定義了信任策略，允許 Lambda 服務承擔該角色。"Version" 指定了策略語言的版本，而 "Statement" 中定義了允許 Lambda 服務承擔角色的操作。

接著，將 AmazonS3ReadOnlyAccess、AmazonDynamoDBFullAccess 和 AWSLambdaBasicExecutionRole 等 IAM 受管策略附加到該 IAM 角色。

aws iam attach-role-policy --role-name AWSCookbook507Lambda \
--policy-arn arn:aws:iam::aws:policy/AmazonS3ReadOnlyAccess

aws iam attach-role-policy --role-name AWSCookbook507Lambda \
--policy-arn arn:aws:iam::aws:policy/AmazonDynamoDBFullAccess

aws iam attach-role-policy --role-name AWSCookbook507Lambda \
--policy-arn arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole

內容解密：

這些步驟將必要的許可權附加到 AWSCookbook507Lambda 角色，使其能夠讀取 S3、存取 DynamoDB 以及執行 Lambda 函式所需的基本許可權。附加 AmazonDynamoDBFullAccess 提供了對 DynamoDB 的完全存取許可權，但最佳實踐是根據需要限制許可權到特定的 DynamoDB 表資源。

步驟二：建立 Lambda 函式並組態觸發器

建立一個名為 lambda_function.py 的檔案，並將其壓縮成 ZIP 檔案。

import boto3
import csv

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

def lambda_handler(event, context):
    # 從 S3 事件中取得 bucket 名稱和物件鍵
    bucket_name = event['Records'][0]['s3']['bucket']['name']
    object_key = event['Records'][0]['s3']['object']['key']

    # 從 S3 下載 CSV 檔案
    s3.download_file(bucket_name, object_key, '/tmp/' + object_key)

    # 將 CSV 資料匯入 DynamoDB
    table = dynamodb.Table('AWSCookbook507')
    with open('/tmp/' + object_key, 'r') as file:
        reader = csv.DictReader(file)
        for row in reader:
            table.put_item(Item=row)

    return {
        'statusCode': 200,
        'statusMessage': 'OK'
    }

內容解密：

此 Lambda 函式被 S3 物件上傳事件觸發，下載上傳的 CSV 檔案，並將其內容匯入到指定的 DynamoDB 表中。使用 boto3 與 AWS 服務進行互動。

建立 Lambda 函式並組態其相關設定。

LAMBDA_ARN=$(aws lambda create-function \
--function-name AWSCookbook507Lambda \
--runtime python3.8 \
--package-type "Zip" \
--zip-file fileb://lambda_function.zip \
--handler lambda_function.lambda_handler --publish \
--environment Variables={bucket=awscookbook507-$RANDOM_STRING} \
--role arn:aws:iam::$AWS_ACCOUNT_ID:role/AWSCookbook507Lambda \
--output text --query FunctionArn)

內容解密：

此命令建立了一個新的 Lambda 函式，指定了執行環境、處理常式以及相關的環境變數和執行角色。$LAMBDA_ARN 儲存了新建立的 Lambda 函式的 ARN。

步驟三：設定 S3 通知以觸發 Lambda 函式

建立一個名為 notification-template.json 的檔案，用於定義 S3 上傳事件觸發 Lambda 函式的組態。

{
    "LambdaFunctionConfigurations": [
        {
            "Id": "awscookbook507event",
            "LambdaFunctionArn": "LAMBDA_ARN",
            "Events": [
                "s3:ObjectCreated:*"
            ],
            "Filter": {
                "Key": {
                    "FilterRules": [
                        {
                            "Name": "prefix",
                            "Value": "sample_data.csv"
                        }
                    ]
                }
            }
        }
    ]
}

內容解密：

此 JSON 組態定義了當 sample_data.csv 被上傳到 S3 時觸發指定的 Lambda 函式。使用 sed 命令替換 LAMBDA_ARN 為實際的 Lambda 函式 ARN。

sed -e "s/LAMBDA_ARN/${LAMBDA_ARN}/g" notification-template.json > notification.json

接著，組態 S3 Bucket 的通知設定。

aws s3api put-bucket-notification-configuration \
--bucket awscookbook507-$RANDOM_STRING \
--notification-configuration file://notification.json

內容解密：

此命令將 S3 Bucket 的通知組態更新為使用剛才生成的 notification.json 檔案，實作當特定物件被上傳時自動觸發 Lambda 函式。

使用Lambda佈建並發來減少冷啟動時間

在無伺服器架構中，AWS Lambda 提供了一種全託管的服務，能夠根據需求自動擴充套件計算資源。然而，當 Lambda 函式第一次被呼叫或在一段時間內未被呼叫時，會出現「冷啟動」的情況，導致延遲。為瞭解決這個問題，AWS 提供了「佈建並發」功能，讓開發者能夠預先準備好多個執行環境，以減少冷啟動的發生。

解決方案

  1. 建立Lambda函式：首先，建立一個Lambda函式，並將其組態為使用佈建並發功能。

    aws lambda create-function --function-name AWSCookbook508Lambda \
--runtime python3.8 --handler lambda_function.lambda_handler \
--role arn:aws:iam::$AWS_ACCOUNT_ID:role/AWSCookbookLambdaRole

  2. 組態佈建並發：使用以下命令組態佈建並發。

    aws lambda put-provisioned-concurrency-config \
--function-name AWSCookbook508Lambda \
--qualifier LATEST \
--provisioned-concurrent-executions 5

  3. 驗證佈建並發的效果：透過多次呼叫Lambda函式來驗證佈建並發的效果。

    aws lambda invoke --function-name AWSCookbook508Lambda response.json &
aws lambda invoke --function-name AWSCookbook508Lambda response.json

程式碼解密：

  1. aws lambda create-function：建立一個新的Lambda函式。

    • --function-name：指定Lambda函式的名稱。
    • --runtime：指定執行環境，這裡使用Python 3.8。
    • --handler：指定處理函式的入口點。
    • --role：指定Lambda函式的執行角色。

  2. aws lambda put-provisioned-concurrency-config：組態Lambda函式的佈建並發。

    • --function-name：指定要組態的Lambda函式名稱。
    • --qualifier：指定要組態的版本或別名，這裡使用LATEST表示最新版本。
    • --provisioned-concurrent-executions：指定要佈建的並發執行數量。

  3. aws lambda invoke：呼叫Lambda函式。

    • --function-name：指定要呼叫的Lambda函式名稱。
    • response.json：將呼叫結果輸出到指定的檔案中。

討論

使用Lambda佈建並發功能，可以減少冷啟動帶來的延遲，從而提高應用的回應速度和並發處理能力。這對於需要毫秒級回應時間的微服務架構應用尤其重要。

挑戰

  1. 組態API Gateway和模擬使用者負載：在Lambda函式前組態API Gateway，並使用工具如bees with machine guns或ApacheBench模擬使用者負載，測試佈建並發的效果。

  2. 組態應用自動擴充套件：根據時間或效能指標（如回應時間）自動調整Lambda函式的佈建並發數量。

Lambda存取VPC資源

在某些情況下，Lambda函式需要存取位於VPC內的資源，如ElastiCache叢集。本文將介紹如何組態Lambda函式以存取VPC內的ElastiCache叢集。

解決方案

  1. 建立具有VPC存取能力的Lambda函式：

    • 為Lambda函式附加必要的IAM策略，以允許其在VPC內執行。
      aws iam attach-role-policy --role-name AWSCookbookLambdaRole \
--policy-arn arn:aws:iam::aws:policy/service-role/AWSLambdaVPCAccessExecutionRole
    • 安裝Redis Python包並封裝Lambda函式程式碼。
      pip install redis -t .
zip -r lambda_function.zip lambda_function.py redis*
    • 建立一個具有VPC組態的Lambda函式。
      LAMBDA_ARN=$(aws lambda create-function \
--function-name AWSCookbook509Lambda \
--runtime python3.8 \
--package-type "Zip" \
--zip-file fileb://lambda_function.zip \
--handler lambda_function.lambda_handler --publish \
--role arn:aws:iam::$AWS_ACCOUNT_ID:role/AWSCookbookLambdaRole \
--output text --query FunctionArn \
--vpc-config SubnetIds=${TRIMMED_ISOLATED_SUBNETS},SecurityGroupIds=$LAMBDA_SG_ID)

  2. 建立ElastiCache叢集：

    • 建立ElastiCache子網組。
      aws elasticache create-cache-subnet-group \
--cache-subnet-group-name "AWSCookbook509CacheSG" \
--cache-subnet-group-description "AWSCookbook509CacheSG" \
--subnet-ids $ISOLATED_SUBNETS
    • 建立ElastiCache Redis叢集。
      aws elasticache create-cache-cluster \
--cache-cluster-id "AWSCookbook509CacheCluster" \
--cache-subnet-group-name AWSCookbook509CacheSG \
--engine redis \
--cache-node-type cache.t3.micro \
--num-cache-nodes 1

  3. 呼叫Lambda函式：將ElastiCache叢集的端點傳遞給Lambda函式，以測試其連線性。

    aws lambda invoke \
--cli-binary-format raw-in-base64-out \
--function-name $LAMBDA_ARN \
--payload '{ "hostname": "HOSTNAME" }' \
response.json && cat response.json

程式碼解密：

  1. aws iam attach-role-policy：為Lambda函式的執行角色附加必要的IAM策略，以允許其在VPC內執行。

  2. pip install redis -t . 和 zip -r lambda_function.zip lambda_function.py redis*：安裝Redis Python包並將其與Lambda函式程式碼一起封裝。

  3. aws lambda create-function（具有VPC組態）：建立一個具有VPC存取能力的Lambda函式。

  4. aws elasticache create-cache-subnet-group 和 aws elasticache create-cache-cluster：建立ElastiCache子網組和Redis叢集。

  5. aws lambda invoke：呼叫Lambda函式，並傳遞ElastiCache叢集的端點，以測試其連線性。

討論

預設情況下，Lambda函式無法存取VPC內的資源。透過組態VPC存取能力，Lambda函式可以存取如ElastiCache這樣的VPC內資源。此外，使用ElastiCache可以為應用提供記憶體永續性儲存，改善效能和可擴充套件性。

挑戰

  1. 擴充套件Lambda函式以讀寫ElastiCache中的資料：進一步開發Lambda函式，使其能夠讀取和寫入ElastiCache中的資料，以實作更複雜的應用邏輯。
玄貓 BlackCat

玄貓 BlackCat

技術愛好者，專注於分享程式開發、雲端技術與 AI 應用的心得體會。