🌺 AWS CDK 101 — 🚂 Dynamodb stream triggering batch deleteItem on dynamodb table

🔰 Начинающие пользователи AWS CDK, пожалуйста, ознакомьтесь с моими предыдущими статьями в этом цикле.

Если вы пропустили предыдущую статью, найдите ее по ссылкам ниже.

🔁 Оригинальный предыдущий пост 🔗 Dev Post

🔁 Reposted previous post at 🔗 dev to @aravindvcyber

В этой статье мы усовершенствуем мою предыдущую статью, в которой было показано использование пакетного потока dynamodb для удаления элементов из другой таблицы. Здесь мы преобразуем простое действие deleteItem в действие batchWrite, которое имеет большие преимущества.

Преимущества данного подхода 💦

• Как уже говорилось в предыдущей статье, мы стараемся сделать эту настройку очистки как можно более эффективной.
• Эта интеграция полностью асинхронна и не блокирует существующий поток процессов, а также использует наименьшее количество вызовов API и ресурсов, о чем будет сказано ниже.
• Привлечение batchWrite заставляет нас выполнять операцию удаления на максимальном размере чанка 25.
• Поскольку вместо 25 вызовов используется только один запрос, можно избежать сценариев дросселирования запросов на запись, когда мы имеем меньшие предоставленные мощности.
• К этому времени вы также должны проверить UnprocessedItems, чтобы повторить все неудачные ключи, если они возвращаются в худшем случае.
• Резюмируя предыдущую статью, мы теперь можем выполнять пакетное получение объекта потока с размером 100 max, который, в свою очередь, выполняет пакетную запись 25 max за раз.
• Меньше вызовов обработчиков и меньше вызовов API dynamodb.

Планирование и построение 🚣

Как уже упоминалось, мы оптимизируем предыдущую вспомогательную функцию deleteItem и преобразуем вызов нашего потока в действие пакетной записи в таблицу dynamodb с кусками по 25 максимум.

Начнем с того, что немного преломим мою предыдущую лямбду, чтобы достичь желаемых целей по оптимизации эффектов.

Здесь вы можете обнаружить, что мы нацелились на имя события INSERT, таким образом, мы можем более тонко контролировать желаемый результат во время этих вызовов потока, как показано ниже.

Необходимые импорты 💮

В этой статье используются следующие импорты.

import { DynamoDBStreams } from "aws-sdk";
import {
  DeleteItemInput,
  BatchWriteItemInput,
  WriteRequest,
} from "aws-sdk/clients/dynamodb";

Новая логика функции-обработчика ⚓

Здесь мы оптимизировали лямбда-обработчик следующим образом. Это включает создание keyMap из данных потока и дальнейшую нарезку на фрагменты максимум 25 для выполнения операции batchWrite, вызывающей наш вспомогательный метод.

exports.created = async function (event: any) {
  console.log("Received stream:", JSON.stringify(event, undefined, 2));

  const keyMap: any[] = [];
  event.Records.map((Record: DynamoDBStreams.Record) => {
    console.log(JSON.stringify(Record, undefined, 2));
    if (Record.eventName === "INSERT") {
      keyMap.push(Record.dynamodb?.Keys);
    }
  });

  const chunkList = [...chunks(keyMap, 25)];
  await Promise.all(
    chunkList.map(async (chunk: any[]) => {
      const results = await batchDeleteDbItems(chunk);
      Object.entries(results).forEach((entry) => {
        console.log(JSON.stringify(entry, undefined, 2));
      });
    })
  );
};

Вспомогательная функция dynamodb deleteItem 💐

Простая вспомогательная функция для выполнения deleteItem из таблицы dynamodb.

Здесь мы используем список ключей для генерации коллекции WriteRequest действия Delete следующим образом.

const batchDeleteDbItems: any = async (keys: any) => {
  console.log("Deleting: ", { keys });

  const writeItems: WriteRequest[] = [];
  keys.map((key: any) => {
    const writeItem: WriteRequest = {
      DeleteRequest: {
        Key: {
          ...key,
        },
      },
    };
    writeItems.push(writeItem);
  });
  const params: BatchWriteItemInput = {
    RequestItems: {
      stgMessagesTable: writeItems,
    },
    ReturnConsumedCapacity: "TOTAL",
    ReturnItemCollectionMetrics: "SIZE",
  };

  console.log("deleteItem: ", JSON.stringify(params, undefined, 2));

  return await dynamo.batchWriteItem(params).promise();
};

Незначительные изменения в определении таблицы dynamodb 🍊

