自動匯入DynamoDB中的s3桶資料

如何使用NODEJS、DynamoDB和AWS lambda在DynamoDB中自動匯入s3 bucket JSON資料。

uj5u.com熱心網友回復：

import type { AWS } from '@serverless/typescript';

const serverlessConfiguration: AWS = {
  service: 'raj',
  frameworkVersion: '2',
  自定義。{
    webpack: {
      webpackConfig: './webpack.config.js',
      includeModules: true,
    },
  },
  插件。['serverless-webpack']。
  提供者。{
    name: 'aws',
    runtime : 'nodejs14.x',
    profile : 'server',
    apiGateway。{
      minimumCompressionSize：1024。
      shouldStartNameWithService: true,
    },
    環境。{
      aws_nodejs_connection_reuse_enabled: '1',
    },
    lambdaHashingVersion: '20201221',
  },
  // 通過路徑匯入函式
  函式。{
    messageAdd : {
      handler : "src/now.handler",
      事件。[
        {
          http：{
            path : 'addData',
            method : 'POST',
            cors : true,
          }
        }
      ]
    }
   },
};

module.exports = serverlessConfiguration。

uj5u.com熱心網友回復：

const AWS = require('aws-sdk') ;
const docClient = new AWS.DynamoDB.DocumentClient（）。

// Lambda處理程式
exports.handler = async (event) => {
  AWS.config.update({
    region: 'us-east-1', // Use appropriate region
    accessKeyId: '', // 使用你的訪問密鑰
    secretAccessKey: '' // 使用你的秘密密鑰
});
  const s3 = new AWS.S3();
  const ddbTable = "s3todyb";
  console.log(JSON.stringify(event, null, 2))。
  console.log('Using DDB table: ', ddbTable);

  await Promise.all(
    event.Records.map(async (record) => {
      try {
        console.log('Incoming record: ', record);

        // 從傳入事件的物件中獲取原始文本
        const originalText = await s3.getObject({
          Bucket: event.Records[0].s3.bucket.name,
          key: event.Records[0].s3.object.key
        }).promise()。

        //上傳JSON到DynamoDB
        const jsonData = JSON.parse(originalText.Body.toString('utf-8'))。
        await ddbLoader(jsonData);

      } catch (err) {
        console.error(err)。
      }
    })
  );
};

// 將JSON資料加載到DynamoDB表中
const ddbLoader = async (data) => {
  // 分離成批次進行上傳
  let batches = [];
  const BATCH_SIZE = 25;

  while (data.length > 0) {
    batches.push(data.splicice(0, BATCH_SIZE))。
  }

  console.log(`Total batches: ${batches.length}`);

  讓 batchCount = 0;

  // 保存每個批次
  await Promise.all(
    batches.map(async (item_data) => {

      // 設定DDB呼叫的params物件
      const params = {
        RequestItems: {}
      };
      params.RequestItems[ddbTable] = [];

      item_data.forEach(item => {
        for (let key of Object.keys(item)) {
          // 一個AttributeValue不能包含一個空字串
          if (item[key] === '') 
            洗掉 item[key]。
        }

        // 建立引數
        params.RequestItems[ddbTable].push({
          PutRequest: {
            專案: {
              ...項
            }
          }
        });
      });

      // 分批推送到DynamoDB
      try {
        batchCount  ;
        console.log('Trying batch: ', batchCount);
        const result = await docClient.batchWrite(params).promise()。
        console.log('Success: ', result);
      } catch (err) {
        console.error('Error: ', err)。
      }
    })
  );

};

標籤：

上一篇：從AWSS3桶中讀取JKS檔案并檢查證書的到期日？

下一篇：Lambda無法檢測到該層