🚁 AWS CDK 101 — 🐬 Получение JSON из dynamodb vs S3 с помощью stepfunction

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

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

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

🔁 Reposted previous post at 🔗 dev to @aravindvcyber

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

Визуализация этого сценария с помощью S3 ⛵

Визуализация этого сценария с помощью dynamodb 🚀.

Преимущества получения JSON из dynamodb по сравнению с S3 🏀

• Проще говоря, первое преимущество использования dynamodb вместо S3 — это сокращение времени, затрачиваемого на операции ввода-вывода.
• S3 занимает больше времени, чем аналогичная операция dynamodb.
• Хотя dynamodb стоит дороже, чем хранилище S3, при использовании хорошей архивной системы это можно использовать в наших интересах.
• Дальнейшая операция S3 — это внешний запрос, который имеет накладные расходы, такие как поиск DNS и создание httpAgent.
• В то время как операция с dynamodb рассматривается как транзакция базы данных и выполняется быстрее.
• Более того, вы можете перенести метаданные S3 в поля dynamodb для более удобного манипулирования.
• Это также поможет получить возможность запрашивать и извлекать только нужное поле на каждом этапе обработки, не загружая полное содержимое объекта, как в S3.
• Также S3 будет иметь лучшую пропускную способность, когда у нас будет больше префиксов, которые могут изменить путь ключа. В то время как в dynamodb у нас есть хороший рычаг влияния на пропускную способность за счет использования модели provisioned и pay per request.
• Мы также можем точно настроить емкость чтения и записи на основе запросов, которые могут быть дросселированы для эффективного управления затратами и надежности с полным контролем резервирования.
• Кстати, почему мы хотим сократить время ввода-вывода, это потому, что мы всегда запускаем его с вычислительного ресурса, который измеряется по мощности и времени выполнения, и который будет дольше простаивать.
• Данные — dynamodb, можем ли мы эффективно архивировать как S3 с помощью различных интеграций сервисов, кроме того, у dynamodb есть специальное свойство TTL, которое можно использовать для удаления временных объектов, как в нашем случае.

Окончательные результаты анализа 🏭

Я попытался проследить это с помощью трассировок x-ray и newrelic и обнаружил следующие наблюдения.

Я понимаю, что может быть некоторая разница, так как обе трассировки из x-ray и newrelic отслеживают тайминги по-разному.

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

Некоторые результаты трассировки из xray для putObject ⛲

Углубленная трассировка из newrelic для putItem 🎠

Некоторые результаты трассировки из xray для getObject ⛵

Углубленная трассировка из newrelic для getItem ⛺

Строительство 🚢

Начнем с создания новой таблицы, как показано ниже, в нашем основном стеке.

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

И предоставим доступ как обработчику входа, так и косвенно вызываемому обработчику бэкенд-функции шага.

stgMessages.grantWriteData(eventCounterBus.handler);
stgMessages.grantReadData(messageRecorder);

Рефакторинг обработчика ввода сообщения 🎪

Существующий обработчик ввода сообщения, ранее использовавшийся для сохранения в S3, теперь будет усовершенствован, как показано ниже.

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

const dbPut: any = async (msg: any) => {
  const dynamo = new DynamoDB();
  const crt_time: number = new Date(msg.Metadata?.timestamp).getTime();
  const putData: PutItemInput = {
    TableName: process.env.STAGING_MESSAGES_TABLE_NAME || "",
    Item: {
      messageId: { S: msg.Metadata?.UUID },
      createdAt: { N: `${crt_time}` },
      content: { S: JSON.stringify(msg) },
    },
    ReturnConsumedCapacity: "TOTAL",
  };
  await dynamo.putItem(putData).promise();
};

Добавление putItem один раз после putObject 🔱.

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

const save = async (uploadParams: PutObjectRequest) => {
  let putResult: PromiseResult<PutObjectOutput, AWSError> | undefined =
    undefined;
  let putDBResult: PromiseResult<PutObjectOutput, AWSError> | undefined =
    undefined;
  try {
    putResult = await s3.putObject(uploadParams).promise();
    putDBResult = await dbPut(uploadParams);
  } catch (e) {
    console.log(e);
  } finally {
    console.log(putResult);
    console.log(putDBResult);
  }
  return putResult;
};

Создание дополнительных свойств в JSON 🚥

На этом шаге мы добавим дополнительные данные в JSON, которые помогут нам получить информацию из dynamodb.

message.uuid = getUuid();
message.handler = context.awsRequestId;
message.key = `uploads/${message.uuid}.json`;
message.bucket = process.env.BucketName || "";
message.table = process.env.STAGING_MESSAGES_TABLE_NAME || "";
message.timestamp = new Date().toUTCString();

Изменения в полезной нагрузке шаговой функции 🚀

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

const sfnTaskPayload = sfn.TaskInput.fromObject({
   MyTaskToken: sfn.JsonPath.taskToken,
   Record: {
     "messageId.$": "$.detail.message.uuid",
     "createdAt.$": "$.detail.message.timestamp",
     "bucket.$": "$.detail.message.bucket",
     "key.$": "$.detail.message.key",
     "table.$": "$.detail.message.table",
     "uuid.$": "$.detail.message.uuid",
     "timestamp.$": "$.detail.message.timestamp"
   },
});

Рефакторинг обработчика регистратора сообщений 🎁

После получения данных из S3 из предыдущей статьи внутри функции message recorder, давайте теперь добавим новый шаг для получения данных непосредственно из dynamo, игнорируя getObject следующим образом.

После завершения вызова s3Get мы использовали dbGet, о чем будет сказано в трассировке, показанной выше.

try {
  const crt_time: number = new Date(msg.timestamp).getTime();
  const getData: GetItemInput = {
    TableName: msg.table,
    Key: {
      messageId: { S: msg.uuid },
      createdAt: { N: `${crt_time}` },
    },
    ProjectionExpression: "messageId, createdAt, content",
    ConsistentRead: true,
    ReturnConsumedCapacity: "TOTAL",
  };
  const dbGet = await dynamo.getItem(getData).promise();
  data = JSON.parse(dbGet.Item.content.S).Body;

  //OLD code with S3
  if (data) {
    msg.event = data;
    const token = JSON.parse(Record.body).MyTaskToken;
    await dbPut(msg)
      .then(async (data: any) => {
        await funcSuccess(data, msg.messageId, token);
      })
      .catch(async (err: any) => {
        await funcFailure(err, msg.messageId, token);
      });
  }
} catch (err) {
  console.log(err);
}

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

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

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

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

https://dev.to/aravindvcyber/aws-cdk-101-using-batched-dynamodb-stream-to-delete-item-on-another-dynamodb-table-4oag

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

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

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

🔁 Reposted at 🔗 dev to @aravindvcyber