將 DynamoDB 與 Amazon Managed Streaming for Apache Kafka 整合

執行 CloudFormation 範本 S3bucket.yaml 以建立 S3 儲存貯體。對於任何後續指令碼或操作，請在相同區域中執行它們。輸入 ForMSKTestS3 做為 CloudFormation 堆疊名稱。

完成後，記下輸出下方的 S3 儲存貯體名稱輸出。您在步驟 3 中將需要此名稱。

將下載的 ZIP 檔案 fromMSK.zip 上傳至您剛建立的 S3 儲存貯體。

執行 CloudFormation 範本 VPC.yaml 來建立 VPC、Amazon MSK 叢集和 Lambda 函式。在參數輸入畫面上，輸入您在步驟 1 中被要求提供 S3 儲存貯體時建立的 S3 儲存貯體名稱。將 CloudFormation 堆疊名稱設定為 ForMSKTestVPC。

準備在 CloudShell 中執行 Python 指令碼的環境。您可以在 AWS 管理主控台 上使用 CloudShell。如需使用 CloudShell 的詳細資訊，請參閱 AWS CloudShell 入門。啟動 CloudShell 後，建立屬於您剛建立的 VPC 的 CloudShell，以連線至 Amazon MSK 叢集。在私有子網路中建立 CloudShell。填寫下列欄位：

屬於私有子網路的 CloudShell 啟動後，請執行下列命令：

pip install boto3 kafka-python aws-msk-iam-sasl-signer-python

從 S3 儲存貯體下載 Python 指令碼。

aws s3 cp s3://[YOUR-BUCKET-NAME]/pythonScripts.zip ./
unzip pythonScripts.zip

檢查管理主控台，並在 Python 指令碼中設定代理程式 URL 和區域值的環境變數。在管理主控台中檢查 Amazon MSK 叢集代理程式端點。

在 CloudShell 上設定環境變數。如果您使用的是美國西部 (奧勒岡)：

export AWS_REGION="us-west-2"
export MSK_BROKER="boot-YOURMSKCLUSTER.c3.kafka-serverless.ap-southeast-1.amazonaws.com:9098"

執行下列 Python 指令碼。

建立 Amazon MSK 主題：

python ./createTopic.py

建立 DynamoDB 資料表：

python ./createTable.py

將測試資料寫入 Amazon MSK 主題：

python ./kafkaDataGen.py

檢查所建立 Amazon MSK、Lambda 和 DynamoDB 資源的 CloudWatch 指標，並使用 DynamoDB Data Explorer 驗證存放在 device_status 資料表中的資料，以確保所有程序都正確執行。如果每個程序執行均無錯誤，您可以檢查從 CloudShell 寫入 Amazon MSK 的測試資料是否也會寫入 DynamoDB。

當您完成此範例時，請刪除本教學課程中建立的資源。刪除兩個 CloudFormation 堆疊：ForMSKTestS3 和 ForMSKTestVPC。如果堆疊刪除成功完成，則會刪除所有資源。