Я выделил необходимые изменения, которые нужны для генерации потока dynamodb для нашей таблицы.

Самое главное, я запросил только ключи, которые будут содержать все необходимые нам данные.

Кроме того, в этой статье мы стараемся оптимизировать как можно больше для наименьшей занимаемой площади.

const messages = new dynamodb.Table(this, "MessagesTable", {
      tableName: process.env.messagesTable,
      sortKey: { name: "createdAt", type: dynamodb.AttributeType.NUMBER },
      partitionKey: { name: "messageId", type: dynamodb.AttributeType.STRING },
      encryption: dynamodb.TableEncryption.AWS_MANAGED,
      readCapacity: 5,
      writeCapacity: 5,
      stream: dynamodb.StreamViewType.KEYS_ONLY
});

Образец потока dynamodb с одной записью 🥣

Ниже я представил объект потока dynamodb, используемый в качестве полезной нагрузки для вызова нашей лямбды-обработчика.

{
  "eventID": "961320567faf6d890d6498a08fd1f34c",
  "eventName": "INSERT",
  "eventVersion": "1.1",
  "eventSource": "aws:dynamodb",
  "awsRegion": "ap-south-1",
  "dynamodb": {
    "ApproximateCreationDateTime": 1652637075,
    "Keys": {
      "createdAt": {
        "N": "1652637067000"
      },
      "messageId": {
        "S": "615f1e20-8261-4492-bb69-c281d3ea3382"
      }
    },
    "SequenceNumber": "36634400000000018872472773",
    "SizeBytes": 61,
    "StreamViewType": "KEYS_ONLY"
  },
  "eventSourceARN": "arn:aws:dynamodb:ap-south-1:57*****7078*****55:table/MessagesTable/stream/2022-05-15T17:48:27.902"
}

Консольный журнал во время выполнения 🍿

Наконец, после выполнения, мы можем найти вышеупомянутую полезную нагрузку JSON, которую мы получили в объекте event и которая затем используется для пакетного удаления из нашей таблицы staging. Вы можете найти результаты ниже в журналах облачных часов.

Здесь выполняется простой нагрузочный тест k6 в течение 15 секунд с 10 целями, и мы можем получить 27 запросов, из которых 25 успешны. 2 не прошли из-за ограничения скорости в нашем API шлюзе.

Журналы Cloudwatch с 25 записями ключей ⛲

Результат BatchWrite ⛲

Визуализация задержки 🚁

Здесь мы учитываем 25 успешных запросов puts, 25 запросов get и один пакетный запрос deletes от отдельных функций-обработчиков. Хотя каждый запрос имеет свой тип, общий фактор, связанный с использованием вызовов API dynamodb по сети, заключается в наличии других накладных расходов, таких как задержка.

Запросы PutItem

Запросы GetItem

Пакетные запросы DeleteItem

Следовательно, batchWrite или batchGet гораздо более эффективны, поскольку позволяют избежать задержки, и поэтому мы попробовали batchWrite для удаления ненужных данных.
Что касается нас, я узнал, что 25 — это максимальный предел для коллекции запросов пакетной записи.

Также необходимо проверятьUnprocessedItems в результате операции пакетной записи, и это лучшая практика проверять и повторять неудачные попытки один раз, когда у нас есть некоторые исключения из-за емкости или другие сбои из-за размера сообщения.

В следующей статье мы продемонстрируем, как мы будем использовать аналогичный подход для удаления объекта из S3, который мы ранее создали.

В следующих статьях мы будем добавлять больше соединений в наш стек и делать его более удобным в использовании, создавая новые конструкции, поэтому рассмотрите возможность следовать за мной и подписаться на мою рассылку.

⏭ У нас есть следующая статья по serverless, загляните на

https://dev.to/aravindvcyber/aws-cdk-101-dynamodb-streams-triggering-batch-deleteobjects-s3-2abn

🎉 Спасибо за поддержку! 🙏

Будет здорово, если вы захотите ☕ Купить мне кофе, чтобы поддержать мои усилия.

🔁 Оригинальный пост 🔗 Dev Post

🔁 Reposted at 🔗 dev to @aravindvcyber

🌺 AWS CDK 101 — 🚂 Dynamodb streams triggering batch deleteItem on another dynamodb table @hashnode

Смотрите больше на моей страницеhttps://t.co/CuYxnKr0Ig#TheHashnodeWriteathon#aws #awscdk #dynamodb #typescript #thwcloud-computing https://t.co/b3sOi0UmBu

— Aravind V (@Aravind_V7) 15 мая 2022 г.